certain/backend/scripts/pruefe_doppelte_uebertraege.py

#!/usr/bin/env python3
"""Findet Überträge, deren Gegenseite zusätzlich als eigene Buchung dasteht.

Ein Kontoübertrag bewegt Geld in EINER Buchung. Liefert die Bank dieselbe
Bewegung zusätzlich als Gutschrift oder Belastung, steht das Geld auf dem
betroffenen Konto zweimal. Genau das ist passiert, solange Certain beim
Anlegen eines Übertrags nicht nachgesehen hat, ob die Gegenseite schon
gebucht ist.

Das Skript ÄNDERT NICHTS. Es öffnet die Datenbank nur lesend und sagt, was
zu prüfen ist – aufräumen tust du von Hand, damit nichts Richtiges mit
weggeräumt wird.

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

# So viele Tage darf die Gegenbuchung abweichen. Banken buchen eine Bewegung
# oft ein bis zwei Tage nach dem Auslösen – dieselbe Spanne wie beim Import.
TAGE = 3
# Über Währungen hinweg ist der erwartete Betrag gerechnet, der echte kommt
# von der Bank. Ohne Spielraum fände man diese Fälle nie.
FX_SPIELRAUM = 0.025

DB = sys.argv[1] if len(sys.argv) > 1 else "/var/lib/certain-finance/app.db"
con = sqlite3.connect(f"file:{DB}?mode=ro", uri=True)
con.row_factory = sqlite3.Row


def geld(cents, waehrung=""):
    return f"{cents / 100:,.2f}".replace(",", "'") + (f" {waehrung}" if waehrung else "")


konten = {r["id"]: r for r in con.execute("SELECT id, name, currency FROM accounts")}

# Welche Konten holt Certain selbst von der Bank? Daran haengt, welche der
# beiden Zeilen die falsche ist – und das muss niemand raten.
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()          # aeltere Datenbank ohne Bankanbindung

verdacht = []

for u in con.execute("""
    SELECT id, date, account_id, to_account_id, amount_cents, to_amount_cents,
           payee, description
      FROM transactions
     WHERE type = 'transfer'
     ORDER BY date, id
"""):
    tag = date.fromisoformat(u["date"])
    von, bis = tag - timedelta(days=TAGE), tag + timedelta(days=TAGE)

    # Beide Seiten prüfen: Die überzählige Buchung kann auf dem Abgangs- wie
    # auf dem Zugangskonto liegen, je nachdem, welche Seite umgewidmet wurde.
    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
        gleiche_waehrung = (konten.get(u["account_id"], {})["currency"]
                            == konten.get(u["to_account_id"], {})["currency"]
                            if u["to_account_id"] in konten and u["account_id"] in konten
                            else True)
        spielraum = 0 if gleiche_waehrung else max(1, round(betrag * FX_SPIELRAUM))
        for k in con.execute("""
            SELECT id, date, amount_cents, payee, description, category_id
              FROM transactions
             WHERE account_id = ? AND type = ? AND date BETWEEN ? AND ? AND id != ?
        """, (konto_id, richtung, von.isoformat(), bis.isoformat(), u["id"])):
            if abs(k["amount_cents"] - betrag) <= spielraum:
                verdacht.append((u, konto_id, betrag, k, richtung))

if not verdacht:
    print("Nichts gefunden – kein Übertrag hat eine zusätzlich gebuchte Gegenseite.")
    sys.exit(0)

print(f"{len(verdacht)} Stelle(n) zu prüfen.\n")
abweichung = {}
for u, konto_id, betrag, k, richtung in verdacht:
    konto = konten.get(konto_id, {"name": "?", "currency": ""})
    a = konten.get(u["account_id"], {"name": "?"})
    b = konten.get(u["to_account_id"], {"name": "?"})
    gegen_id = u["to_account_id"] if konto_id == u["account_id"] else u["account_id"]

    print(f"Übertrag #{u['id']}  {u['date']}  {a['name']} → {b['name']}  "
          f"{geld(u['amount_cents'])}")
    print(f"  {u['payee'] or u['description'] or 'ohne Text'}")
    print(f"  daneben auf '{konto['name']}': Buchung #{k['id']}  {k['date']}  "
          f"{geld(k['amount_cents'], konto['currency'])}")
    print(f"  {k['payee'] or k['description'] or 'ohne Text'}")

    # Das Vorzeichen entscheidet, in welche Richtung das Konto danebenliegt.
    vorzeichen = 1 if richtung == "income" else -1
    zu = "zu hoch" if vorzeichen > 0 else "zu niedrig"
    print(f"  → '{konto['name']}' ist um {geld(k['amount_cents'])} {zu}")

    if u["account_id"] in angebunden and u["to_account_id"] in angebunden:
        print(f"  → BEIDE Konten sind angebunden: Die Bank liefert beide Seiten "
              f"selbst.\n     Falsch ist der ÜBERTRAG #{u['id']} – löschen. Danach "
              f"'{a['name']}' neu abrufen,\n     dann verbindet Certain die beiden.")
    elif gegen_id not in angebunden:
        gegen = konten.get(gegen_id, {"name": "?"})
        print(f"  → '{gegen['name']}' wird NICHT abgerufen: Der Übertrag ist richtig.\n"
              f"     Überzählig ist Buchung #{k['id']} – die kann weg.")
    else:
        print(f"  → Nur ein Konto ist angebunden. Sieh dir das Paar von Hand an.")
    print()
    abweichung[konto_id] = abweichung.get(konto_id, 0) + vorzeichen * k["amount_cents"]

print("Wenn sich der Verdacht bestätigt, liegen die Konten so daneben:")
for konto_id, cents in sorted(abweichung.items()):
    konto = konten.get(konto_id, {"name": "?", "currency": ""})
    richtung = "zu hoch" if cents > 0 else "zu niedrig"
    print(f"  {konto['name']:<28} {geld(abs(cents), konto['currency'])} {richtung}")

print("""
Lösch je Paar nur EINE der beiden Zeilen – welche, steht oben.

Im Zweifel: Kontenseite ansehen. Steht dort "Saldo stimmt mit der Bank
überein", ist das Konto in Ordnung, egal was hier steht.""")