certain/backend/scripts/pruefe_gegen_auszug.py

#!/usr/bin/env python3
"""Vergleicht die gebuchten Daten Tag für Tag mit dem Original-Auszug.

Stimmt ein Kontostand nicht, sagt das nur DASS etwas fehlt. Dieses Skript sagt
WO: Es rechnet je Tag zusammen, was im Auszug steht und was in der Datenbank
gelandet ist, und listet die Tage auf, an denen beides auseinandergeht.

Aufruf (Spalten heissen wie in der Kopfzeile der Datei):
  pruefe_gegen_auszug.py --konto 2 --datum "Datum des Abschlusses" --betrag Betrag \
      --gebuehr "Gebühr" --status Status --filter "Produkt=Giro" auszug.csv [db]
"""
import argparse
import csv
import sqlite3
from collections import defaultdict
from datetime import date

NICHT_GEBUCHT = {"PENDING", "AUSSTEHEND", "REVERTED", "STORNIERT", "DECLINED",
                 "ABGELEHNT", "FAILED", "FEHLGESCHLAGEN", "CANCELLED"}

p = argparse.ArgumentParser()
p.add_argument("datei")
p.add_argument("db", nargs="?", default="/var/lib/certain-finance/app.db")
p.add_argument("--konto", type=int, required=True)
p.add_argument("--datum", required=True)
p.add_argument("--betrag", required=True)
p.add_argument("--gebuehr")
p.add_argument("--status")
p.add_argument("--filter", dest="filt")
p.add_argument("--trenner", default=",")
a = p.parse_args()


def zahl(s) -> int:
    """Betragstext -> Cent. Verträgt '1'234.56', '1.234,56' und leere Felder."""
    s = str(s or "").strip().replace("'", "").replace(" ", "")
    if not s:
        return 0
    if "," in s and "." in s:
        s = s.replace(".", "").replace(",", ".") if s.rindex(",") > s.rindex(".") else s.replace(",", "")
    elif "," in s:
        s = s.replace(",", ".")
    try:
        return round(float(s) * 100)
    except ValueError:
        return 0


def tag(s) -> date | None:
    s = str(s or "").strip()[:10]
    for f in ("%Y-%m-%d", "%d.%m.%Y", "%d/%m/%Y", "%m/%d/%Y"):
        try:
            return date.strptime(s, f) if False else __import__("datetime").datetime.strptime(s, f).date()
        except ValueError:
            continue
    return None


# --- Auszug einlesen ---------------------------------------------------------
fs, fv = (a.filt.split("=", 1) if a.filt else (None, None))
je_tag_datei: dict[date, int] = defaultdict(int)
zeilen = 0
with open(a.datei, encoding="utf-8-sig", newline="") as f:
    for r in csv.DictReader(f, delimiter=a.trenner):
        if fs and (r.get(fs) or "").strip() != fv:
            continue
        if a.status and (r.get(a.status) or "").strip().upper() in NICHT_GEBUCHT:
            continue
        d = tag(r.get(a.datum))
        if d is None:
            continue
        betrag = zahl(r.get(a.betrag)) - zahl(r.get(a.gebuehr)) if a.gebuehr else zahl(r.get(a.betrag))
        je_tag_datei[d] += betrag
        zeilen += 1

# --- Datenbank ---------------------------------------------------------------
con = sqlite3.connect(f"file:{a.db}?mode=ro", uri=True)
je_tag_db: dict[date, int] = defaultdict(int)
for d, typ, betrag, ziel, zielbetrag in con.execute(
        "SELECT date, type, amount_cents, to_account_id, to_amount_cents FROM transactions "
        "WHERE account_id = ? OR to_account_id = ?", (a.konto, a.konto)):
    dd = tag(d)
    if dd is None:
        continue
    if typ == "income":
        je_tag_db[dd] += betrag
    elif typ == "expense":
        je_tag_db[dd] -= betrag
    elif ziel == a.konto:
        je_tag_db[dd] += zielbetrag if zielbetrag is not None else betrag
    else:
        je_tag_db[dd] -= betrag


def geld(c):
    return f"{c / 100:,.2f}".replace(",", "'")


print(f"Auszug: {a.datei}  ({zeilen} gebuchte Zeilen)")
print(f"Konto {a.konto} in {a.db}\n")
print(f"Summe Auszug:     {geld(sum(je_tag_datei.values())):>12}")
print(f"Summe Datenbank:  {geld(sum(je_tag_db.values())):>12}")
print(f"Unterschied:      {geld(sum(je_tag_db.values()) - sum(je_tag_datei.values())):>12}\n")

print("=== Monate mit Abweichung ===")
mon_d, mon_b = defaultdict(int), defaultdict(int)
for d, v in je_tag_datei.items():
    mon_d[d.strftime("%Y-%m")] += v
for d, v in je_tag_db.items():
    mon_b[d.strftime("%Y-%m")] += v
for m in sorted(set(mon_d) | set(mon_b)):
    if mon_d[m] != mon_b[m]:
        print(f"  {m}   Auszug {geld(mon_d[m]):>11}   DB {geld(mon_b[m]):>11}   "
              f"Differenz {geld(mon_b[m] - mon_d[m]):>10}")

print("\n=== Tage mit Abweichung (die 25 grössten) ===")
tage = [(d, je_tag_db[d] - je_tag_datei[d]) for d in set(je_tag_datei) | set(je_tag_db)
        if je_tag_db[d] != je_tag_datei[d]]
tage.sort(key=lambda t: -abs(t[1]))
for d, diff in tage[:25]:
    print(f"  {d}   Auszug {geld(je_tag_datei[d]):>10}   DB {geld(je_tag_db[d]):>10}   "
          f"Differenz {geld(diff):>10}")
if not tage:
    print("  keine – der Auszug ist vollständig und korrekt gebucht.")