GuidoandClaude Opus 5
74641aeeac
815 geladene Banken bei ueber 2500 verfuegbaren, und alle Beispiele beginnen mit A oder B – das deutet auf eine abgeschnittene Liste, nicht auf einen falschen Suchbegriff. Die Sandbox-Vermutung ist damit widerlegt: "Aachener Bank" und "BBVA" sind echte Institute. Die Antwort nennt jetzt den Bereich von/bis und alle vorhandenen Anfangsbuchstaben. Fehlen N und R, ist die Sache entschieden und die Abfrage braucht Blaettern – dann suche ich nicht laenger am falschen Ende. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
332 lines
13 KiB
Python
332 lines
13 KiB
Python
"""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 ueber den ganzen Bereich streuen, nicht nur den Anfang zeigen.
|
||
# Endet die Liste alphabetisch bei „B", ist sie abgeschnitten – und man
|
||
# sucht sonst ewig nach einer Bank, die schlicht nicht mitgeliefert wurde.
|
||
namen = sorted({str(b.get("name")) for b in roh if b.get("name")})
|
||
beispiele = (namen[:3] + namen[len(namen) // 2: len(namen) // 2 + 2] + namen[-3:]
|
||
if len(namen) > 8 else namen)
|
||
return {
|
||
"gesamt": len(roh),
|
||
"treffer": treffer[:MAX_TREFFER],
|
||
"gekuerzt": len(treffer) > MAX_TREFFER,
|
||
"beispiele": beispiele,
|
||
"von": namen[0] if namen else None,
|
||
"bis": namen[-1] if namen else None,
|
||
# Fängt überhaupt eine Bank mit diesen Buchstaben an? Wenn nicht, fehlt
|
||
# der Teil des Alphabets, und der Suchbegriff ist unschuldig.
|
||
"anfangsbuchstaben": sorted({n[:1].upper() for n in namen}),
|
||
}
|
||
|
||
|
||
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)
|