GuidoandClaude Opus 5
f53efc4767
Die Liste ist nicht abgeschnitten – sie reicht von "Aachener Bank" bis "Oelands Bank" und enthaelt N und R. Damit sind beide bisherigen Vermutungen widerlegt (Sandbox, alphabetisch gekuerzt), und Raten ueber Namensvarianten bringt nichts mehr. Bleibt die Suche ohne Treffer, stehen jetzt alle Eintraege mit demselben Anfangsbuchstaben da. Damit ist auf einen Blick zu sehen, ob die Bank unter einem anderen Namen gefuehrt wird – oder ob sie in diesen 815 von ueber 2500 schlicht fehlt. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
339 lines
13 KiB
Python
339 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}),
|
||
# Alles mit demselben Anfangsbuchstaben wie der Suchbegriff. Damit
|
||
# sieht man, unter welchem Namen die gesuchte Bank dort wirklich steht
|
||
# – Raten über Namensvarianten hat lange genug gedauert.
|
||
"gleicher_buchstabe": (
|
||
[n for n in namen if begriff and n[:1].lower() == begriff[:1]][:40]
|
||
if begriff else []
|
||
),
|
||
}
|
||
|
||
|
||
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)
|