certain/backend/scripts/repariere_doppelte_uebertraege.py

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

Welche der beiden Zeilen falsch ist, hängt daran, ob Certain das Gegenkonto
selbst abholt – und das steht in der Datenbank:

  Beide Konten angebunden   Die Bank liefert beide Seiten. Der ÜBERTRAG ist
                            falsch; er wird in die Buchung zurückverwandelt,
                            aus der er entstand. Beide Seiten stehen danach
                            wieder da und lassen sich verbinden.

  Gegenkonto nicht abgerufen  Der Übertrag ist richtig. Die zusätzliche
                            Buchung ist überzählig und wird gelöscht.

Ohne ``--wirklich`` wird nur gezeigt, was passieren würde. Mit ``--wirklich``
legt das Skript ZUERST eine Kopie der Datenbank an und sagt, wo sie liegt.

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

TAGE = 3
FX_SPIELRAUM = 0.025

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

plan, offen = [], []
# Je Uebertrag hoechstens EIN Eingriff. Findet sich auf beiden Seiten eine
# Dublette, waeren "zurueckverwandeln" und "Buchung loeschen" widerspruechlich –
# dann lieber melden als raten.
behandelt = set()

for u in con.execute("""
    SELECT id, date, account_id, to_account_id, amount_cents, to_amount_cents
      FROM transactions WHERE type = 'transfer' ORDER BY date, id
"""):
    tag = date.fromisoformat(u["date"])
    gleiche_waehrung = (konten[u["account_id"]]["currency"]
                        == konten[u["to_account_id"]]["currency"]
                        if u["account_id"] in konten and u["to_account_id"] in konten
                        else True)

    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
        spielraum = 0 if gleiche_waehrung else max(1, round(betrag * FX_SPIELRAUM))
        for k in con.execute("""
            SELECT id, date, amount_cents, payee FROM transactions
             WHERE account_id = ? AND type = ? AND date BETWEEN ? AND ? AND id != ?
        """, (konto_id, richtung, (tag - timedelta(days=TAGE)).isoformat(),
              (tag + timedelta(days=TAGE)).isoformat(), u["id"])):
            if abs(k["amount_cents"] - betrag) > spielraum:
                continue

            if u["id"] in behandelt:
                offen.append(f"Übertrag #{u['id']}: passende Buchungen auf BEIDEN "
                             f"Seiten – bitte selbst ansehen")
                continue
            behandelt.add(u["id"])

            a, b = konten.get(u["account_id"]), konten.get(u["to_account_id"])
            kopf = (f"Übertrag #{u['id']}  {u['date']}  {a['name']} → {b['name']}  "
                    f"{geld(u['amount_cents'])}\n"
                    f"  daneben: Buchung #{k['id']} auf '{konten[konto_id]['name']}'  "
                    f"{geld(k['amount_cents'])}")

            if u["account_id"] in angebunden and u["to_account_id"] in angebunden:
                # Der Uebertrag entstand aus einer echten Buchung. Welche das
                # war, sagt der Eingangs-Eintrag, der auf ihn zeigt.
                quelle = con.execute("""
                    SELECT id, account_id, amount_cents FROM inbox_items
                     WHERE transaction_id = ? AND source != 'transfer' LIMIT 1
                """, (u["id"],)).fetchone()
                if quelle is None:
                    offen.append(kopf + "\n  → von Hand angelegt, ich kann nicht "
                                        "wissen, wie er vorher aussah")
                    continue
                if konten[quelle["account_id"]]["currency"] != a["currency"]:
                    offen.append(kopf + "\n  → über zwei Währungen, der "
                                        "Ursprungsbetrag ist nicht mehr eindeutig")
                    continue
                plan.append(("zurueck", u, k, quelle, kopf))
            elif (u["to_account_id"] if konto_id == u["account_id"]
                  else u["account_id"]) not in angebunden:
                plan.append(("loeschen", u, k, None, kopf))
            else:
                offen.append(kopf + "\n  → nur ein Konto angebunden, bitte selbst ansehen")

if not plan and not offen:
    print("Nichts zu tun – kein Übertrag hat eine zusätzlich gebuchte Gegenseite.")
    sys.exit(0)

for art, u, k, quelle, kopf in plan:
    print(kopf)
    if art == "zurueck":
        typ = "Ausgabe" if quelle["amount_cents"] < 0 else "Einnahme"
        print(f"  → beide Konten angebunden: Übertrag #{u['id']} wird wieder zur "
              f"{typ} auf '{konten[quelle['account_id']]['name']}'\n")
    else:
        print(f"  → Gegenkonto wird nicht abgerufen: Buchung #{k['id']} wird "
              f"gelöscht\n")
for text in offen:
    print(text + "\n")

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

kopie = f"{DB}.vor-reparatur-{datetime.now().strftime('%Y%m%d-%H%M%S')}"
with sqlite3.connect(kopie) as ziel:
    con.backup(ziel)          # nimmt auch das WAL mit, anders als ein Dateikopie
print(f"Kopie der Datenbank: {kopie}\n")

for art, u, k, quelle, _ in plan:
    if art == "zurueck":
        typ = "expense" if quelle["amount_cents"] < 0 else "income"
        con.execute("""
            UPDATE transactions
               SET type = ?, account_id = ?, amount_cents = ?,
                   to_account_id = NULL, to_amount_cents = NULL, fx_rate = NULL
             WHERE id = ?
        """, (typ, quelle["account_id"], abs(quelle["amount_cents"]), u["id"]))
        # Die Gegenseite, die Certain fuer den Uebertrag angelegt hatte, ist
        # damit gegenstandslos.
        con.execute("DELETE FROM inbox_items WHERE transaction_id = ? AND source = 'transfer'",
                    (u["id"],))
        print(f"Übertrag #{u['id']} zurückverwandelt.")
    else:
        # Eingangs-Eintrag verwerfen statt loeschen: Sonst brächte ihn der
        # nächste Abruf oder Auszug wieder herein.
        con.execute("""UPDATE inbox_items SET status = 'discarded', transaction_id = NULL
                        WHERE transaction_id = ?""", (k["id"],))
        con.execute("DELETE FROM transactions WHERE id = ?", (k["id"],))
        print(f"Buchung #{k['id']} gelöscht.")
con.commit()
print("\nFertig. Sieh dir jetzt die Kontenseite an – dort muss bei den "
      "angebundenen Konten\n'Saldo stimmt mit der Bank überein' stehen.")