certain/backend/app/services/bank_sync.py

"""Bankkonten verbinden und Umsätze abholen.

Der Weg in drei Schritten, genau wie ihn die Schnittstelle vorgibt:

    1. ``starte_verbindung``  → Adresse, zu der der Nutzer geschickt wird
    2. ``schliesse_verbindung`` → der Rückleitungs-Code wird zur Sitzung,
       und wir erfahren, welche Konten es dort gibt
    3. ``hole_umsaetze``      → Umsätze je zugeordnetem Konto

Abgeholte Umsätze werden NICHT gebucht. Sie landen als Vorschlag im Eingang,
dieselbe Strecke wie Mails und Kurzbefehl-Erfassungen – und laufen dabei durch
``detect_rows``, sodass Duplikatschutz und Schlagwortregeln ohne
Sonderbehandlung greifen. Eine dritte Quelle, kein dritter Sonderweg.
"""

from __future__ import annotations

import logging
import secrets
from datetime import date, datetime, timedelta, timezone

from sqlalchemy.orm import Session

from ..models import (
    Account,
    BankAccountLink,
    BankApp,
    BankConnection,
    InboxItem,
    InboxSource,
    InboxStatus,
    User,
)
from . import csv_import, geheimnis
from .bank_api import ERSTER_ABRUF_TAGE, BankClient, BankFehler

log = logging.getLogger(__name__)

# Wie lange die Zustimmung gelten soll. Die Bank kürzt selbst, wenn sie weniger
# erlaubt; mehr als 180 Tage lässt die Regelung nicht zu.
ZUSTIMMUNG_TAGE = 180
# Ab wann gewarnt wird, dass neu autorisiert werden muss. Läuft die Zustimmung
# ab, hört der Abruf sonst stumm auf – dieselbe Falle wie „0 neu" ohne Grund.
WARNUNG_TAGE = 14


def _client(db: Session, user: User) -> BankClient:
    app = db.query(BankApp).filter(BankApp.owner_id == user.id).first()
    if app is None:
        raise BankFehler(
            "Es ist noch keine Bank-Anwendung hinterlegt. Trag Application ID "
            "und Schlüsseldatei in den Einstellungen ein."
        )
    schluessel = geheimnis.entschluesseln(app.private_key_enc)
    if schluessel is None:
        raise BankFehler(
            "Der gespeicherte Schlüssel lässt sich nicht mehr entschlüsseln. "
            "Wurde das Server-Geheimnis getauscht? Dann bitte neu hinterlegen."
        )
    return BankClient(app.application_id, schluessel)


# Nicht die ganze Liste ins Formular kippen – über alle Länder sind es
# tausende. Wer sucht, bekommt Treffer; wer ein Land wählt, dessen Banken.
MAX_TREFFER = 200


def banken(db: Session, user: User, land: str | None = None,
           suche: str | None = None) -> dict:
    """Banken suchen – in einem Land oder über alle hinweg.

    Die Suche über alle Länder ist kein Luxus: Das Sitzland einer Bank steht
    nicht in der IBAN. Revolut gibt deutsche IBANs aus, ist aber in Litauen
    ansässig – wer unter „Deutschland" sucht, findet es dort nie.
    """
    roh = _client(db, user).banken(land)
    begriff = (suche or "").strip().lower()
    treffer = []
    for b in roh:
        name = b.get("name")
        if not name:
            continue
        # Auch im BIC und im Gruppennamen suchen: Banken heissen in der Liste
        # nicht immer so wie im Alltag.
        heu = " ".join(str(x) for x in (name, b.get("bic"), b.get("group")) if x).lower()
        if begriff and begriff not in heu:
            continue
        treffer.append({"name": name, "country": b.get("country"),
                        "logo": b.get("logo"), "psu_types": b.get("psu_types", []),
                        "beta": bool(b.get("beta"))})
    treffer.sort(key=lambda b: (b["name"].lower(), b.get("country") or ""))
    # „gesamt" mitliefern: Sonst sieht „keine Treffer" gleich aus, egal ob die
    # Liste leer kam oder der Suchbegriff nicht passte – und man sucht den
    # Fehler an der falschen Stelle.
    return {"gesamt": len(roh), "treffer": treffer[:MAX_TREFFER],
            "gekuerzt": len(treffer) > MAX_TREFFER}


def starte_verbindung(db: Session, user: User, aspsp_name: str, land: str,
                      psu_type: str = "personal") -> dict:
    """Autorisierung anstossen und die Weiterleitungs-Adresse liefern."""
    app = db.query(BankApp).filter(BankApp.owner_id == user.id).first()
    if app is None:
        raise BankFehler("Es ist noch keine Bank-Anwendung hinterlegt.")

    state = secrets.token_urlsafe(24)
    verbindung = BankConnection(
        owner_id=user.id, aspsp_name=aspsp_name, aspsp_country=land.upper(),
        pending_state=state,
    )
    db.add(verbindung)
    db.flush()

    antwort = _client(db, user).starte_autorisierung(
        aspsp_name, land, f"{app.base_url.rstrip('/')}/bank/rueckkehr",
        state, ZUSTIMMUNG_TAGE, psu_type,
    )
    db.commit()
    return {"url": antwort.get("url"), "connection_id": verbindung.id, "state": state}


def schliesse_verbindung(db: Session, user: User, state: str, code: str) -> dict:
    """Code gegen die Sitzung tauschen und die gefundenen Konten ablegen.

    Die Konten werden nur VERZEICHNET, nicht zugeordnet: Welches Bankkonto zu
    welchem Konto in Certain gehört, entscheidet der Nutzer. Raten wäre hier
    besonders teuer, weil eine falsche Zuordnung fremde Umsätze in ein Konto
    schreiben würde.
    """
    verbindung = (
        db.query(BankConnection)
        .filter(BankConnection.owner_id == user.id, BankConnection.pending_state == state)
        .first()
    )
    if verbindung is None:
        raise BankFehler("Zu dieser Rückmeldung gibt es keine offene Verbindung.")

    daten = _client(db, user).sitzung(code)
    verbindung.session_id = daten.get("session_id")
    verbindung.pending_state = None
    verbindung.last_error = None
    gueltig = (daten.get("access") or {}).get("valid_until")
    if gueltig:
        try:
            verbindung.valid_until = datetime.fromisoformat(gueltig.replace("Z", "+00:00"))
        except ValueError:
            verbindung.valid_until = None

    neu = 0
    for konto in daten.get("accounts", []):
        uid = konto.get("uid")
        if not uid:
            continue
        vorhanden = (
            db.query(BankAccountLink)
            .filter(BankAccountLink.owner_id == user.id, BankAccountLink.remote_uid == uid)
            .first()
        )
        if vorhanden is not None:
            vorhanden.connection_id = verbindung.id
            continue
        db.add(BankAccountLink(
            owner_id=user.id,
            connection_id=verbindung.id,
            remote_uid=uid,
            iban=(konto.get("account_id") or {}).get("iban"),
            remote_name=konto.get("name"),
            currency=(konto.get("currency") or "EUR").upper(),
        ))
        neu += 1

    db.commit()
    return {"connection_id": verbindung.id, "konten": neu,
            "valid_until": verbindung.valid_until.isoformat() if verbindung.valid_until else None}


def _betrag_cents(umsatz: dict) -> int | None:
    """Vorzeichenbehafteter Betrag in Cent.

    PSD2 führt den Betrag ohne Vorzeichen und die Richtung getrennt in
    ``credit_debit_indicator`` (CRDT/DBIT). Manche Banken liefern trotzdem ein
    Vorzeichen mit – beides wird berücksichtigt, damit eine Ausgabe nicht als
    Einnahme ankommt.
    """
    betrag = (umsatz.get("transaction_amount") or {}).get("amount")
    if betrag is None:
        return None
    cents = csv_import.parse_amount(str(betrag), ".")
    if cents is None:
        return None
    richtung = (umsatz.get("credit_debit_indicator") or "").upper()
    if richtung == "DBIT":
        return -abs(cents)
    if richtung == "CRDT":
        return abs(cents)
    return cents


def _text(umsatz: dict, ausgabe: bool) -> tuple[str | None, str | None]:
    """Empfänger und Verwendungszweck. Bei einer Ausgabe zählt der Empfänger,
    bei einer Einnahme der Absender."""
    gegen = umsatz.get("creditor") if ausgabe else umsatz.get("debtor")
    name = (gegen or {}).get("name")
    zweck = umsatz.get("remittance_information")
    if isinstance(zweck, list):
        zweck = " ".join(str(z) for z in zweck if z)
    return (str(name)[:255] if name else None,
            str(zweck)[:255] if zweck else None)


def hole_umsaetze(db: Session, user: User, verbindung: BankConnection) -> dict:
    """Umsätze aller zugeordneten Konten dieser Verbindung in den Eingang holen."""
    verknuepfungen = [
        v for v in db.query(BankAccountLink).filter(
            BankAccountLink.connection_id == verbindung.id,
            BankAccountLink.owner_id == user.id,
        )
        if v.account_id is not None
    ]
    if not verknuepfungen:
        return {"geholt": 0, "neu": 0, "ohne_zuordnung": True}

    client = _client(db, user)
    konten = {a.id: a for a in db.query(Account).filter(Account.owner_id == user.id)}
    seit = (verbindung.last_sync_at.date() - timedelta(days=3)
            if verbindung.last_sync_at
            else date.today() - timedelta(days=ERSTER_ABRUF_TAGE))

    geholt = neu = 0
    for verknuepfung in verknuepfungen:
        konto = konten.get(verknuepfung.account_id)
        if konto is None:
            continue
        for umsatz in client.umsaetze(verknuepfung.remote_uid, seit):
            geholt += 1
            kennung = (umsatz.get("entry_reference") or umsatz.get("transaction_id")
                       or umsatz.get("uid"))
            if not kennung:
                continue          # ohne Kennung kein Schutz vor Doppelabruf
            kennung = f"{verknuepfung.remote_uid}:{kennung}"[:255]
            schon = (
                db.query(InboxItem.id)
                .filter(InboxItem.owner_id == user.id,
                        InboxItem.source == InboxSource.bank,
                        InboxItem.external_id == kennung)
                .first()
            )
            if schon is not None:
                continue

            cents = _betrag_cents(umsatz)
            tag = _datum(umsatz)
            if cents is None or cents == 0 or tag is None:
                continue
            payee, zweck = _text(umsatz, cents < 0)

            # Durch dieselbe Erkennung wie CSV und Kurzbefehl: Schlagwortregeln
            # schlagen die Kategorie vor, der Eingangs-Abgleich verhindert, dass
            # eine bereits erfasste Zahlung ein zweites Mal auftaucht.
            erkannt = csv_import.detect_rows(db, user, konto, [{
                "date": tag.isoformat(), "amount_cents": cents,
                "payee": payee, "purpose": zweck,
            }])
            zeile = erkannt[0] if erkannt else {}
            if zeile.get("is_duplicate"):
                continue

            db.add(InboxItem(
                owner_id=user.id,
                source=InboxSource.bank,
                status=InboxStatus.pending,
                account_id=konto.id,
                date=tag,
                amount_cents=cents,
                currency=(umsatz.get("transaction_amount") or {}).get("currency")
                or konto.currency,
                payee=payee,
                note=zweck,
                category_id=zeile.get("category_id"),
                external_id=kennung,
            ))
            neu += 1

    verbindung.last_sync_at = datetime.now(timezone.utc)
    verbindung.last_error = None
    db.commit()
    return {"geholt": geholt, "neu": neu, "ohne_zuordnung": False}


def _datum(umsatz: dict) -> date | None:
    for feld in ("booking_date", "value_date", "transaction_date"):
        roh = umsatz.get(feld)
        if not roh:
            continue
        try:
            return date.fromisoformat(str(roh)[:10])
        except ValueError:
            continue
    return None


def laeuft_bald_ab(verbindung: BankConnection) -> bool:
    """Läuft die Zustimmung demnächst ab?

    SQLite gibt Zeitstempel ohne Zeitzone zurück, auch wenn die Spalte als
    ``timezone=True`` deklariert ist. Ohne diese Ergänzung scheitert der
    Vergleich mit einem TypeError – und zwar erst beim Anzeigen der
    Verbindungsliste, also weit weg von der Ursache.
    """
    if verbindung.valid_until is None:
        return False
    bis = verbindung.valid_until
    if bis.tzinfo is None:
        bis = bis.replace(tzinfo=timezone.utc)
    return bis - datetime.now(timezone.utc) < timedelta(days=WARNUNG_TAGE)