From 6745f5490f2c9b94ffbb819c5d2fb01bd6ae5476 Mon Sep 17 00:00:00 2001 From: Guido Date: Fri, 21 Aug 2026 15:34:35 +0200 Subject: [PATCH] feat: Skript raeumt falsche Uebertraege im Altbestand auf Zwei Fehlerbilder: Uebertraege ohne Beleg (ein zu kurzer Erkennungs-Text schlug mitten im Wort an) werden wieder zu Ausgabe/Einnahme, und zwar auf dem Konto, aus dessen Auszug die Zeile stammt. Spiegel-Paare (dieselbe Bewegung steht in beiden Auszuegen, etwa eine Revolut-Aufrundung) werden gemeldet; welche Richtung bleibt, waehlt --behalte. Ohne --anwenden aendert das Skript nichts, und vor dem Schreiben legt es eine Kopie der Datenbank an. Co-Authored-By: Claude Opus 5 --- backend/scripts/raeume_uebertraege.py | 179 ++++++++++++++++++++++++++ 1 file changed, 179 insertions(+) create mode 100755 backend/scripts/raeume_uebertraege.py diff --git a/backend/scripts/raeume_uebertraege.py b/backend/scripts/raeume_uebertraege.py new file mode 100755 index 0000000..d9f2c44 --- /dev/null +++ b/backend/scripts/raeume_uebertraege.py @@ -0,0 +1,179 @@ +#!/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")