certain/backend/scripts/raeume_doppelte_bankumsaetze.py

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

Zwei Fälle:

  Bank gegen Datei   Ein Abruf über einen längeren Zeitraum greift in Monate
                     zurück, die schon per CSV eingelesen sind.
  Bank gegen Bank    Dieselbe Bank ist zweimal verbunden (Enable Banking führt
                     sie je Land getrennt und vergibt eigene Kennungen). Jeder
                     Zugang meldet dieselbe Wirklichkeit – die Fassung eines
                     Zugangs bleibt stehen.

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 – und ueber WELCHES Bankkonto? Die
# Kennung steht vorn in external_id ("<uid>:<referenz>"). Sie unterscheidet
# zwei Zugaenge zu derselben Bank, die Enable Banking je Land getrennt fuehrt.
von_bank = {}
uid_von = {}
for r in con.execute("""
    SELECT id, transaction_id, external_id FROM inbox_items
     WHERE source = 'bank' AND transaction_id IS NOT NULL
"""):
    von_bank[r["transaction_id"]] = r["id"]
    uid_von[r["transaction_id"]] = (r["external_id"] or "").split(":")[0]

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

# --- Zweiter Durchgang: dieselbe Zahlung ueber ZWEI Bankzugaenge ----------
# Wer dieselbe Bank zweimal verbindet (etwa Revolut ueber DE und ueber LT),
# bekommt fuer dieselbe Zahlung zwei verschiedene Kennungen – der
# Duplikatschutz kann sie deshalb nicht als dieselbe erkennen. Jeder Zugang
# meldet dieselbe Wirklichkeit, also bleibt die Fassung EINES Zugangs stehen.
nach_gruppe: dict[tuple, list] = {}
for t in alle:
    if t["id"] not in von_bank:
        continue
    nach_gruppe.setdefault(
        (t["account_id"], t["type"], t["date"], t["amount_cents"]), []).append(t)

for gruppe in nach_gruppe.values():
    je_uid: dict[str, list] = {}
    for t in gruppe:
        je_uid.setdefault(uid_von.get(t["id"], ""), []).append(t)
    if len(je_uid) < 2:
        continue                      # nur ein Zugang – nichts zu entscheiden
    # Der Zugang mit den meisten Zeilen beschreibt die Wirklichkeit am
    # vollstaendigsten; bei Gleichstand der aeltere (kleinere Buchungs-ID).
    behalten = max(je_uid.values(), key=lambda zeilen: (len(zeilen), -zeilen[0]["id"]))
    for zeilen in je_uid.values():
        if zeilen is behalten:
            continue
        for t in zeilen:
            funde.append((t, behalten[0]))

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