"""CSV-Import: Datei parsen, Vorschau mit Erkennung (Duplikate, interne Transfers, Auto-Kategorisierung), und die eigentliche Verbuchung inkl. Batch für Undo. """ import csv import hashlib import io import json import logging from collections import Counter from datetime import date, datetime from decimal import Decimal, InvalidOperation, ROUND_HALF_UP from sqlalchemy.orm import Session from ..models import ( Account, AccountIdentifier, InboxItem, InboxStatus, AmountMode, BudgetTreatment, CategorizationRule, Category, CategoryKind, ImportBatch, ImportStatus, ImportTemplate, MatchField, RuleSource, SAVINGS_ACCOUNT_TYPES, Transaction, TxnType, User, ) from . import currency log = logging.getLogger(__name__) # Vorgefertigte Bank-Vorlagen (aus echten Export-Beispielen). Erscheinen im # Import als auswählbare Presets und füllen die Format-/Spalten-Einstellungen. PRESETS: list[dict] = [ { "name": "Raiffeisen (CH)", # Kopfzeilen, an denen sich die Bank sicher erkennen laesst. "signature": ("iban", "booked at", "credit/debit amount", "balance"), "delimiter": ";", "decimal_sep": ".", "date_format": "%Y-%m-%d %H:%M:%S.%f", "encoding": "utf-8", "has_header": True, "skip_rows": 0, "amount_mode": "signed", # Spalten: 0=eigene IBAN, 1=Datum, 2=Text, 3=Betrag, 4=Saldo, 5=Valuta. "column_map": {"date": 1, "amount": 3, "payee": 2, "balance": 4, "purpose": None, "counterparty_iban": None}, }, { "name": "N26", "signature": ("booking date", "partner name", "payment reference"), "delimiter": ",", "decimal_sep": ".", "date_format": "%Y-%m-%d", "encoding": "utf-8", "has_header": True, "skip_rows": 0, "amount_mode": "signed", # Spalten: 0=Buchungsdatum, 1=Valuta, 2=Partner, 3=Partner-IBAN, 4=Typ, # 5=Verwendungszweck, 6=Kontoname, 7=Betrag (EUR), 8=Originalbetrag, # 9=Originalwährung, 10=Kurs. Eine Saldo-Spalte gibt es hier nicht, # deshalb ist beim Import kein Saldo-Abgleich möglich. "column_map": {"date": 0, "amount": 7, "payee": 2, "purpose": 5, "counterparty_iban": 3}, }, { "name": "Revolut", "signature": ("produkt", "beschreibung", "betrag", "kontostand"), # Der englische Export derselben Bank. "signature_alt": ("product", "description", "amount", "balance"), "delimiter": ",", "decimal_sep": ".", "date_format": "%Y-%m-%d %H:%M:%S", "encoding": "utf-8", "has_header": True, "skip_rows": 0, "amount_mode": "signed", # Spalten: 0=Typ, 1=Produkt, 2=Start, 3=Abgeschlossen, 4=Beschreibung, # 5=Betrag, 6=Gebühr, 7=Währung, 8=Status, 9=Saldo. "column_map": {"date": 3, "amount": 5, "payee": 4, "fee": 6, "currency": 7, "state": 8, "balance": 9, "purpose": None, "counterparty_iban": None}, }, ] # --- Format erkennen ------------------------------------------------------- # Reihenfolge zaehlt: Das erste Format, das die Beispielwerte vollstaendig # liest, gewinnt. Mit Uhrzeit zuerst, weil ein Datumsformat OHNE Uhrzeit auch # auf Zeilen MIT Uhrzeit passt (parse_date ist nachsichtig) – umgekehrt nicht. DATUMSFORMATE = ( "%Y-%m-%d %H:%M:%S.%f", "%Y-%m-%d %H:%M:%S", "%Y-%m-%d", "%d.%m.%Y %H:%M:%S", "%d.%m.%Y", "%d/%m/%Y", "%m/%d/%Y", ) TRENNZEICHEN = (";", ",", "\t", "|") # Kopfzeilen-Wörter, die auf ein Feld hindeuten. Der Inhalt entscheidet, aber # bei mehreren gleich guten Spalten gibt die Beschriftung den Ausschlag. KOPF_HINWEISE = { "date": ("buchung", "booked", "datum", "date", "valuta"), "amount": ("betrag", "amount", "umsatz", "credit", "debit"), "balance": ("saldo", "balance", "kontostand"), "payee": ("empfaenger", "empfänger", "partner", "beschreibung", "description", "text", "name", "auftraggeber"), "purpose": ("zweck", "reference", "verwendung", "mitteilung", "referenz"), "counterparty_iban": ("iban",), } def _spaltenwerte(zeilen: list[list[str]], i: int) -> list[str]: return [r[i].strip() for r in zeilen if i < len(r) and r[i].strip()] def _datumsanteil(werte: list[str], fmt: str) -> float: """Streng prüfen, nicht mit ``parse_date``. ``parse_date`` ist absichtlich nachsichtig: Es liest „01.02.2026“ auch mit dem Format „%d.%m.%Y %H:%M:%S“. Beim Erkennen ist das schädlich – dann gewinnt ein Format mit Uhrzeit gegen das richtige ohne, und in den Einstellungen steht hinterher etwas Falsches. """ if not werte: return 0.0 treffer = 0 for w in werte: try: datetime.strptime(w.strip(), fmt) treffer += 1 except (ValueError, TypeError): pass return treffer / len(werte) def _geldanteil(werte: list[str], sep: str) -> float: """Anteil der Werte, die wie ein Geldbetrag aussehen. Reine Ganzzahlen zaehlen nur halb: Referenznummern und Kurse sehen sonst aus wie Betraege, und eine Spalte „Exchange Rate" als Betrag zu nehmen waere ein stiller, teurer Fehler. """ if not werte: return 0.0 punkte = 0.0 for w in werte: if parse_amount(w, sep) is None: continue punkte += 1.0 if (sep in w or "-" in w) else 0.5 return punkte / len(werte) def _kopf_passt(kopf: str, feld: str) -> bool: k = normalize_text(kopf) return any(h in k for h in KOPF_HINWEISE.get(feld, ())) def erkenne_format(data: bytes, encoding: str = "utf-8") -> dict: """Aus einer CSV-Datei ableiten, wie sie zu lesen ist. Zuerst gegen die Bank-Vorlagen: Stimmt die Kopfzeile, ist alles bekannt. Sonst inhaltsbasiert – welche Spalte enthaelt Daten, welche Betraege, welche den laengsten Text. Das erspart das Ausfuellen des Formulars, das bisher bei jedem Import noetig war. Liefert die Einstellung, eine Begruendung je Feld und wie sicher der Fund ist. Unsicheres wird gemeldet, nicht verschwiegen. """ text = decode_bytes(data, encoding) erste = next((z for z in text.splitlines() if z.strip()), "") # Trennzeichen: das haeufigste in der Kopfzeile, das ueberhaupt trennt. trenner = max(TRENNZEICHEN, key=lambda t: erste.count(t)) if erste.count(trenner) == 0: trenner = ";" rohzeilen = [r for r in csv.reader(io.StringIO(text), delimiter=trenner) if any((c or "").strip() for c in r)] if not rohzeilen: return {"ok": False, "grund": "Die Datei enthält keine lesbaren Zeilen."} kopf = [c.strip() for c in rohzeilen[0]] zeilen = rohzeilen[1:41] # eine Handvoll reicht zum Erkennen if not zeilen: return {"ok": False, "grund": "Die Datei enthält nur eine Kopfzeile."} kopf_klein = [normalize_text(c) for c in kopf] for p in PRESETS: for schluessel in ("signature", "signature_alt"): sig = p.get(schluessel) if sig and all(any(s in k for k in kopf_klein) for s in sig): return { "ok": True, "preset": p["name"], "sicherheit": "hoch", "begruendung": f"Kopfzeile passt zur Vorlage „{p['name']}“.", "config": {k: v for k, v in p.items() if k not in ("name", "signature", "signature_alt")}, "header": kopf, } # Zuerst die Datumsspalte: Deutsche Datumsangaben („15.03.2026") stecken # voller Punkte und würden sonst die Erkennung des Dezimaltrenners kippen. datum_i, datum_fmt, bestes = None, "%d.%m.%Y", 0.0 for i in range(len(kopf)): werte = _spaltenwerte(zeilen, i) for fmt in DATUMSFORMATE: anteil = _datumsanteil(werte, fmt) if anteil > bestes or (anteil == bestes and anteil > 0.8 and _kopf_passt(kopf[i], "date")): datum_i, datum_fmt, bestes = i, fmt, anteil if datum_i is None or bestes < 0.6: return {"ok": False, "grund": "Keine Spalte mit Datumswerten gefunden.", "header": kopf} # ALLE Datumsspalten heraushalten, nicht nur die gewählte: Viele Auszüge # führen Buchungs- UND Valutadatum, und beide zählen sonst als Punkte mit. datumsspalten = { i for i in range(len(kopf)) if any(_datumsanteil(_spaltenwerte(zeilen, i), f) > 0.8 for f in DATUMSFORMATE) } ohne_datum = " ".join( " ".join(c for j, c in enumerate(r) if j not in datumsspalten) for r in zeilen ) dezimal = "," if ohne_datum.count(",") > ohne_datum.count(".") and trenner != "," else "." geld = sorted( ((i, _geldanteil(_spaltenwerte(zeilen, i), dezimal)) for i in range(len(kopf)) if i != datum_i), key=lambda x: (-x[1], not _kopf_passt(kopf[x[0]], "amount")), ) geldspalten = [i for i, anteil in geld if anteil > 0.6] if not geldspalten: return {"ok": False, "grund": "Keine Spalte mit Beträgen gefunden.", "header": kopf} betrag_i = next((i for i in geldspalten if _kopf_passt(kopf[i], "amount")), geldspalten[0]) saldo_i = next((i for i in geldspalten if i != betrag_i and _kopf_passt(kopf[i], "balance")), None) text_spalten = sorted( ((i, sum(len(w) for w in _spaltenwerte(zeilen, i)) / max(1, len(zeilen))) for i in range(len(kopf)) if i not in (datum_i, betrag_i, saldo_i)), key=lambda x: -x[1], ) lange = [i for i, laenge in text_spalten if laenge >= 3] payee_i = next((i for i in lange if _kopf_passt(kopf[i], "payee")), lange[0] if lange else None) purpose_i = next((i for i in lange if i != payee_i and _kopf_passt(kopf[i], "purpose")), None) iban_i = next((i for i in range(len(kopf)) if i not in (datum_i, betrag_i) and _kopf_passt(kopf[i], "counterparty_iban") and any(len(w) > 14 for w in _spaltenwerte(zeilen, i))), None) spalten = {"date": datum_i, "amount": betrag_i, "payee": payee_i, "purpose": purpose_i, "balance": saldo_i, "counterparty_iban": iban_i} begruendung = " · ".join( f"{feld}: Spalte {i} „{kopf[i]}“" for feld, i in spalten.items() if i is not None ) return { "ok": True, "preset": None, "sicherheit": "hoch" if bestes > 0.9 and payee_i is not None else "mittel", "begruendung": begruendung, "config": { "delimiter": trenner, "decimal_sep": dezimal, "date_format": datum_fmt, "encoding": encoding, "has_header": True, "skip_rows": 0, "amount_mode": "signed", "column_map": spalten, }, "header": kopf, } def erkenne_konto(db: Session, user: User, data: bytes, erkannt: dict) -> dict: """Zu welchem Konto gehört diese Datei? Damit reicht EIN Kurzbefehl fürs Telefon statt einer je Konto. Zwei Anhaltspunkte, in dieser Reihenfolge: 1. Eine Spalte mit der EIGENEN IBAN (Raiffeisen führt sie in Spalte 0). Trifft sie einen Erkennungs-Text eines Kontos, ist die Sache eindeutig. 2. Die Währung der Zeilen zusammen mit der erkannten Bank. Revolut liefert je Währung eine eigene Datei; welches Konto gemeint ist, verrät der Import-Verlauf – dorthin ging zuletzt eine Datei derselben Bank. Bleibt es mehrdeutig, wird NICHT geraten: Ein Auszug im falschen Konto verfälscht zwei Salden auf einmal. """ if not erkannt.get("ok"): return {"account_id": None, "grund": erkannt.get("grund", "Format unbekannt")} config = erkannt["config"] text = decode_bytes(data, config.get("encoding") or "utf-8") zeilen = [r for r in csv.reader(io.StringIO(text), delimiter=config["delimiter"]) if any((c or "").strip() for c in r)] if len(zeilen) < 2: return {"account_id": None, "grund": "Zu wenige Zeilen"} proben = zeilen[1:31] konten = db.query(Account).filter(Account.owner_id == user.id, Account.is_archived.is_(False)).all() # 1. Eigene IBAN in einer Spalte. idents: dict[str, int] = {} for acc_id, wert in db.query(AccountIdentifier.account_id, AccountIdentifier.value_norm).join( Account, Account.id == AccountIdentifier.account_id ).filter(Account.owner_id == user.id): if wert: idents[normalize_iban(wert)] = acc_id breite = max(len(r) for r in proben) for i in range(breite): werte = {normalize_iban(r[i]) for r in proben if i < len(r) and r[i].strip()} werte = {w for w in werte if len(w) >= 15} if len(werte) == 1: treffer = idents.get(next(iter(werte))) if treffer: return {"account_id": treffer, "grund": "eigene IBAN in der Datei", "sicher": True} # 2. Währung plus zuletzt benutztes Konto derselben Bank. waehrung_i = config["column_map"].get("currency") waehrungen = set() if waehrung_i is not None: waehrungen = {r[waehrung_i].strip().upper() for r in proben if waehrung_i < len(r) and r[waehrung_i].strip()} if len(waehrungen) == 1: w = next(iter(waehrungen)) passend = [a for a in konten if a.currency.upper() == w] if len(passend) == 1: return {"account_id": passend[0].id, "grund": f"einziges Konto in {w}", "sicher": True} if passend and erkannt.get("preset"): frueher = { b.account_id for b in db.query(ImportBatch).filter( ImportBatch.owner_id == user.id, ImportBatch.preset == erkannt["preset"], ImportBatch.status == ImportStatus.committed, ) } treffer = [a for a in passend if a.id in frueher] if len(treffer) == 1: return {"account_id": treffer[0].id, "grund": f"{w} und zuletzt für „{erkannt['preset']}“ benutzt", "sicher": True} return {"account_id": None, "grund": "Das Konto lässt sich aus der Datei nicht sicher ableiten"} # --- Kleine Parser-Helfer -------------------------------------------------- def normalize_text(s: str | None) -> str: return (s or "").strip().lower() def normalize_iban(s: str | None) -> str: return "".join(ch for ch in (s or "").upper() if ch.isalnum()) def parse_amount(raw: str | None, decimal_sep: str) -> int | None: """Betragstext -> Cent (mit Vorzeichen). Erkennt Tausender-Trenner und Klammern.""" s = str(raw or "").strip() if not s: return None neg = ("-" in s) or (s.startswith("(") and s.endswith(")")) thousands = "." if decimal_sep == "," else "," s2 = s.replace(thousands, "").replace(decimal_sep, ".") digits = "".join(ch for ch in s2 if ch.isdigit() or ch == ".") if digits in ("", "."): return None try: val = Decimal(digits) except InvalidOperation: return None cents = int((val * 100).to_integral_value(rounding=ROUND_HALF_UP)) return -cents if neg else cents def parse_date(raw: str | None, fmt: str) -> date | None: """Datumstext -> Datum. Nachsichtig gegenüber einer Uhrzeit, die im Format fehlt oder umgekehrt: Banken exportieren dieselbe Spalte mal als „2026-01-06", mal als „2026-01-06 23:11:14" (auch mit Sekundenbruchteilen). Sonst müsste man im Dropdown für jede Variante ein eigenes Format anbieten.""" s = str(raw or "").strip() if not s: return None try: return datetime.strptime(s, fmt).date() except ValueError: pass hat_zeit_im_format = any(x in fmt for x in ("%H", "%M", "%S", "%f")) # Wert bringt eine Uhrzeit mit, das Format kennt keine -> nur den Datumsteil. if not hat_zeit_im_format and (" " in s or "T" in s): kopf = s.replace("T", " ").split(" ", 1)[0] try: return datetime.strptime(kopf, fmt).date() except ValueError: pass # Format erwartet eine Uhrzeit, der Wert hat keine -> Format kürzen. if hat_zeit_im_format: nur_datum = fmt.replace("T", " ").split(" ", 1)[0] try: return datetime.strptime(s.replace("T", " ").split(" ", 1)[0], nur_datum).date() except ValueError: pass return None def decode_bytes(data: bytes, encoding: str) -> str: for enc in (encoding, "utf-8-sig", "utf-8", "latin-1"): try: return data.decode(enc) except (UnicodeDecodeError, LookupError): continue return data.decode("utf-8", errors="replace") def read_rows(data: bytes, template: ImportTemplate) -> list[list[str]]: text = decode_bytes(data, template.encoding) delim = (template.delimiter or ";")[0] reader = csv.reader(io.StringIO(text), delimiter=delim) rows = [r for r in reader if any((c or "").strip() for c in r)] start = template.skip_rows + (1 if template.has_header else 0) rows = rows[start:] # Zeilenfilter: Revolut legt alle Pockets in eine Datei, jedes mit eigenem # Saldo-Verlauf. Ohne Filter landet alles auf demselben Konto. spalte = template.filter_column wert = (template.filter_value or "").strip().lower() if spalte is not None and wert: rows = [ r for r in rows if spalte < len(r) and (r[spalte] or "").strip().lower() == wert ] return rows # Eine Spalte taugt als Zeilenfilter, wenn sie die Datei in wenige, klar # benannte Gruppen teilt – etwa „Produkt" bei Revolut (Giro, Erspartes, # Anlagekonto). Mehr als eine Handvoll Werte ist keine Gruppe mehr, sondern # ein Datenfeld wie Empfänger oder Betrag. MAX_FILTERWERTE = 12 def preview_raw(data: bytes, template: ImportTemplate, limit: int = 8) -> dict: """Rohe erste Zeilen (zum Spalten-Zuordnen im Template-Editor). Liefert zusätzlich je Spalte die wählbaren Filterwerte mit Zeilenzahl. Revolut legt alle Produkte in EINE Datei, jedes mit eigener Kontostand- Spalte; ohne Filter zählen die Bewegungen doppelt und der Saldo-Abgleich kann gar nicht aufgehen. Die Werte hier zu kennen erspart das Raten. """ text = decode_bytes(data, template.encoding) delim = (template.delimiter or ";")[0] reader = csv.reader(io.StringIO(text), delimiter=delim) rows = [r for r in reader if any((c or "").strip() for c in r)] header = None body_start = template.skip_rows if template.has_header and len(rows) > body_start: header = rows[body_start] body_start += 1 body = rows[body_start:] filterwerte: dict[int, list[dict]] = {} breite = max((len(r) for r in body), default=0) for i in range(breite): zaehler = Counter((r[i] or "").strip() for r in body if i < len(r)) zaehler.pop("", None) # Wiederholen sich die Werte nicht, ist es keine Gruppe, sondern ein # Datenfeld wie Empfänger oder Betrag. if 2 <= len(zaehler) <= MAX_FILTERWERTE and len(zaehler) < len(body): filterwerte[i] = [{"value": v, "count": n} for v, n in zaehler.most_common()] return { "header": header, "rows": body[:limit], "row_count": len(body), "filter_values": filterwerte, } def import_hash(account_id: int, d: date | None, amount_cents: int | None, payee: str | None) -> str: key = f"{account_id}|{d}|{amount_cents}|{normalize_text(payee)}" return hashlib.sha256(key.encode("utf-8")).hexdigest() # --- Vorschau -------------------------------------------------------------- def _rule_sort_key(rule: CategorizationRule) -> tuple: """Reihenfolge der Regelanwendung: Betrags-Regeln (spezifischer) zuerst, dann längstes Schlagwort, dann manuelle vor gelernten, dann Priorität/Treffer.""" return ( 1 if rule.amount_cents is not None else 0, len(rule.pattern or ""), 1 if rule.source == RuleSource.manual else 0, rule.priority, rule.hits, ) def _pattern_hit(rule: CategorizationRule, p: str, z: str) -> bool: if rule.match_field == MatchField.payee: return rule.pattern in p if rule.match_field == MatchField.purpose: return rule.pattern in z return rule.pattern in p or rule.pattern in z # any: Empfänger ODER Zweck def _match_rule( rules: list[CategorizationRule], payee: str, purpose: str, amount: int | None, allowed_kind: CategoryKind | None, cat_kind: dict[int, CategoryKind | None], ) -> CategorizationRule | None: """Erste passende Regel (Muster + optionale Betragsbedingung). Kategorie-Regeln greifen nur, wenn die Kategorie-Art (Einnahme/Ausgabe) zur Buchung passt; Transfer-Regeln kennen keine Art.""" p = normalize_text(payee) z = normalize_text(purpose) absamt = abs(amount) if amount is not None else None for rule in rules: # bereits sortiert (Betrag/längstes/manuell zuerst) if not rule.pattern: continue if rule.amount_cents is not None and (absamt is None or absamt != rule.amount_cents): continue if rule.is_transfer: if _pattern_hit(rule, p, z): return rule continue if not rule.category_id: continue # Neutrale Kategorien („Kontoübertrag") passen in beide Richtungen: Geld # geht auf dem einen Konto raus und kommt auf dem anderen rein. art = cat_kind.get(rule.category_id) if allowed_kind is not None and art is not None and art != allowed_kind: continue if _pattern_hit(rule, p, z): return rule return None # Status-Werte, die eine Buchung NICHT wirksam werden lassen. Bewusst als # Sperrliste (nicht Positivliste): Bank-Exporte sind lokalisiert – „COMPLETED" # heißt auf Deutsch „ABGESCHLOSSEN". Unbekannte Werte werden gebucht, damit # keine echten Buchungen still verloren gehen. UNBOOKABLE_STATES = ( "PENDING", "AUSSTEHEND", "SCHWEBEND", "REVERTED", "RUECKGAENGIG", "RÜCKGÄNGIG", "STORNIERT", "STORNO", "DECLINED", "ABGELEHNT", "FAILED", "FEHLGESCHLAGEN", "CANCELLED", "CANCELED", "ABGEBROCHEN", ) def state_is_bookable(state: str | None) -> bool: s = (state or "").strip().upper() if not s: return True return not any(bad in s for bad in UNBOOKABLE_STATES) # Fenster, in dem eine Buchung als Gegenseite eines bestehenden Transfers gilt # (Buchungs-/Valutadatum können ein paar Tage auseinanderliegen). # Wie weit der tatsächlich gutgeschriebene Betrag vom geschätzten abweichen darf, # damit er noch als dieselbe Bewegung gilt. Bei einem Währungstausch rechnen wir # mit dem EZB-Kurs, die Bank nutzt ihren eigenen – ein paar Prozent Unterschied # sind normal, eine andere Buchung liegt praktisch nie so nah. def parse_file_to_candidates(data: bytes, template: ImportTemplate) -> list[dict]: """Zerlegt die Datei anhand des Templates in rohe Kandidaten-Zeilen.""" cmap = json.loads(template.column_map or "{}") def col(field: str) -> int | None: idx = cmap.get(field) return idx if isinstance(idx, int) and idx >= 0 else None out: list[dict] = [] vorheriger_saldo: int | None = None for r in read_rows(data, template): def cell(field: str) -> str: c = col(field) return r[c].strip() if c is not None and c < len(r) else "" d = parse_date(cell("date"), template.date_format) if template.amount_mode == AmountMode.debit_credit: debit = parse_amount(cell("debit"), template.decimal_sep) credit = parse_amount(cell("credit"), template.decimal_sep) amount = (credit or 0) - (abs(debit) if debit else 0) elif template.amount_mode == AmountMode.abs_expense: a = parse_amount(cell("amount"), template.decimal_sep) amount = -abs(a) if a is not None else None else: amount = parse_amount(cell("amount"), template.decimal_sep) # Getrennte Gebührenspalte (z. B. Revolut) mindert den Betrag. fee = parse_amount(cell("fee"), template.decimal_sep) balance = parse_amount(cell("balance"), template.decimal_sep) if amount is not None and fee: if amount == 0: # Reine Gebührenzeile: Ob belastet oder erstattet („Erstattung für # Transaktion"), steht nicht im Betrag – beide Male 0.00 bei # gleicher Gebühr. Die Saldo-Spalte weiß es; ohne sie ist eine # Belastung der Normalfall. delta = (balance - vorheriger_saldo) if ( balance is not None and vorheriger_saldo is not None ) else None amount = delta if delta is not None else -abs(fee) else: amount -= abs(fee) if balance is not None: vorheriger_saldo = balance out.append({ "date": d.isoformat() if d else None, "amount_cents": amount, "payee": cell("payee") or None, "purpose": cell("purpose") or None, "counterparty_iban": normalize_iban(cell("counterparty_iban")) or None, "state": (cell("state").strip().upper() or None), "row_currency": (cell("currency").strip().upper() or None), # Laufender Kontostand aus der Datei – Grundlage für den Saldo-Abgleich. "balance_cents": balance, }) return out INBOX_TAGE = 3 # Bank bucht oft ein bis zwei Tage nach der Zahlung def _find_inbox_match( eingang: list[tuple], verbraucht: set[int], d: date, amount: int ) -> int | None: """Passenden, noch nicht verbrauchten Eingangs-Eintrag suchen. Verglichen wird der Betrag exakt – bei Geld ist Toleranz gefährlich – und das Datum mit ein paar Tagen Spielraum. Der zeitlich nächste gewinnt, damit bei wiederkehrenden Beträgen (jeden Tag derselbe Kaffee) die richtige Zeile getroffen wird. """ beste: tuple[int, int] | None = None # (Abstand in Tagen, ID) for eid, edatum, ebetrag in eingang: if eid in verbraucht or ebetrag != amount: continue abstand = abs((d - edatum).days) if abstand > INBOX_TAGE: continue if beste is None or abstand < beste[0]: beste = (abstand, eid) return beste[1] if beste else None def detect_rows(db: Session, user: User, account: Account, candidates: list[dict]) -> list[dict]: """Reichert Kandidaten an: Typ, Auto-Kategorie, Duplikat. Wird von CSV- UND Beleg-Import genutzt (gleiche Prüf-/Buchungslogik). Konten werden hier NICHT mehr miteinander verbunden – jede Zeile ist Ausgabe oder Einnahme. Überträge bucht man von Hand. """ rules = db.query(CategorizationRule).filter(CategorizationRule.owner_id == user.id).all() rules.sort(key=_rule_sort_key, reverse=True) # längstes/​manuell zuerst # None = neutral: zählt für Ausgabe UND Einnahme (siehe _match_rule). cat_kind: dict[int, CategoryKind | None] = { cid: (None if neutral else kind) for cid, kind, neutral in db.query( Category.id, Category.type, Category.excluded_from_stats ).filter(Category.owner_id == user.id) } # Duplikate als Multiset zählen: eine Zeile gilt nur dann als Duplikat, wenn # bereits GENAU SO VIELE identische Buchungen in der DB stehen. So werden echte, # gleich aussehende Buchungen desselben Tags (z. B. zwei Gutschriften über 324) # behalten, ein erneuter Import derselben Datei aber weiter erkannt. existing: Counter = Counter( h[0] for h in db.query(Transaction.import_hash).filter( Transaction.owner_id == user.id, Transaction.import_hash.isnot(None) ) ) # Neutrale Kategorie („Kontoübertrag"): Buchungen, die nur Geld zwischen # eigenen Konten verschieben, werden damit kategorisiert statt zu einem # Übertrag verbunden. Sie verändern die Salden korrekt und bleiben aus der # Statistik heraus. neutrale_kategorie = next( (cid for cid, k in cat_kind.items() if k is None), None ) # Zahlungen, die schon von aussen erfasst wurden (Kurzbefehl, später Mail). # Der Import-Schlüssel greift hier NICHT: Der Text vom Telefon ist selten # wortgleich mit dem des Auszugs, und das Buchungsdatum der Bank liegt oft # ein bis zwei Tage später. Deshalb wird zusätzlich unscharf verglichen – # gleicher Betrag, Datum in der Nähe. eingang = [ (eid, edatum, ebetrag) for eid, edatum, ebetrag in db.query( InboxItem.id, InboxItem.date, InboxItem.amount_cents ).filter( InboxItem.owner_id == user.id, InboxItem.account_id == account.id, InboxItem.status.in_((InboxStatus.booked, InboxStatus.discarded)), ) ] eingang_verbraucht: set[int] = set() seen: Counter = Counter() out: list[dict] = [] for i, cand in enumerate(candidates): d_iso = cand.get("date") amount = cand.get("amount_cents") payee = cand.get("payee") purpose = cand.get("purpose") iban = cand.get("counterparty_iban") d = date.fromisoformat(d_iso) if d_iso else None # Nur abgeschlossene Buchungen zählen. Revolut liefert auch stornierte # (REVERTED) und schwebende (PENDING) Zeilen mit – die dürfen den Saldo # nicht verändern. state = (cand.get("state") or "").upper() state_ok = state_is_bookable(state) # Zeile in fremder Währung (Revolut-Export enthält alle Währungs-Pockets). row_cur = (cand.get("row_currency") or "").upper() currency_ok = not row_cur or row_cur == account.currency.upper() valid = ( d is not None and amount is not None and amount != 0 and state_ok and currency_ok ) # Jede Zeile ist Ausgabe oder Einnahme – der Import verbindet keine # Konten mehr selbst. Das Raten des Gegenkontos hat mehr Schaden # angerichtet als Nutzen gebracht: geratene Ziele, verdrehte Richtungen # und doppelt gebuchte Bewegungen, wenn beide Auszüge eingelesen wurden. # Wer zwei Buchungen bewusst verbinden will, bucht den Übertrag von Hand. ttype = "expense" if (amount or 0) < 0 else "income" category_id: int | None = None matched_rule_id: int | None = None to_account_id: int | None = None kind = CategoryKind.expense if ttype == "expense" else CategoryKind.income rule = _match_rule(rules, payee or "", purpose or "", amount, kind, cat_kind) if rule is not None: matched_rule_id = rule.id # Frühere Übertrags-Schlagwörter kennzeichnen jetzt die neutrale # Kategorie, statt eine Verbindung zwischen Konten herzustellen. category_id = rule.category_id or (neutrale_kategorie if rule.is_transfer else None) h = import_hash(account.id, d, amount, payee) is_dup = seen[h] < existing[h] # nur so viele Duplikate wie schon in der DB seen[h] += 1 dup_grund = "hash" if is_dup else None if not is_dup and d is not None and amount: treffer = _find_inbox_match(eingang, eingang_verbraucht, d, amount) if treffer is not None: eingang_verbraucht.add(treffer) is_dup = True dup_grund = "inbox" is_counterpart = False braucht_ziel = False out.append({ "index": i, "date": d.isoformat() if d else None, "amount_cents": amount, "payee": payee or None, "purpose": purpose or None, "counterparty_iban": iban or None, "valid": valid, "skip_reason": ( None if valid or d is None or amount is None else ("state" if not state_ok else "currency" if not currency_ok else None) ), "state": state or None, "row_currency": row_cur or None, "balance_cents": cand.get("balance_cents"), "type": ttype, "category_id": category_id, "matched_rule_id": matched_rule_id, "to_account_id": to_account_id, "import_hash": h, "is_duplicate": is_dup, "duplicate_reason": dup_grund, "is_counterpart": is_counterpart, # Damit commit_import die geschätzte Gegenseite auf den echten Wert setzt. "counterpart_txn_id": None, "include": valid and not is_dup and not is_counterpart and not braucht_ziel, "needs_target": braucht_ziel, }) return out def reconcile(rows: list[dict]) -> dict: """Gleicht die erkannten Beträge gegen die Saldo-Spalte der Datei ab. Die Saldo-Spalte ist der Schiedsrichter: Sie sagt für jede Zeile, wie sich der Kontostand tatsächlich verändert hat. Weicht unser Betrag ab, stimmt die Spaltenzuordnung oder die Gebühren-/Vorzeichen-Behandlung nicht.""" # Zeilen anderer Währung gehören zu einem anderen Saldo-Verlauf (Revolut legt # alle Pockets in eine Datei) – die würden die Kette zerreißen. seq = [ r for r in rows if r.get("balance_cents") is not None and r.get("skip_reason") != "currency" ] if len(seq) < 2: return {"available": False} def gebucht(r: dict) -> int: """Betrag, der im Konto ankommt – 0, wenn die Zeile nirgends landet. Duplikate und Gegenbuchungen zählen mit: deren Betrag steckt bereits in einer früher gebuchten Buchung. Eine Zeile, die auf ein Gegenkonto wartet, zählt dagegen NICHT – sie ist unterwegs verloren, und genau das soll der Abgleich melden.""" if not r.get("valid") or r.get("needs_target"): return 0 return r.get("amount_cents") or 0 # Manche Banken exportieren neueste Buchung zuerst – für den Abgleich drehen. dated = [r["date"] for r in seq if r.get("date")] if len(dated) >= 2 and dated[0] > dated[-1]: seq = list(reversed(seq)) opening = seq[0]["balance_cents"] - gebucht(seq[0]) final = seq[-1]["balance_cents"] computed = opening + sum(gebucht(r) for r in seq) first_mismatch = None mismatches = 0 prev = seq[0]["balance_cents"] for r in seq[1:]: expected = r["balance_cents"] - prev # was die Bank verbucht hat actual = gebucht(r) # was wir tatsächlich buchen if expected != actual: mismatches += 1 if first_mismatch is None: first_mismatch = { "index": r.get("index"), "date": r.get("date"), "payee": r.get("payee") or r.get("purpose"), "expected_cents": expected, "actual_cents": actual, } prev = r["balance_cents"] return { "available": True, "opening_balance_cents": opening, "final_balance_cents": final, "computed_final_cents": computed, "difference_cents": final - computed, "checked": len(seq) - 1, "mismatches": mismatches, "ok": final == computed and mismatches == 0, "first_mismatch": first_mismatch, } def build_preview(db: Session, user: User, account: Account, template: ImportTemplate, data: bytes) -> list[dict]: return detect_rows(db, user, account, parse_file_to_candidates(data, template)) # --- Verbuchen ------------------------------------------------------------- def _make_transaction(db: Session, user: User, account: Account, accounts: dict[int, Account], row: dict, batch_id: int) -> Transaction | None: d = date.fromisoformat(row["date"]) if row.get("date") else None amount = row.get("amount_cents") if d is None or not amount: return None ttype = row.get("type") or ("expense" if amount < 0 else "income") payee = row.get("payee") purpose = row.get("purpose") notes = None if ttype == "transfer": other_id = row.get("to_account_id") other = accounts.get(other_id) if other_id else None if other is None: ttype = "expense" if amount < 0 else "income" # Fallback ohne Zielkonto if ttype == "transfer": other = accounts[row["to_account_id"]] mag = abs(amount) if amount < 0: from_acc, to_acc = account, other from_amount = mag # in Kontowährung (bekannt aus dem Auszug) else: from_acc, to_acc = other, account # from-Betrag in der Währung des anderen Kontos schätzen. conv = currency.convert(db, mag, account.currency, from_acc.currency, d) from_amount = conv if conv is not None else mag to_amount = mag if amount < 0 else mag # Konsistente to_amount-Berechnung: if from_acc.currency == to_acc.currency: to_amount = from_amount else: conv = currency.convert(db, from_amount, from_acc.currency, to_acc.currency, d) to_amount = conv if conv is not None else from_amount # Sparen/Depot-Ziel budgetieren (überschreibbar über row["budget_treatment"]). treatment = row.get("budget_treatment") budget_cat = row.get("budget_category_id") txn = Transaction( owner_id=user.id, type=TxnType.transfer, date=d, account_id=from_acc.id, to_account_id=to_acc.id, amount_cents=from_amount, to_amount_cents=to_amount, amount_eur_cents=currency.compute_amount_eur_cents(db, from_amount, from_acc.currency, d), budget_treatment=BudgetTreatment(treatment) if treatment in ("budgeted", "neutral") else None, budget_category_id=budget_cat, description=purpose, payee=payee, notes=notes, import_batch_id=batch_id, import_hash=row.get("import_hash"), ) return txn mag = abs(amount) txn = Transaction( owner_id=user.id, type=TxnType.expense if ttype == "expense" else TxnType.income, date=d, account_id=account.id, amount_cents=mag, amount_eur_cents=currency.compute_amount_eur_cents(db, mag, account.currency, d), category_id=row.get("category_id"), payee=payee, description=purpose, notes=notes, import_batch_id=batch_id, import_hash=row.get("import_hash"), ) return txn def _learn_rule(db: Session, user: User, row: dict) -> None: """Merkt sich Empfänger -> Kategorie für künftige Importe.""" cat = row.get("category_id") payee = row.get("payee") if not cat or not payee or row.get("type") == "transfer": return pattern = normalize_text(payee)[:255] if not pattern: return # Art der gelernten Regel = Art der Kategorie (Einnahme/Ausgabe). Gehört zur # Eindeutigkeit (owner, feld, wort, kind), damit dasselbe Wort je Art zählt. ckind = db.query(Category.type).filter( Category.id == cat, Category.owner_id == user.id ).scalar() kind = ckind.value if ckind is not None else "expense" existing = ( db.query(CategorizationRule) .filter( CategorizationRule.owner_id == user.id, CategorizationRule.match_field == MatchField.payee, CategorizationRule.pattern == pattern, CategorizationRule.kind == kind, ) .first() ) if existing is None: db.add(CategorizationRule( owner_id=user.id, match_field=MatchField.payee, pattern=pattern, category_id=cat, kind=kind, source=RuleSource.learned, priority=1, hits=1, )) # Sichtbar machen (Session ist autoflush=False), damit dasselbe Muster ein # zweites Mal im selben Import die vorhandene Regel findet statt zu doppeln. db.flush() else: existing.category_id = cat existing.hits += 1 def _track_and_learn(db: Session, user: User, row: dict) -> None: """Zählt für die zutreffende Regel Anwendung/Korrektur (Basis fürs Flaggen von Schlagwörtern) und lernt anschließend Empfänger -> Kategorie weiter.""" rid = row.get("matched_rule_id") final_cat = row.get("category_id") matched_manual_kept = False if rid: rule = db.get(CategorizationRule, rid) if rule is not None and rule.owner_id == user.id: rule.applied_count = (rule.applied_count or 0) + 1 if final_cat != rule.category_id: rule.overridden_count = (rule.overridden_count or 0) + 1 elif rule.source == RuleSource.manual: matched_manual_kept = True # Schlagwort greift schon – nicht doppelt lernen if not matched_manual_kept: _learn_rule(db, user, row) def _apply_counterpart_amount(db: Session, user: User, account: Account, row: dict) -> bool: """Trägt den tatsächlichen Betrag einer Umtausch-Gegenbuchung in den bereits gebuchten Transfer ein. Diese Zeile stammt aus dem Auszug der anderen Seite und nennt exakt, was dort angekommen bzw. abgegangen ist.""" txn_id = row.get("counterpart_txn_id") amount = row.get("amount_cents") if not txn_id or not amount or not row.get("is_counterpart"): return False txn = db.get(Transaction, txn_id) if txn is None or txn.owner_id != user.id or txn.type != TxnType.transfer: return False absamt = abs(amount) if txn.to_account_id == account.id: # hier ist das Geld angekommen if txn.to_amount_cents == absamt: return False txn.to_amount_cents = absamt elif txn.account_id == account.id: # hier ist es abgegangen if txn.amount_cents == absamt: return False txn.amount_cents = absamt txn.amount_eur_cents = currency.compute_amount_eur_cents( db, absamt, account.currency, txn.date ) else: return False return True def commit_import(db: Session, user: User, account: Account, template_id: int | None, filename: str, rows: list[dict], preset: str | None = None) -> ImportBatch: accounts = {a.id: a for a in db.query(Account).filter(Account.owner_id == user.id)} batch = ImportBatch( owner_id=user.id, account_id=account.id, template_id=template_id, preset=(preset or None), filename=(filename or "")[:255], row_count=len(rows), imported_count=0, status=ImportStatus.committed, ) db.add(batch) db.flush() imported = 0 korrigiert = 0 for row in rows: if not row.get("include", True): # Gegenbuchung eines Umtauschs: nicht buchen, aber den echten Betrag # dieser Seite in den bestehenden Transfer eintragen. Sonst bliebe dort # die EZB-Schätzung stehen und der Saldo wiche um die Kursdifferenz ab. if _apply_counterpart_amount(db, user, account, row): korrigiert += 1 continue txn = _make_transaction(db, user, account, accounts, row, batch.id) if txn is None: continue db.add(txn) _track_and_learn(db, user, row) imported += 1 batch.imported_count = imported db.commit() # Frisch gebuchte Zeilen auf Überträge prüfen. Läuft NACH dem Buchen, nicht # währenddessen: Der Import bleibt damit unangetastet, und wenn die Suche # etwas falsch macht, ist es ein Etikett – keine Buchung. try: from . import uebertraege uebertraege.finde(db, user) except Exception: # Nie den Import daran scheitern lassen log.exception("Übertrags-Suche nach dem Import fehlgeschlagen") db.refresh(batch) return batch def undo_batch(db: Session, user: User, batch: ImportBatch) -> int: n = ( db.query(Transaction) .filter(Transaction.owner_id == user.id, Transaction.import_batch_id == batch.id) .delete() ) batch.status = ImportStatus.undone db.commit() return n