"""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" und eine Kostprobe mitliefern: Sonst sieht „keine Treffer" # gleich aus, egal ob die Liste leer kam, der Suchbegriff nicht passte – # oder die Anwendung als Sandbox angelegt wurde und lauter Test-Institute # liefert. An der Kostprobe sieht man Letzteres sofort. beispiele = sorted({str(b.get("name")) for b in roh if b.get("name")})[:8] return {"gesamt": len(roh), "treffer": treffer[:MAX_TREFFER], "gekuerzt": len(treffer) > MAX_TREFFER, "beispiele": beispiele} 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)