feat: Phase 4 – CSV-Import (Templates, Prüfansicht, Duplikat/Transfer-Erkennung, Undo)

Backend:
- Modelle + Migration 0005: import_templates, import_batches, categorization_rules;
  transactions um import_batch_id + import_hash erweitert.
- services/csv_import: Datei parsen (Trennzeichen, Dezimal-/Datumsformat, Kodierung,
  Header/Skip, Vorzeichen vs. Soll/Haben), Vorschau mit Duplikat-Erkennung (Hash),
  interner-Transfer-Erkennung via IBAN, lernender Auto-Kategorisierung; Buchen mit
  Batch und Undo.
- routers/imports: Templates (CRUD), Roh-Vorschau, Prüf-Vorschau, Buchen, Verlauf,
  Rückgängig. pytest: Parsing, Duplikate, Transfer, Buchen, Undo, Lernen.

Frontend:
- Import-Seite: Konto+Datei, Format/Spaltenzuordnung (mit Roh-Vorschau),
  editierbare Prüfansicht (Kategorie/Typ/Zielkonto, Duplikate markiert),
  Buchen, Import-Verlauf mit Undo; Vorlagen speichern; i18n DE/EN.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

backend/app/services/csv_import.py

@@ -0,0 +1,340 @@
+"""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,
+    ImportBatch,
+    ImportStatus,
+    ImportTemplate,
+    MatchField,
+    SAVINGS_ACCOUNT_TYPES,
+    Transaction,
+    TxnType,
+    User,
+)
+from . import currency
+
+
+# --- 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 _apply_rules(rules: list[CategorizationRule], payee: str, purpose: str) -> int | None:
+    p = normalize_text(payee)
+    z = normalize_text(purpose)
+    for rule in rules:  # bereits nach Priorität sortiert
+        hay = p if rule.match_field == MatchField.payee else z
+        if rule.pattern and rule.pattern in hay and rule.category_id:
+            return rule.category_id
+    return None
+
+
+def build_preview(db: Session, user: User, account: Account, template: ImportTemplate, data: bytes) -> list[dict]:
+    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
+
+    rows = read_rows(data, template)
+
+    # Identifikatoren aller Konten des Nutzers -> Konto-ID (für Transfer-Erkennung).
+    idents: dict[str, int] = {}
+    for value_norm, acc_id in (
+        db.query(AccountIdentifier.value_norm, AccountIdentifier.account_id)
+        .join(Account, Account.id == AccountIdentifier.account_id)
+        .filter(Account.owner_id == user.id)
+    ):
+        idents[value_norm] = acc_id
+
+    rules = (
+        db.query(CategorizationRule)
+        .filter(CategorizationRule.owner_id == user.id)
+        .order_by(CategorizationRule.priority.desc(), CategorizationRule.hits.desc())
+        .all()
+    )
+
+    existing = {
+        h[0]
+        for h in db.query(Transaction.import_hash).filter(
+            Transaction.owner_id == user.id, Transaction.import_hash.isnot(None)
+        )
+    }
+    seen_in_file: set[str] = set()
+
+    out: list[dict] = []
+    for i, r in enumerate(rows):
+        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)
+
+        payee = cell("payee")
+        purpose = cell("purpose")
+        iban = normalize_iban(cell("counterparty_iban"))
+
+        valid = d is not None and amount is not None and amount != 0
+        ttype = "transfer" if False else ("expense" if (amount or 0) < 0 else "income")
+        category_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:
+            category_id = _apply_rules(rules, payee, purpose)
+
+        h = import_hash(account.id, d, amount, payee)
+        is_dup = h in existing or h in seen_in_file
+        seen_in_file.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,
+            "to_account_id": to_account_id,
+            "import_hash": h,
+            "is_duplicate": is_dup,
+            "include": valid and not is_dup,
+        })
+    return out
+
+
+# --- 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
+    existing = (
+        db.query(CategorizationRule)
+        .filter(
+            CategorizationRule.owner_id == user.id,
+            CategorizationRule.match_field == MatchField.payee,
+            CategorizationRule.pattern == pattern,
+        )
+        .first()
+    )
+    if existing is None:
+        db.add(CategorizationRule(
+            owner_id=user.id, match_field=MatchField.payee, pattern=pattern,
+            category_id=cat, priority=1, hits=1,
+        ))
+    else:
+        existing.category_id = cat
+        existing.hits += 1
+
+
+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)
+        _learn_rule(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