#!/usr/bin/env python3 """Macht Überträge rückgängig, die nie welche waren. Zwei Fehlerbilder werden gesucht: 1. OHNE BELEG – ein zu kurzer Erkennungs-Text hat mitten in einem Wort angeschlagen („Euro" in „PayPal Europe"). Solche Zeilen sind gewöhnliche Ausgaben/Einnahmen; als Übertrag schreiben sie dem Zielkonto Geld gut, das nie geflossen ist. 2. SPIEGEL-PAARE – dieselbe Bewegung steht in beiden Auszügen (Revolut führt eine Aufrundung im Giro- UND im Pocket-Konto). Wurden beide Zeilen gebucht, existiert der Übertrag doppelt, einmal je Richtung. Ohne ``--anwenden`` wird nur berichtet, nichts geändert. Aufruf: raeume_uebertraege.py [db] nur berichten raeume_uebertraege.py --anwenden [db] Fall 1 bereinigen raeume_uebertraege.py --anwenden --behalte "2>15" [db] zusätzlich Spiegel-Paare: Richtung Konto 2 -> Konto 15 behalten """ import datetime import re import shutil import sqlite3 import sys argv = [a for a in sys.argv[1:]] ANWENDEN = "--anwenden" in argv BEHALTE: set[tuple[int, int]] = set() while "--behalte" in argv: i = argv.index("--behalte") a, b = argv[i + 1].split(">") BEHALTE.add((int(a), int(b))) del argv[i:i + 2] rest = [a for a in argv if not a.startswith("--")] DB = rest[0] if rest else "/var/lib/certain-finance/app.db" con = sqlite3.connect(DB) con.row_factory = sqlite3.Row konten = {r["id"]: r for r in con.execute("SELECT id, name, currency FROM accounts")} def name(i): return konten[i]["name"] if i in konten else f"?{i}" def tag(s: str) -> datetime.date: return datetime.date.fromisoformat(s[:10]) def geld(c): return f"{(c or 0) / 100:,.2f}".replace(",", "'") def als_wort(nadel: str, heu: str) -> bool: if not nadel: return False return re.search(rf"(? bool: """Spricht ausser dem alten Teilstring-Treffer irgendetwas für einen Übertrag?""" text = f"{(t['payee'] or '').lower()} {(t['description'] or '').lower()}" for konto in (t["account_id"], t["to_account_id"]): for val in idents.get(konto, []): if als_wort(val, text): return True for muster, ziel in regeln: if muster and muster in text and (ziel is None or ziel in (t["account_id"], t["to_account_id"])): return True return False transfers = con.execute( "SELECT t.id, t.date, t.account_id, t.to_account_id, t.amount_cents, t.to_amount_cents, " "t.payee, t.description, t.import_batch_id, b.account_id AS batch_account " "FROM transactions t LEFT JOIN import_batches b ON b.id = t.import_batch_id " "WHERE t.type='transfer' AND t.to_account_id IS NOT NULL").fetchall() print(f"Datenbank: {DB}\n{len(transfers)} Überträge geprüft\n") # --- Fall 1: ohne Beleg ----------------------------------------------------- ohne = [t for t in transfers if not hat_beleg(t)] print(f"=== 1. Überträge ohne Beleg: {len(ohne)} ===") if ohne: je_paar: dict[tuple[int, int], list] = {} for t in ohne: je_paar.setdefault((t["account_id"], t["to_account_id"]), []).append(t) for (a, b), liste in sorted(je_paar.items(), key=lambda kv: -len(kv[1])): summe = sum(t["amount_cents"] for t in liste) print(f" {name(a)[:22]:<24}-> {name(b)[:22]:<24}{len(liste):>4}x {geld(summe):>10}") for t in liste[:2]: print(f" {t['date']} {((t['payee'] or '') + ' ' + (t['description'] or '')).strip()[:60]}") else: print(" keine") # --- Fall 2: Spiegel-Paare --------------------------------------------------- nach_id = {t["id"]: t for t in transfers} frei = set(nach_id) paare: dict[tuple[int, int], list[tuple[int, int]]] = {} nach_paar: dict[tuple[int, int], list] = {} for t in transfers: nach_paar.setdefault((t["account_id"], t["to_account_id"]), []).append(t) for (a, b), liste in nach_paar.items(): gegen = nach_paar.get((b, a), []) if not gegen or a > b: continue offen = [g for g in gegen] for t in liste: if t["id"] not in frei: continue for g in offen: if g["id"] not in frei: continue if abs(t["amount_cents"]) != abs(g["amount_cents"]): continue if abs((tag(t["date"]) - tag(g["date"])).days) > 2: continue frei.discard(t["id"]); frei.discard(g["id"]) paare.setdefault((a, b), []).append((t["id"], g["id"])) break print(f"\n=== 2. Spiegel-Paare (dieselbe Bewegung aus beiden Auszügen) ===") if not paare: print(" keine") for (a, b), liste in paare.items(): summe = sum(abs(nach_id[x]["amount_cents"]) for x, _ in liste) print(f" {name(a)} <-> {name(b)}: {len(liste)} Paare, {geld(summe)}") print(f" Nur EINE Richtung ist richtig. Wähle sie mit:") print(f" --behalte \"{a}>{b}\" (behält {name(a)} -> {name(b)})") print(f" --behalte \"{b}>{a}\" (behält {name(b)} -> {name(a)})") # --- Anwenden --------------------------------------------------------------- if not ANWENDEN: print("\nNichts geändert (Probelauf). Mit --anwenden ausführen.") raise SystemExit(0) sicherung = DB + ".vor-aufraeumen" shutil.copy(DB, sicherung) print(f"\nSicherung: {sicherung}") zurueck = 0 for t in ohne: # Aus welchem Auszug stammt die Zeile? Diese Seite behält die Buchung. quelle = t["batch_account"] if quelle == t["account_id"] or quelle is None: con.execute("UPDATE transactions SET type='expense', to_account_id=NULL, " "to_amount_cents=NULL WHERE id=?", (t["id"],)) else: betrag = t["to_amount_cents"] if t["to_amount_cents"] is not None else t["amount_cents"] con.execute("UPDATE transactions SET type='income', account_id=?, amount_cents=?, " "to_account_id=NULL, to_amount_cents=NULL WHERE id=?", (quelle, betrag, t["id"])) zurueck += 1 print(f"{zurueck} Überträge in Ausgabe/Einnahme zurückverwandelt") geloescht = 0 for (a, b), liste in paare.items(): if (a, b) in BEHALTE: weg = [y for _, y in liste] elif (b, a) in BEHALTE: weg = [x for x, _ in liste] else: print(f" {name(a)} <-> {name(b)}: übersprungen (keine Richtung gewählt)") continue con.executemany("DELETE FROM transactions WHERE id=?", [(i,) for i in weg]) geloescht += len(weg) print(f"{geloescht} doppelte Überträge gelöscht") con.commit() print("Fertig. Dienst neu starten: sudo systemctl restart certain-finance")