"""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)
    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


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


# 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.
COUNTERPART_TOLERANCE = 0.10


def _find_counterpart(
    side_index: dict, verbraucht: set, account: Account, d: date, amount: int,
    waehrung: dict, nur_konto: int | None,
):
    """Sucht einen bereits gebuchten Transfer, dessen Gegenseite diese Zeile ist.

    Gesucht wird über die Seite des importierten Kontos – die Zeile braucht also
    selbst keine Transfer-Regel. Genau das ist der Empfangs-Auszug: Im CHF-Auszug
    entsteht der Transfer, im EUR-Auszug taucht dieselbe Bewegung als Eingang auf.
    Bei einem Umtausch stehen zwei Währungen und damit zwei Beträge gegenüber;
    dann zählt ein plausibler Betrag statt eines exakten."""
    absamt = abs(amount)
    unscharf = None
    for eintrag in side_index.get(account.id, []):
        txn_id, t_date, erwartet, other_id, richtung = eintrag
        if txn_id in verbraucht or abs((t_date - d).days) > COUNTERPART_DAYS:
            continue
        # Die Richtung muss stimmen: Wo Geld ankam, kann keine Ausgabe die
        # Gegenseite sein. Ohne diese Prüfung verschluckt eine Ausgabe über
        # denselben Betrag den erwarteten Eingang – der Saldo liefe zu hoch.
        if (amount < 0) != (richtung < 0):
            continue
        # Hat eine Regel ein Gegenkonto gesetzt, muss es dazu passen.
        if nur_konto is not None and other_id != nur_konto:
            continue
        if absamt == erwartet:
            return eintrag
        cross = waehrung.get(other_id, account.currency) != account.currency
        if cross and erwartet and abs(absamt - erwartet) <= erwartet * COUNTERPART_TOLERANCE:
            unscharf = unscharf or eintrag
    return unscharf


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


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).
    # Je Konto: welcher Betrag dort aus einem bestehenden Transfer erwartet wird.
    side_index: dict[int, list[tuple]] = {}
    for t_id, t_acc, t_to, t_date, t_amt, t_to_amt in (
        db.query(Transaction.id, 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))
    ):
        gutschrift = t_to_amt if t_to_amt is not None else t_amt
        side_index.setdefault(t_acc, []).append((t_id, t_date, t_amt, t_to, -1))
        side_index.setdefault(t_to, []).append((t_id, t_date, gutschrift, t_acc, +1))
    verbraucht: set[int] = set()
    waehrung_je_konto: dict[int, str] = dict(
        db.query(Account.id, Account.currency).filter(Account.owner_id == user.id)
    )
    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 –
        # der tatsächliche Betrag dieser Zeile korrigiert aber die Schätzung der
        # anderen Seite (beim Umtausch rechneten wir dort mit dem EZB-Kurs).
        gegenbuchung = None
        if d is not None and amount is not None and not is_dup:
            gegenbuchung = _find_counterpart(
                side_index, verbraucht, account, d, amount, waehrung_je_konto, to_account_id
            )
        if gegenbuchung is not None:
            verbraucht.add(gegenbuchung[0])
            ttype = "transfer"                      # in der Prüfansicht als Transfer zeigen
            to_account_id = to_account_id or gegenbuchung[3]
        is_counterpart = gegenbuchung is not None

        # Transfer-Regel ohne erkennbares Gegenkonto: NICHT still als Einnahme/
        # Ausgabe buchen, sondern in der Prüfansicht das Konto abfragen.
        braucht_ziel = ttype == "transfer" and to_account_id is None

        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,
            # Damit commit_import die geschätzte Gegenseite auf den echten Wert setzt.
            "counterpart_txn_id": gegenbuchung[0] if gegenbuchung else 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 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 _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]) -> 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
    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()
    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