"""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 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, "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, "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: s = str(raw or "").strip() if not s: return None try: return datetime.strptime(s, fmt).date() except ValueError: 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 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) 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, }) 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) } existing = { h[0] for h in db.query(Transaction.import_hash).filter( Transaction.owner_id == user.id, Transaction.import_hash.isnot(None) ) } seen: set[str] = set() 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 valid = d is not None and amount is not None and amount != 0 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 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 = h in existing or h in seen seen.add(h) 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, "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, "include": valid and not is_dup, }) return out 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