feat: Skript teilt bestehende Uebertraege in zwei Buchungen
Damit der Altbestand zum neuen Verhalten passt: Jeder Uebertrag wird zu einer Ausgabe beim Quellkonto und einer Einnahme beim Zielkonto, beide mit der neutralen Kategorie. Die Salden bleiben dabei exakt gleich, nur die Art aendert sich. Ohne --anwenden nur ein Bericht; vor dem Schreiben wird die Datenbank kopiert. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Executable
+110
@@ -0,0 +1,110 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Teilt bestehende Überträge in zwei gewöhnliche Buchungen auf.
|
||||
|
||||
Certain verbindet Konten nicht mehr selbst. Damit die Salden trotzdem exakt so
|
||||
bleiben wie bisher, wird jeder Übertrag zu genau dem, was in den Auszügen steht:
|
||||
einer Ausgabe beim Quellkonto und einer Einnahme beim Zielkonto. Beide bekommen
|
||||
die neutrale Kategorie, zählen also nicht in die Statistik.
|
||||
|
||||
Ohne ``--anwenden`` wird nur berichtet. Vor dem Schreiben legt das Skript eine
|
||||
Kopie der Datenbank an.
|
||||
|
||||
Aufruf: teile_uebertraege.py [--anwenden] [db]
|
||||
"""
|
||||
import shutil
|
||||
import sqlite3
|
||||
import sys
|
||||
|
||||
argv = sys.argv[1:]
|
||||
ANWENDEN = "--anwenden" in argv
|
||||
rest = [x for x in argv if not x.startswith("--")]
|
||||
DB = rest[0] if rest else "/var/lib/certain-finance/app.db"
|
||||
|
||||
con = sqlite3.connect(DB)
|
||||
con.row_factory = sqlite3.Row
|
||||
konten = {r["id"]: r for r in con.execute("SELECT id, name, currency FROM accounts")}
|
||||
|
||||
|
||||
def geld(c):
|
||||
return f"{(c or 0) / 100:,.2f}".replace(",", "'")
|
||||
|
||||
|
||||
def neutrale_kategorie(owner_id: int) -> int | None:
|
||||
"""Die Kategorie, die aus Auswertungen herausfällt („Kontoübertrag")."""
|
||||
r = con.execute(
|
||||
"SELECT id FROM categories WHERE owner_id = ? AND excluded_from_stats = 1 "
|
||||
"ORDER BY (name LIKE '%bertrag%') DESC, id LIMIT 1", (owner_id,)).fetchone()
|
||||
return r["id"] if r else None
|
||||
|
||||
|
||||
def eur_cents(betrag: int, waehrung: str, tag: str, ersatz: int) -> int:
|
||||
"""Betrag in EUR – für die Statistik. Ohne Kurs bleibt der bisherige Wert."""
|
||||
if waehrung.upper() == "EUR":
|
||||
return betrag
|
||||
r = con.execute(
|
||||
"SELECT rate_per_eur FROM exchange_rates WHERE currency = ? AND date <= ? "
|
||||
"ORDER BY date DESC LIMIT 1", (waehrung.upper(), tag)).fetchone()
|
||||
if not r or not r["rate_per_eur"]:
|
||||
return ersatz
|
||||
return round(betrag / float(r["rate_per_eur"]))
|
||||
|
||||
|
||||
uebertraege = con.execute(
|
||||
"SELECT * FROM transactions WHERE type = 'transfer' AND to_account_id IS NOT NULL"
|
||||
).fetchall()
|
||||
print(f"Datenbank: {DB}\n{len(uebertraege)} Überträge gefunden\n")
|
||||
if not uebertraege:
|
||||
raise SystemExit(0)
|
||||
|
||||
je_paar: dict[tuple[int, int], list] = {}
|
||||
for t in uebertraege:
|
||||
je_paar.setdefault((t["account_id"], t["to_account_id"]), []).append(t)
|
||||
print(f"{'von':<24}{'nach':<24}{'Anz':>5}{'Summe':>13}")
|
||||
print("-" * 66)
|
||||
for (a, b), liste in sorted(je_paar.items(), key=lambda kv: -len(kv[1])):
|
||||
na = konten[a]["name"] if a in konten else f"?{a}"
|
||||
nb = konten[b]["name"] if b in konten else f"?{b}"
|
||||
print(f"{na[:23]:<24}{nb[:23]:<24}{len(liste):>5}"
|
||||
f"{geld(sum(x['amount_cents'] for x in liste)):>13}")
|
||||
|
||||
fehlende = {t["owner_id"] for t in uebertraege if neutrale_kategorie(t["owner_id"]) is None}
|
||||
if fehlende:
|
||||
print("\nACHTUNG: Es gibt keine Kategorie mit „nicht in Auswertungen einrechnen“.")
|
||||
print("Lege eine an (z. B. „Kontoübertrag“) – sonst bleiben die Buchungen ohne Kategorie")
|
||||
print("und tauchen in der Statistik auf.")
|
||||
|
||||
if not ANWENDEN:
|
||||
print("\nNichts geändert (Probelauf). Mit --anwenden ausführen.")
|
||||
raise SystemExit(0)
|
||||
|
||||
sicherung = DB + ".vor-aufteilen"
|
||||
shutil.copy(DB, sicherung)
|
||||
print(f"\nSicherung: {sicherung}")
|
||||
|
||||
neu = 0
|
||||
for t in uebertraege:
|
||||
kat = neutrale_kategorie(t["owner_id"])
|
||||
ziel = konten.get(t["to_account_id"])
|
||||
gutschrift = t["to_amount_cents"] if t["to_amount_cents"] is not None else t["amount_cents"]
|
||||
|
||||
# Abgang bleibt beim Quellkonto – als Ausgabe.
|
||||
con.execute(
|
||||
"UPDATE transactions SET type='expense', to_account_id=NULL, to_amount_cents=NULL, "
|
||||
"category_id=COALESCE(category_id, ?) WHERE id = ?", (kat, t["id"]))
|
||||
|
||||
# Zugang wird eine eigene Einnahme beim Zielkonto.
|
||||
if ziel is not None:
|
||||
con.execute(
|
||||
"INSERT INTO transactions (owner_id, type, date, time, account_id, amount_cents, "
|
||||
"amount_eur_cents, category_id, payee, description, notes, import_batch_id, "
|
||||
"created_at, updated_at) VALUES (?, 'income', ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
||||
(t["owner_id"], t["date"], t["time"], ziel["id"], gutschrift,
|
||||
eur_cents(gutschrift, ziel["currency"], t["date"], t["amount_eur_cents"]),
|
||||
kat, t["payee"], t["description"], t["notes"], t["import_batch_id"],
|
||||
t["created_at"], t["updated_at"]))
|
||||
neu += 1
|
||||
|
||||
con.commit()
|
||||
print(f"{len(uebertraege)} Überträge zu Ausgaben gemacht, {neu} Einnahmen ergänzt")
|
||||
print("Die Salden bleiben unverändert. Dienst neu starten:")
|
||||
print(" sudo systemctl restart certain-finance")
|
||||
Reference in New Issue
Block a user