feat: Vergleich der Buchungen mit dem Original-Auszug
Ein falscher Saldo sagt nur, DASS etwas fehlt. Das Skript rechnet je Tag zusammen, was im Auszug steht und was in der Datenbank gelandet ist, und listet die Tage auf, an denen beides auseinandergeht. Gebuehren werden wie im Auszug verrechnet (Betrag minus Gebuehr); gegen die Saldo-Spalte von Revolut geprueft. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Executable
+124
@@ -0,0 +1,124 @@
|
||||
#!/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.")
|
||||
Reference in New Issue
Block a user