ScarriffleandClaude Opus 5
991a8e4fc2
Bisher lag sie nur am Eingangs-Eintrag. Das reicht nicht: Der Eintrag ist eine Zwischenstation und faellt weg, wenn ein Konto geloescht wird oder ein aelteres Backup ihn nicht kannte. Dann war die Zahlung fuer den naechsten Abruf neu, obwohl sie laengst gebucht war. Am schlimmsten bei einem als Uebertrag gebuchten Umsatz: Der ist von der Aehnlichkeitspruefung ausgenommen, dort griff also gar nichts mehr. Die Kennung gehoert deshalb an die Buchung (transactions.source_uid, mit der Quelle davor). Dort ueberlebt sie alles, was mit dem Eingang passiert, und die Eindeutigkeit haelt die Datenbank selbst durch – nicht erst der pruefende Code davor. Gestempelt wird an allen drei Buchungswegen: Eingang, Kurzbefehl und Uebertrag. Vorhandene Buchungen bekommen die Kennung nachgetragen, solange ihr Eingangs-Eintrag noch da ist. Ohne Kennung – von Hand erfasst, aus einer CSV-Datei – bleibt es beim bisherigen Schutz ueber die Aehnlichkeit. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
122 lines
4.2 KiB
Python
122 lines
4.2 KiB
Python
"""Aus einem Eingangs-Vorschlag eine Buchung machen.
|
||
|
||
Als Dienst und nicht im Router, weil es zwei Aufrufer gibt: die Oberflaeche
|
||
(„Buchen") und die Bankanbindung, die ihre Umsaetze gleich selbst bucht. Zwei
|
||
Kopien derselben Umrechnung waeren genau die Stelle, an der Salden auseinander
|
||
laufen.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
|
||
from sqlalchemy.orm import Session
|
||
|
||
from ..models import (
|
||
Account,
|
||
InboxItem,
|
||
InboxSource,
|
||
InboxStatus,
|
||
Transaction,
|
||
TxnType,
|
||
User,
|
||
)
|
||
from . import currency
|
||
|
||
log = logging.getLogger(__name__)
|
||
|
||
|
||
def quell_kennung(source: InboxSource, external_id: str | None) -> str | None:
|
||
"""Die Kennung, wie sie an der Buchung steht – oder None ohne Kennung.
|
||
|
||
Der Name der Quelle gehoert davor: Eine Nachrichten-Kennung aus dem
|
||
Postfach und eine Umsatzkennung der Bank duerfen nie zufaellig gleich
|
||
lauten, sonst verschluckt die eine Zahlung die andere.
|
||
"""
|
||
if not external_id:
|
||
return None
|
||
return f"{source.value}:{external_id}"[:320]
|
||
|
||
|
||
def gebuchte_mit_kennung(db: Session, user: User,
|
||
kennungen: list[str | None]) -> Transaction | None:
|
||
"""Die Buchung zu einer dieser Kennungen, falls es sie schon gibt.
|
||
|
||
Der Weg an ``inbox_items`` vorbei ist der Punkt: Der Eingangs-Eintrag ist
|
||
eine Zwischenstation und kann fehlen, die Buchung bleibt.
|
||
"""
|
||
werte = [k for k in kennungen if k]
|
||
if not werte:
|
||
return None
|
||
return (
|
||
db.query(Transaction)
|
||
.filter(Transaction.owner_id == user.id, Transaction.source_uid.in_(werte))
|
||
.first()
|
||
)
|
||
|
||
|
||
def buche(db: Session, user: User, eintrag: InboxItem) -> Transaction | None:
|
||
"""Aus einem Vorschlag eine Buchung machen. Ohne Konto geht es nicht."""
|
||
if eintrag.status == InboxStatus.booked or eintrag.transaction_id is not None:
|
||
return None
|
||
if eintrag.account_id is None:
|
||
return None
|
||
konto = db.get(Account, eintrag.account_id)
|
||
if konto is None or konto.owner_id != user.id:
|
||
return None
|
||
|
||
kennung = quell_kennung(eintrag.source, eintrag.external_id)
|
||
if kennung is not None:
|
||
vorhanden = gebuchte_mit_kennung(db, user, [kennung])
|
||
if vorhanden is not None:
|
||
# Diese Zahlung ist schon gebucht – unter einem frueheren
|
||
# Eingangs-Eintrag, den es nicht mehr gibt. Der Eintrag wird
|
||
# trotzdem auf die vorhandene Buchung gezeigt und geschlossen:
|
||
# Bliebe er offen, luede er beim naechsten Ansehen zum zweiten
|
||
# Buchen ein, und genau das soll die Kennung verhindern.
|
||
eintrag.transaction_id = vorhanden.id
|
||
eintrag.status = InboxStatus.booked
|
||
log.info("Eingang %s zeigt auf die vorhandene Buchung %s (%s)",
|
||
eintrag.id, vorhanden.id, kennung)
|
||
return None
|
||
|
||
betrag = abs(eintrag.amount_cents)
|
||
txn = Transaction(
|
||
owner_id=user.id,
|
||
type=TxnType.expense if eintrag.amount_cents < 0 else TxnType.income,
|
||
date=eintrag.date,
|
||
account_id=konto.id,
|
||
amount_cents=betrag,
|
||
amount_eur_cents=currency.compute_amount_eur_cents(
|
||
db, betrag, konto.currency, eintrag.date),
|
||
category_id=eintrag.category_id,
|
||
payee=eintrag.payee,
|
||
description=eintrag.note,
|
||
source_uid=kennung,
|
||
)
|
||
db.add(txn)
|
||
db.flush()
|
||
eintrag.transaction_id = txn.id
|
||
eintrag.status = InboxStatus.booked
|
||
return txn
|
||
|
||
|
||
def suche_uebertraege(db: Session, user: User) -> int:
|
||
"""Nach dem Buchen nach Uebertraegen sehen – so wie nach einem Import.
|
||
|
||
Die Gegenseite eines Uebertrags entsteht erst beim Buchen. Ohne diesen
|
||
Aufruf wuerde sie nie gefunden, und beide Seiten zaehlten dauerhaft als
|
||
Ausgabe und Einnahme in der Auswertung mit.
|
||
|
||
Ein Fehler hier darf das Buchen nicht kippen: Die Buchung ist schon
|
||
festgeschrieben, die Verknuepfung ist nur eine Schicht darueber.
|
||
"""
|
||
from . import uebertraege
|
||
|
||
try:
|
||
return uebertraege.finde(db, user).get("verknuepft", 0)
|
||
except Exception: # pragma: no cover - Sicherheitsnetz
|
||
log.exception("Uebertragssuche nach dem Buchen fehlgeschlagen")
|
||
db.rollback()
|
||
return 0
|