#!/usr/bin/env python3 """Räumt Überträge auf, deren Gegenseite zusätzlich als Buchung dasteht. Welche der beiden Zeilen falsch ist, hängt daran, ob Certain das Gegenkonto selbst abholt – und das steht in der Datenbank: Beide Konten angebunden Die Bank liefert beide Seiten. Der ÜBERTRAG ist falsch; er wird in die Buchung zurückverwandelt, aus der er entstand. Beide Seiten stehen danach wieder da und lassen sich verbinden. Gegenkonto nicht abgerufen Der Übertrag ist richtig. Die zusätzliche Buchung ist überzählig und wird gelöscht. Ohne ``--wirklich`` wird nur gezeigt, was passieren würde. Mit ``--wirklich`` legt das Skript ZUERST eine Kopie der Datenbank an und sagt, wo sie liegt. Aufruf: repariere_doppelte_uebertraege.py [pfad-zur-db] [--wirklich] """ import sqlite3 import sys from datetime import date, datetime, timedelta TAGE = 3 FX_SPIELRAUM = 0.025 argumente = [a for a in sys.argv[1:] if not a.startswith("--")] WIRKLICH = "--wirklich" in sys.argv DB = argumente[0] if argumente else "/var/lib/certain-finance/app.db" con = sqlite3.connect(DB) con.row_factory = sqlite3.Row def geld(cents): return f"{cents / 100:,.2f}".replace(",", "'") konten = {r["id"]: r for r in con.execute("SELECT id, name, currency FROM accounts")} try: angebunden = {r[0] for r in con.execute( "SELECT account_id FROM bank_account_links WHERE account_id IS NOT NULL")} except sqlite3.OperationalError: angebunden = set() plan, offen = [], [] # Je Uebertrag hoechstens EIN Eingriff. Findet sich auf beiden Seiten eine # Dublette, waeren "zurueckverwandeln" und "Buchung loeschen" widerspruechlich – # dann lieber melden als raten. behandelt = set() for u in con.execute(""" SELECT id, date, account_id, to_account_id, amount_cents, to_amount_cents FROM transactions WHERE type = 'transfer' ORDER BY date, id """): tag = date.fromisoformat(u["date"]) gleiche_waehrung = (konten[u["account_id"]]["currency"] == konten[u["to_account_id"]]["currency"] if u["account_id"] in konten and u["to_account_id"] in konten else True) for konto_id, betrag, richtung in ( (u["account_id"], u["amount_cents"], "expense"), (u["to_account_id"], u["to_amount_cents"] or u["amount_cents"], "income"), ): if konto_id is None or not betrag: continue spielraum = 0 if gleiche_waehrung else max(1, round(betrag * FX_SPIELRAUM)) for k in con.execute(""" SELECT id, date, amount_cents, payee FROM transactions WHERE account_id = ? AND type = ? AND date BETWEEN ? AND ? AND id != ? """, (konto_id, richtung, (tag - timedelta(days=TAGE)).isoformat(), (tag + timedelta(days=TAGE)).isoformat(), u["id"])): if abs(k["amount_cents"] - betrag) > spielraum: continue if u["id"] in behandelt: offen.append(f"Übertrag #{u['id']}: passende Buchungen auf BEIDEN " f"Seiten – bitte selbst ansehen") continue behandelt.add(u["id"]) a, b = konten.get(u["account_id"]), konten.get(u["to_account_id"]) kopf = (f"Übertrag #{u['id']} {u['date']} {a['name']} → {b['name']} " f"{geld(u['amount_cents'])}\n" f" daneben: Buchung #{k['id']} auf '{konten[konto_id]['name']}' " f"{geld(k['amount_cents'])}") if u["account_id"] in angebunden and u["to_account_id"] in angebunden: # Der Uebertrag entstand aus einer echten Buchung. Welche das # war, sagt der Eingangs-Eintrag, der auf ihn zeigt. quelle = con.execute(""" SELECT id, account_id, amount_cents FROM inbox_items WHERE transaction_id = ? AND source != 'transfer' LIMIT 1 """, (u["id"],)).fetchone() if quelle is None: offen.append(kopf + "\n → von Hand angelegt, ich kann nicht " "wissen, wie er vorher aussah") continue if konten[quelle["account_id"]]["currency"] != a["currency"]: offen.append(kopf + "\n → über zwei Währungen, der " "Ursprungsbetrag ist nicht mehr eindeutig") continue plan.append(("zurueck", u, k, quelle, kopf)) elif (u["to_account_id"] if konto_id == u["account_id"] else u["account_id"]) not in angebunden: plan.append(("loeschen", u, k, None, kopf)) else: offen.append(kopf + "\n → nur ein Konto angebunden, bitte selbst ansehen") if not plan and not offen: print("Nichts zu tun – kein Übertrag hat eine zusätzlich gebuchte Gegenseite.") sys.exit(0) for art, u, k, quelle, kopf in plan: print(kopf) if art == "zurueck": typ = "Ausgabe" if quelle["amount_cents"] < 0 else "Einnahme" print(f" → beide Konten angebunden: Übertrag #{u['id']} wird wieder zur " f"{typ} auf '{konten[quelle['account_id']]['name']}'\n") else: print(f" → Gegenkonto wird nicht abgerufen: Buchung #{k['id']} wird " f"gelöscht\n") for text in offen: print(text + "\n") if not WIRKLICH: print(f"Nur angesehen, nichts geändert. Zum Ausführen:\n" f" {sys.argv[0]} {DB} --wirklich") sys.exit(0) kopie = f"{DB}.vor-reparatur-{datetime.now().strftime('%Y%m%d-%H%M%S')}" with sqlite3.connect(kopie) as ziel: con.backup(ziel) # nimmt auch das WAL mit, anders als ein Dateikopie print(f"Kopie der Datenbank: {kopie}\n") for art, u, k, quelle, _ in plan: if art == "zurueck": typ = "expense" if quelle["amount_cents"] < 0 else "income" con.execute(""" UPDATE transactions SET type = ?, account_id = ?, amount_cents = ?, to_account_id = NULL, to_amount_cents = NULL, fx_rate = NULL WHERE id = ? """, (typ, quelle["account_id"], abs(quelle["amount_cents"]), u["id"])) # Die Gegenseite, die Certain fuer den Uebertrag angelegt hatte, ist # damit gegenstandslos. con.execute("DELETE FROM inbox_items WHERE transaction_id = ? AND source = 'transfer'", (u["id"],)) print(f"Übertrag #{u['id']} zurückverwandelt.") else: # Eingangs-Eintrag verwerfen statt loeschen: Sonst brächte ihn der # nächste Abruf oder Auszug wieder herein. con.execute("""UPDATE inbox_items SET status = 'discarded', transaction_id = NULL WHERE transaction_id = ?""", (k["id"],)) con.execute("DELETE FROM transactions WHERE id = ?", (k["id"],)) print(f"Buchung #{k['id']} gelöscht.") con.commit() print("\nFertig. Sieh dir jetzt die Kontenseite an – dort muss bei den " "angebundenen Konten\n'Saldo stimmt mit der Bank überein' stehen.")