GuidoandClaude Opus 4.8
d97e08e913
Der Revolut-Export führt "2026-01-06 23:11:14" in der Datumsspalte. Wählte man
im Dropdown JJJJ-MM-TT, scheiterte das Parsen an der Uhrzeit – jede Zeile hatte
kein Datum und galt als ungültig ("137 Zeilen erkannt, 0 werden gebucht"),
obwohl die Auswahl richtig aussah.
- parse_date schneidet eine überzählige Uhrzeit ab und kommt auch damit klar,
wenn das Format eine Uhrzeit erwartet, der Wert aber keine hat (ISO-T ebenso)
- Dropdown um die Varianten mit Uhrzeit ergänzt, damit die Bank-Vorlagen ihren
echten Eintrag zeigen statt eines leeren Feldes
- Test: reines Datumsformat auf Datei mit Uhrzeit -> alle Zeilen erkannt, 84 grün
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
672 lines
27 KiB
Python
672 lines
27 KiB
Python
"""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
|
||
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,
|
||
AmountMode,
|
||
BudgetTreatment,
|
||
CategorizationRule,
|
||
Category,
|
||
CategoryKind,
|
||
IdentifierKind,
|
||
ImportBatch,
|
||
ImportStatus,
|
||
ImportTemplate,
|
||
MatchField,
|
||
RuleSource,
|
||
SAVINGS_ACCOUNT_TYPES,
|
||
Transaction,
|
||
TxnType,
|
||
User,
|
||
)
|
||
from . import currency
|
||
|
||
|
||
# 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)",
|
||
"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": "Revolut",
|
||
"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},
|
||
},
|
||
]
|
||
|
||
|
||
# --- 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)
|
||
return rows[start:]
|
||
|
||
|
||
def preview_raw(data: bytes, template: ImportTemplate, limit: int = 8) -> dict:
|
||
"""Rohe erste Zeilen (zum Spalten-Zuordnen im Template-Editor)."""
|
||
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
|
||
return {"header": header, "rows": rows[body_start: body_start + limit]}
|
||
|
||
|
||
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],
|
||
) -> 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
|
||
if allowed_kind is not None and cat_kind.get(rule.category_id) != allowed_kind:
|
||
continue
|
||
if _pattern_hit(rule, p, z):
|
||
return rule
|
||
return None
|
||
|
||
|
||
def _guess_target_account(text_idents: list[tuple[str, int]], payee: str, purpose: str, source_id: int) -> int | None:
|
||
"""Zielkonto eines Transfers aus dem Buchungstext raten: längster passender
|
||
Erkennungs-Text je Konto gewinnt."""
|
||
hay = f"{normalize_text(payee)} {normalize_text(purpose)}"
|
||
best: tuple[int, int] | None = None # (Länge, Konto-ID)
|
||
for val, acc_id in text_idents:
|
||
if val and acc_id != source_id and val in hay and (best is None or len(val) > best[0]):
|
||
best = (len(val), acc_id)
|
||
return best[1] if best else 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).
|
||
COUNTERPART_DAYS = 5
|
||
|
||
|
||
def _is_counterpart(transfer_index: dict, account_id: int, other_id: int, d: date, amount: int) -> bool:
|
||
"""True, wenn schon ein Transfer zwischen denselben Konten existiert, dessen
|
||
Betrag (Soll- oder Haben-Seite) zu dieser Buchung passt – dann ist es die
|
||
bereits gebuchte Gegenseite und darf nicht erneut zählen."""
|
||
absamt = abs(amount)
|
||
for t_date, amt, to_amt in transfer_index.get(frozenset((account_id, other_id)), []):
|
||
if abs((t_date - d).days) <= COUNTERPART_DAYS and absamt in (amt, to_amt):
|
||
return True
|
||
return False
|
||
|
||
|
||
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] = []
|
||
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 – eine Gebühr
|
||
# ist immer ein Kostenposten, unabhängig vom Vorzeichen in der Datei.
|
||
fee = parse_amount(cell("fee"), template.decimal_sep)
|
||
if amount is not None and fee:
|
||
amount -= abs(fee)
|
||
|
||
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": parse_amount(cell("balance"), template.decimal_sep),
|
||
})
|
||
return out
|
||
|
||
|
||
def detect_rows(db: Session, user: User, account: Account, candidates: list[dict]) -> list[dict]:
|
||
"""Reichert Kandidaten an: Typ, Auto-Kategorie, interner Transfer, Duplikat.
|
||
|
||
Wird von CSV- UND Beleg-Import genutzt (gleiche Prüf-/Buchungslogik)."""
|
||
idents: dict[str, int] = {} # IBAN/Nummer -> Konto (für Transfer via IBAN)
|
||
text_idents: list[tuple[str, int]] = [] # freie Erkennungs-Texte -> Konto
|
||
for value_norm, acc_id, kind in (
|
||
db.query(AccountIdentifier.value_norm, AccountIdentifier.account_id, AccountIdentifier.kind)
|
||
.join(Account, Account.id == AccountIdentifier.account_id)
|
||
.filter(Account.owner_id == user.id)
|
||
):
|
||
if kind == IdentifierKind.text:
|
||
text_idents.append((value_norm, acc_id))
|
||
else:
|
||
idents[value_norm] = acc_id
|
||
|
||
rules = db.query(CategorizationRule).filter(CategorizationRule.owner_id == user.id).all()
|
||
rules.sort(key=_rule_sort_key, reverse=True) # längstes/manuell zuerst
|
||
cat_kind: dict[int, CategoryKind] = {
|
||
cid: kind
|
||
for cid, kind in db.query(Category.id, Category.type).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)
|
||
)
|
||
)
|
||
# Bestehende Transfers je Konto-Paar: erkennt die Gegenbuchung, wenn beide
|
||
# Auszüge eines Übertrags importiert werden (sonst zählt die Bewegung doppelt).
|
||
transfer_index: dict[frozenset, list[tuple]] = {}
|
||
for t_acc, t_to, t_date, t_amt, t_to_amt in (
|
||
db.query(Transaction.account_id, Transaction.to_account_id, Transaction.date,
|
||
Transaction.amount_cents, Transaction.to_amount_cents)
|
||
.filter(Transaction.owner_id == user.id, Transaction.type == TxnType.transfer,
|
||
Transaction.to_account_id.isnot(None))
|
||
):
|
||
transfer_index.setdefault(frozenset((t_acc, t_to)), []).append(
|
||
(t_date, t_amt, t_to_amt if t_to_amt is not None else t_amt)
|
||
)
|
||
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
|
||
)
|
||
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
|
||
|
||
other = idents.get(iban) if iban else None
|
||
if other is None:
|
||
# Symmetrisch: enthält der Text den Erkennungs-Text eines ANDEREN
|
||
# Kontos, ist es ein Transfer von/zu diesem Konto (auch Eingänge).
|
||
other = _guess_target_account(text_idents, payee or "", purpose or "", account.id)
|
||
if other is not None and other != account.id:
|
||
ttype = "transfer"
|
||
to_account_id = other
|
||
else:
|
||
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
|
||
if rule.is_transfer:
|
||
ttype = "transfer"
|
||
# Festes Zielkonto der Regel bevorzugen, sonst aus dem Text raten.
|
||
if rule.target_account_id and rule.target_account_id != account.id:
|
||
to_account_id = rule.target_account_id
|
||
else:
|
||
to_account_id = _guess_target_account(text_idents, payee or "", purpose or "", account.id)
|
||
else:
|
||
category_id = rule.category_id
|
||
|
||
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
|
||
|
||
# Gegenbuchung eines bereits gebuchten Transfers? Dann nicht erneut buchen.
|
||
is_counterpart = (
|
||
ttype == "transfer" and to_account_id is not None
|
||
and d is not None and amount is not None
|
||
and _is_counterpart(transfer_index, account.id, to_account_id, d, amount)
|
||
)
|
||
|
||
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,
|
||
"is_counterpart": is_counterpart,
|
||
"include": valid and not is_dup and not is_counterpart,
|
||
})
|
||
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 tatsächlich ins Konto fließt – 0 für nicht gebuchte Zeilen
|
||
(storniert, ungültig). Nur so fällt auf, wenn die Bank etwas verbucht hat,
|
||
wir aber nichts (z. B. nicht zugeordnete Gebührenspalte)."""
|
||
return (r.get("amount_cents") or 0) if r.get("valid") else 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 commit_import(db: Session, user: User, account: Account, template_id: int | None, filename: str, rows: list[dict]) -> 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,
|
||
filename=(filename or "")[:255], row_count=len(rows),
|
||
imported_count=0, status=ImportStatus.committed,
|
||
)
|
||
db.add(batch)
|
||
db.flush()
|
||
|
||
imported = 0
|
||
for row in rows:
|
||
if not row.get("include", True):
|
||
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()
|
||
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
|