certain/backend/scripts/raeume_doppelte_bankumsaetze.py

#!/usr/bin/env python3
"""Findet Bank-Umsätze, die eine schon vorhandene Buchung wiederholen.

Ein Abruf über einen längeren Zeitraum greift in Monate zurück, die schon per
CSV eingelesen sind. Der Duplikatschutz erkennt das nicht: Sein Schlüssel
enthält den Empfängertext, und der lautet bei der Bank anders als im Auszug.
Die Zahlung steht danach zweimal da.

Gesucht wird deshalb unscharf: gleiches Konto, gleiche Richtung, gleicher
Betrag, Datum ±3 Tage. Jede vorhandene Buchung kann dabei nur EINE
Bank-Buchung erklären – sonst verschluckt eine Kaffeezahlung alle anderen
desselben Betrags.

Weggeräumt wird immer die von der BANK, nie deine eigene: Deine trägt die
Kategorie, die du vergeben hast. Sie wird nicht gelöscht, sondern verworfen –
so holt sie kein späterer Abruf wieder herein.

Das Skript ändert ohne ``--wirklich`` nichts. Mit ``--wirklich`` legt es zuerst
eine Kopie der Datenbank an.

Aufruf:  raeume_doppelte_bankumsaetze.py [pfad-zur-db] [--wirklich]
"""
import sqlite3
import sys
from datetime import date, datetime, timedelta

TAGE = 3

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")}

# Welche Buchungen kamen von der Bank? Nur die kommen als Dublette in Frage.
von_bank = {
    r["transaction_id"]: r["id"]
    for r in con.execute("""
        SELECT id, transaction_id FROM inbox_items
         WHERE source = 'bank' AND transaction_id IS NOT NULL
    """)
}

alle = list(con.execute("""
    SELECT id, account_id, type, date, amount_cents, payee, description, category_id
      FROM transactions
     WHERE type != 'transfer'
     ORDER BY date, id
"""))

# Eigene Buchungen je (Konto, Richtung, Betrag) – daraus wird gleich geschöpft.
eigene: dict[tuple, list] = {}
for t in alle:
    if t["id"] in von_bank:
        continue
    eigene.setdefault((t["account_id"], t["type"], t["amount_cents"]), []).append(t)

verbraucht: set[int] = set()
funde = []
for t in alle:
    if t["id"] not in von_bank:
        continue
    schluessel = (t["account_id"], t["type"], t["amount_cents"])
    tag = date.fromisoformat(t["date"])
    passend = None
    for kandidat in eigene.get(schluessel, []):
        if kandidat["id"] in verbraucht:
            continue
        if abs((date.fromisoformat(kandidat["date"]) - tag).days) <= TAGE:
            passend = kandidat
            break
    if passend is not None:
        verbraucht.add(passend["id"])
        funde.append((t, passend))

if not funde:
    print("Nichts gefunden – kein Bank-Umsatz wiederholt eine vorhandene Buchung.")
    sys.exit(0)

print(f"{len(funde)} Bank-Umsätze wiederholen eine vorhandene Buchung.\n")
je_konto: dict[int, int] = {}
for bank, eigen in funde:
    konto = konten.get(bank["account_id"], {"name": "?", "currency": ""})
    zeichen = -1 if bank["type"] == "expense" else 1
    print(f"  {bank['date']}  {konto['name']:<22} {geld(bank['amount_cents']):>10}  "
          f"{(bank['payee'] or bank['description'] or '')[:38]}")
    print(f"     wiederholt #{eigen['id']} vom {eigen['date']}: "
          f"{(eigen['payee'] or eigen['description'] or '')[:38]}")
    je_konto[bank["account_id"]] = je_konto.get(bank["account_id"], 0) + zeichen * bank["amount_cents"]

print("\nDie Konten liegen dadurch daneben:")
for konto_id, cents in sorted(je_konto.items()):
    konto = konten.get(konto_id, {"name": "?", "currency": ""})
    richtung = "zu niedrig" if cents < 0 else "zu hoch"
    print(f"  {konto['name']:<24} {geld(abs(cents))} {konto['currency']} {richtung}")

if not WIRKLICH:
    print(f"\nNur angesehen, nichts geändert. Zum Ausführen:\n"
          f"  {sys.argv[0]} {DB} --wirklich")
    sys.exit(0)

kopie = f"{DB}.vor-aufraeumen-{datetime.now().strftime('%Y%m%d-%H%M%S')}"
with sqlite3.connect(kopie) as ziel:
    con.backup(ziel)
print(f"\nKopie der Datenbank: {kopie}")

for bank, _ in funde:
    # Verwerfen statt loeschen: Der Eintrag bleibt als Sperre stehen, damit
    # kein spaeterer Abruf denselben Umsatz noch einmal hereinholt.
    con.execute("""UPDATE inbox_items SET status = 'discarded', transaction_id = NULL
                    WHERE id = ?""", (von_bank[bank["id"]],))
    con.execute("DELETE FROM transactions WHERE id = ?", (bank["id"],))
con.commit()
print(f"{len(funde)} doppelte Bank-Buchungen entfernt.\n"
      "Sieh dir jetzt die Kontenseite an – dort muss bei den angebundenen\n"
      "Konten 'Saldo stimmt mit der Bank überein' stehen.")