418 lines
18 KiB
Python
418 lines
18 KiB
Python
"""Etwas zum Kontouebertrag machen – aus dem Eingang oder nachtraeglich.
|
||
|
||
Zwei Wege fuehren hierher: ein offener Vorschlag im Eingang und eine schon
|
||
gebuchte Ausgabe oder Einnahme. Beide muessen dieselben vier Dinge einhalten,
|
||
deshalb liegen sie hier und nicht in den Routern:
|
||
|
||
1. Die Richtung kommt aus dem Vorzeichen. Ein Abgang geht von hier weg, ein
|
||
Zugang kommt vom Gegenkonto her.
|
||
2. Der Betrag der Gegenseite ist bekannt oder geschaetzt – das entscheidet
|
||
ueber Punkt 3.
|
||
3. Ist er geschaetzt und haengt das Zielkonto selbst an einer Bank, wird
|
||
nachgefragt: Die Bank meldet den Zugang von allein, und ein geschaetzter
|
||
Betrag laesst sich beim Abgleich nicht wiedererkennen.
|
||
4. Auf dem Zielkonto entsteht ein Eingangs-Eintrag, der auf DIESELBE Buchung
|
||
zeigt. Er faengt die Gegenseite ab – die schon vorliegende, indem er sie
|
||
verbraucht, und die spaeter eintreffende, weil ``detect_rows`` gegen
|
||
gebuchte Eingangs-Eintraege prueft.
|
||
|
||
Ohne Punkt 4 stuende das Geld zweimal auf dem Zielkonto: Der Duplikatschutz
|
||
vergleicht ``import_hash`` – den hat eine Uebertrags-Buchung nicht – und
|
||
Eingangs-Eintraege, und ohne diesen legte der Uebertrag auf dem Zielkonto
|
||
keinen an.
|
||
|
||
Nicht betroffen ist das Buchungsformular: Wer dort von Hand einen Uebertrag
|
||
anlegt, waehlt beide Konten bewusst und hat keine Gegenseite, die noch kommt.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from datetime import timedelta
|
||
|
||
from sqlalchemy.orm import Session
|
||
|
||
from ..models import (
|
||
Account,
|
||
BankAccountLink,
|
||
InboxItem,
|
||
InboxSource,
|
||
InboxStatus,
|
||
Transaction,
|
||
TransferLink,
|
||
TxnType,
|
||
User,
|
||
)
|
||
from . import buchung, csv_import, currency, eingang
|
||
from .csv_import import INBOX_TAGE
|
||
from .uebertraege import FX_TOLERANZ
|
||
|
||
|
||
class UebertragFehler(RuntimeError):
|
||
"""Fehler mit einem Text, den man dem Nutzer zeigen kann."""
|
||
|
||
def __init__(self, text: str, nachfragen: bool = False):
|
||
super().__init__(text)
|
||
# True heisst: Es ist kein Fehler, sondern eine Warnung, die sich mit
|
||
# ``trotzdem`` uebergehen laesst.
|
||
self.nachfragen = nachfragen
|
||
|
||
|
||
def ist_angebunden(db: Session, user: User, konto: Account) -> bool:
|
||
"""Holt Certain die Umsaetze dieses Kontos selbst von der Bank?
|
||
|
||
Entscheidend fuer einen Uebertrag: Bei einem angebundenen Gegenkonto
|
||
liefert die Bank die Gegenbuchung selbst.
|
||
"""
|
||
return db.query(BankAccountLink.id).filter(
|
||
BankAccountLink.owner_id == user.id,
|
||
BankAccountLink.account_id == konto.id,
|
||
).first() is not None
|
||
|
||
|
||
def _eingang_zu(db: Session, user: User, txn: Transaction) -> InboxItem | None:
|
||
"""Der Eingangs-Eintrag, aus dem diese Buchung entstand.
|
||
|
||
Er traegt, was die Bank ueber die Umrechnung gesagt hat. Ohne ihn bliebe
|
||
nur der EZB-Kurs – auch dann, wenn die Bank ihren eigenen genannt hat.
|
||
"""
|
||
return db.query(InboxItem).filter(
|
||
InboxItem.owner_id == user.id,
|
||
InboxItem.transaction_id == txn.id,
|
||
InboxItem.account_id == txn.account_id,
|
||
).first()
|
||
|
||
|
||
def _schon_verknuepft(db: Session, user: User, txn: Transaction) -> bool:
|
||
"""Haengt diese Buchung schon als Paar an ihrer Gegenseite?"""
|
||
return db.query(TransferLink.id).filter(
|
||
TransferLink.owner_id == user.id,
|
||
(TransferLink.from_txn_id == txn.id) | (TransferLink.to_txn_id == txn.id),
|
||
).first() is not None
|
||
|
||
|
||
def _gebuchte_gegenseite(db: Session, user: User, konto: Account, ziel: Account,
|
||
eigener_typ: TxnType, erwartet: int | None, datum,
|
||
exakt: bool) -> Transaction | None:
|
||
"""Steht auf dem Zielkonto schon eine passende Buchung?
|
||
|
||
Dann waere ein Uebertrag falsch: Er wuerde das Geld dort ein zweites Mal
|
||
bewegen. Richtig ist in dem Fall, die beiden zu VERKNUEPFEN – beide
|
||
Buchungen bleiben stehen und zaehlen nur nicht mehr in der Auswertung.
|
||
|
||
Ueber Waehrungen hinweg wird mit Spielraum verglichen: Der erwartete Betrag
|
||
ist dann gerechnet, der echte kommt von der Bank.
|
||
"""
|
||
if not erwartet:
|
||
return None
|
||
gegen_typ = TxnType.income if eigener_typ == TxnType.expense else TxnType.expense
|
||
spielraum = 0 if exakt else max(1, round(erwartet * FX_TOLERANZ))
|
||
return next(
|
||
(k for k in db.query(Transaction).filter(
|
||
Transaction.owner_id == user.id,
|
||
Transaction.account_id == ziel.id,
|
||
Transaction.type == gegen_typ,
|
||
Transaction.date >= datum - timedelta(days=INBOX_TAGE),
|
||
Transaction.date <= datum + timedelta(days=INBOX_TAGE),
|
||
) if abs(k.amount_cents - erwartet) <= spielraum),
|
||
None,
|
||
)
|
||
|
||
|
||
def _schon_als_uebertrag(db: Session, user: User, konto: Account, ziel: Account,
|
||
betrag: int | None, datum, exakt: bool) -> Transaction | None:
|
||
"""Bewegt schon ein Uebertrag dasselbe Geld zwischen diesen beiden Konten?
|
||
|
||
Beide Seiten kommen oft getrennt herein – die eine von der Bank, die andere
|
||
aus der Datei. Wer beide einzeln zum Uebertrag macht, verschiebt das Geld
|
||
zweimal. Die Richtung ist dabei egal: Ein Uebertrag hat kein Vorzeichen,
|
||
nur ein Von und ein Nach.
|
||
"""
|
||
if not betrag:
|
||
return None
|
||
spielraum = 0 if exakt else max(1, round(betrag * FX_TOLERANZ))
|
||
paar = {konto.id, ziel.id}
|
||
return next(
|
||
(t for t in db.query(Transaction).filter(
|
||
Transaction.owner_id == user.id,
|
||
Transaction.type == TxnType.transfer,
|
||
Transaction.date >= datum - timedelta(days=INBOX_TAGE),
|
||
Transaction.date <= datum + timedelta(days=INBOX_TAGE),
|
||
) if {t.account_id, t.to_account_id} == paar
|
||
and min(abs(t.amount_cents - betrag),
|
||
abs((t.to_amount_cents or t.amount_cents) - betrag)) <= spielraum),
|
||
None,
|
||
)
|
||
|
||
|
||
def _pruefe_gegenseite(db: Session, user: User, konto: Account, ziel: Account,
|
||
eigener_typ: TxnType, erwartet: int | None, datum,
|
||
exakt: bool, trotzdem: bool) -> None:
|
||
if trotzdem:
|
||
return
|
||
|
||
# Zwischen denselben beiden Konten steht schon ein Uebertrag ueber
|
||
# denselben Betrag. Meist ist das dieselbe Bewegung, zweimal hereingekommen
|
||
# – einmal von der Bank, einmal aus der Datei –, und ein zweiter Uebertrag
|
||
# verschoebe das Geld noch einmal.
|
||
#
|
||
# Aber eben nur meist: Revolut-Aufrundungen sind reihenweise CHF 1.00
|
||
# zwischen demselben Paar in denselben Tagen, und die sind alle echt.
|
||
# Deshalb fragen statt verbieten – wer es weiss, bucht trotzdem.
|
||
doppelt = _schon_als_uebertrag(db, user, konto, ziel, erwartet, datum, exakt)
|
||
if doppelt is not None:
|
||
raise UebertragFehler(
|
||
f"Zwischen '{konto.name}' und '{ziel.name}' steht für den "
|
||
f"{doppelt.date.strftime('%d.%m.%Y')} schon ein Übertrag über "
|
||
f"denselben Betrag. Ist das dieselbe Bewegung, wäre sie damit "
|
||
f"doppelt erfasst – das Quellkonto zweimal belastet, das Ziel "
|
||
f"zweimal gutgeschrieben. Waren es wirklich zwei Zahlungen, buch "
|
||
f"trotzdem.",
|
||
nachfragen=True,
|
||
)
|
||
schon = _gebuchte_gegenseite(db, user, konto, ziel, eigener_typ, erwartet,
|
||
datum, exakt)
|
||
if schon is None:
|
||
return
|
||
# Haengt die gefundene Buchung schon an einer anderen, ist sie NICHT die
|
||
# Gegenseite dieser hier. Der Rat „verbinden" ginge dann ins Leere: Die
|
||
# Vorschlagsliste laesst vergebene Buchungen aus, es kaeme nie ein
|
||
# Vorschlag – und der Nutzer sucht etwas, das es nicht gibt.
|
||
if _schon_verknuepft(db, user, schon):
|
||
raise UebertragFehler(
|
||
f"Auf '{ziel.name}' steht zwar eine passende Buchung vom "
|
||
f"{schon.date.strftime('%d.%m.%Y')}, die ist aber schon mit einer "
|
||
f"anderen Buchung verbunden – also nicht die Gegenseite dieser "
|
||
f"hier. Verbinden geht deshalb nicht. Sieh auf '{ziel.name}' nach, "
|
||
f"ob die vorhandene Verknüpfung stimmt, oder ob der zweite Zugang "
|
||
f"dort schlicht fehlt.",
|
||
nachfragen=True,
|
||
)
|
||
raise UebertragFehler(
|
||
f"Auf '{ziel.name}' steht schon eine passende Buchung vom "
|
||
f"{schon.date.strftime('%d.%m.%Y')}. Beide Seiten sind also bereits "
|
||
f"gebucht – ein Übertrag würde das Geld dort ein zweites Mal bewegen. "
|
||
f"Richtig ist hier, die beiden zu verbinden: Sie bleiben stehen und "
|
||
f"zählen nur nicht mehr als Ausgabe und Einnahme.",
|
||
nachfragen=True,
|
||
)
|
||
|
||
|
||
def _gegenbetrag(db: Session, konto: Account, ziel: Account, betrag: int,
|
||
datum, eintrag: InboxItem | None = None,
|
||
) -> tuple[int | None, float | None, bool]:
|
||
"""Betrag auf der Zielseite, Kurs, und ob der Betrag BEKANNT ist.
|
||
|
||
Drei Stufen, absteigend nach Verlaesslichkeit:
|
||
|
||
1. Der Gegenbetrag, den die Bank selbst genannt hat – exakt.
|
||
2. Ihr Kurs. Genauer als die EZB, aber der Betrag bleibt gerechnet.
|
||
3. Der EZB-Kurs. Eine Naeherung; die Bank rechnet mit ihrem eigenen Kurs.
|
||
|
||
Nur in Stufe 1 gilt der Betrag als bekannt. Daran haengt, ob bei einem
|
||
angebundenen Zielkonto nachgefragt wird: Ein gerechneter Betrag laesst sich
|
||
beim naechsten Abruf nicht als dieselbe Bewegung wiedererkennen.
|
||
"""
|
||
if ziel.currency == konto.currency:
|
||
return betrag, None, True
|
||
|
||
if (eintrag is not None and eintrag.counter_amount_cents
|
||
and (eintrag.counter_currency or "").upper() == ziel.currency.upper()):
|
||
return abs(eintrag.counter_amount_cents), None, True
|
||
|
||
kurs = float(eintrag.fx_rate) if eintrag is not None and eintrag.fx_rate else None
|
||
umgerechnet = currency.convert(db, betrag, konto.currency, ziel.currency, datum, kurs)
|
||
return umgerechnet, kurs, False
|
||
|
||
|
||
def _pruefe(db: Session, user: User, konto: Account, ziel: Account,
|
||
exakt: bool, trotzdem: bool) -> None:
|
||
if ziel.id == konto.id:
|
||
raise UebertragFehler("Ein Übertrag braucht ein anderes Zielkonto")
|
||
if exakt or trotzdem:
|
||
return
|
||
if not ist_angebunden(db, user, ziel):
|
||
return
|
||
raise UebertragFehler(
|
||
f"Der Betrag auf '{ziel.name}' ist eine Umrechnung über den EZB-Kurs, "
|
||
f"nicht der Betrag, den die Bank meldet. Da das Konto selbst angebunden "
|
||
f"ist, kommt der Zugang von dort noch einmal – und liesse sich mit einem "
|
||
f"geschätzten Betrag nicht als derselbe erkennen. Sicherer: hier als "
|
||
f"Ausgabe buchen und beide Seiten danach verbinden lassen.",
|
||
nachfragen=True,
|
||
)
|
||
|
||
|
||
def _lege_an(db: Session, user: User, konto: Account, ziel: Account,
|
||
betrag_signiert: int, ziel_betrag: int | None, datum,
|
||
payee: str | None, beschreibung: str | None,
|
||
kurs: float | None = None) -> Transaction:
|
||
"""Die Uebertrags-Buchung, richtig herum.
|
||
|
||
Bei einem Zugang kennen wir den Betrag auf DIESEM Konto exakt – der gehoert
|
||
dann auf die Zielseite der Buchung, und der Abgang wird umgerechnet.
|
||
"""
|
||
betrag = abs(betrag_signiert)
|
||
if betrag_signiert < 0:
|
||
von, nach = konto, ziel
|
||
von_betrag, nach_betrag = betrag, ziel_betrag
|
||
else:
|
||
von, nach = ziel, konto
|
||
von_betrag, nach_betrag = (ziel_betrag if ziel_betrag is not None else betrag), betrag
|
||
|
||
txn = Transaction(
|
||
owner_id=user.id,
|
||
type=TxnType.transfer,
|
||
date=datum,
|
||
account_id=von.id,
|
||
to_account_id=nach.id,
|
||
amount_cents=von_betrag,
|
||
to_amount_cents=nach_betrag,
|
||
fx_rate=kurs,
|
||
payee=payee,
|
||
description=beschreibung,
|
||
)
|
||
buchung.setze_betraege(db, txn, von, nach)
|
||
db.add(txn)
|
||
db.flush()
|
||
return txn
|
||
|
||
|
||
def _gegenseite(db: Session, user: User, txn: Transaction, ziel: Account,
|
||
betrag_signiert: int, ziel_betrag: int, datum) -> InboxItem:
|
||
"""Den Eingangs-Eintrag auf dem Zielkonto – verbrauchen oder anlegen.
|
||
|
||
Verbraucht wird ein noch OFFENER Eintrag: Er beschreibt dieselbe Bewegung,
|
||
nur von der anderen Seite gesehen, und wuerde sonst ein zweites Mal gebucht.
|
||
"""
|
||
# Auf dem Zielkonto ist die Bewegung anders herum.
|
||
gesucht = -1 if betrag_signiert > 0 else 1
|
||
gesucht *= ziel_betrag
|
||
|
||
offen = [
|
||
(eid, edatum, ebetrag)
|
||
for eid, edatum, ebetrag in db.query(
|
||
InboxItem.id, InboxItem.date, InboxItem.amount_cents
|
||
).filter(
|
||
InboxItem.owner_id == user.id,
|
||
InboxItem.account_id == ziel.id,
|
||
InboxItem.status == InboxStatus.pending,
|
||
)
|
||
]
|
||
# Derselbe Abgleich wie beim Import – nicht eine zweite Vergleichslogik.
|
||
treffer = csv_import.finde_eingangs_treffer(offen, set(), datum, gesucht)
|
||
|
||
eintrag = db.get(InboxItem, treffer) if treffer is not None else None
|
||
if eintrag is None:
|
||
eintrag = InboxItem(
|
||
owner_id=user.id,
|
||
source=InboxSource.transfer,
|
||
account_id=ziel.id,
|
||
date=datum,
|
||
amount_cents=gesucht,
|
||
currency=ziel.currency,
|
||
payee=txn.payee,
|
||
note=txn.description,
|
||
)
|
||
db.add(eintrag)
|
||
eintrag.status = InboxStatus.booked
|
||
eintrag.transaction_id = txn.id
|
||
# Ein Uebertrag traegt keine Kategorie – auch nicht auf der Gegenseite.
|
||
eintrag.category_id = None
|
||
db.flush()
|
||
return eintrag
|
||
|
||
|
||
def aus_eingang(db: Session, user: User, eintrag: InboxItem, ziel: Account,
|
||
trotzdem: bool = False) -> Transaction:
|
||
"""Einen offenen Vorschlag als Kontouebertrag buchen."""
|
||
if eintrag.status == InboxStatus.booked or eintrag.transaction_id is not None:
|
||
raise UebertragFehler("Dieser Eintrag ist schon gebucht")
|
||
if eintrag.account_id is None:
|
||
raise UebertragFehler("Kein Konto zugeordnet")
|
||
konto = db.get(Account, eintrag.account_id)
|
||
if konto is None or konto.owner_id != user.id:
|
||
raise UebertragFehler("Kein Konto zugeordnet")
|
||
|
||
betrag = abs(eintrag.amount_cents)
|
||
ziel_betrag, kurs, exakt = _gegenbetrag(db, konto, ziel, betrag, eintrag.date, eintrag)
|
||
_pruefe(db, user, konto, ziel, exakt, trotzdem)
|
||
_pruefe_gegenseite(
|
||
db, user, konto, ziel,
|
||
TxnType.expense if eintrag.amount_cents < 0 else TxnType.income,
|
||
ziel_betrag, eintrag.date, exakt, trotzdem)
|
||
|
||
# Auch als Uebertrag gebucht bleibt es DIESELBE Zahlung der Bank. Ohne die
|
||
# Kennung an der Buchung waere sie beim naechsten Abruf wieder neu.
|
||
kennung = eingang.quell_kennung(eintrag.source, eintrag.external_id)
|
||
if kennung is not None and eingang.gebuchte_mit_kennung(db, user, [kennung]):
|
||
raise UebertragFehler("Diese Zahlung ist schon gebucht")
|
||
|
||
txn = _lege_an(db, user, konto, ziel, eintrag.amount_cents, ziel_betrag,
|
||
eintrag.date, eintrag.payee, eintrag.note, kurs)
|
||
txn.source_uid = kennung
|
||
eintrag.transaction_id = txn.id
|
||
eintrag.status = InboxStatus.booked
|
||
eintrag.category_id = None
|
||
_gegenseite(db, user, txn, ziel, eintrag.amount_cents,
|
||
ziel_betrag if ziel_betrag is not None else betrag, eintrag.date)
|
||
return txn
|
||
|
||
|
||
def aus_buchung(db: Session, user: User, txn: Transaction, ziel: Account,
|
||
trotzdem: bool = False) -> Transaction:
|
||
"""Eine schon gebuchte Ausgabe oder Einnahme zum Uebertrag machen.
|
||
|
||
Nicht ueber den PATCH-Weg des Buchungs-Routers: Der dreht die Richtung bei
|
||
einer Einnahme nicht um und legt keine Gegenseite an.
|
||
"""
|
||
if txn.type == TxnType.transfer:
|
||
raise UebertragFehler("Diese Buchung ist bereits ein Übertrag")
|
||
konto = db.get(Account, txn.account_id)
|
||
if konto is None or konto.owner_id != user.id:
|
||
raise UebertragFehler("Buchung ohne Konto")
|
||
|
||
# Ist sie schon mit ihrer Gegenseite verknuepft, sind BEIDE gebucht. Ein
|
||
# Uebertrag wuerde das Geld auf dem Zielkonto ein zweites Mal bewegen –
|
||
# und das laesst sich nicht mit "trotzdem" wegdruecken, es ist einfach
|
||
# falsch.
|
||
if _schon_verknuepft(db, user, txn):
|
||
raise UebertragFehler(
|
||
"Diese Buchung ist bereits mit ihrer Gegenseite verbunden – beide "
|
||
"Seiten sind gebucht und zählen schon jetzt in keiner Auswertung. "
|
||
"Ein Übertrag würde das Geld auf dem Zielkonto ein zweites Mal "
|
||
"bewegen. Wenn du es wirklich anders willst, löse zuerst die "
|
||
"Verknüpfung."
|
||
)
|
||
|
||
signiert = txn.amount_cents if txn.type == TxnType.income else -txn.amount_cents
|
||
# Der Eingangs-Eintrag traegt den Kurs der Bank – ohne ihn bliebe hier nur
|
||
# die EZB-Naeherung, auch wenn die Bank ihren eigenen genannt hat.
|
||
eintrag = _eingang_zu(db, user, txn)
|
||
ziel_betrag, kurs, exakt = _gegenbetrag(db, konto, ziel, abs(signiert),
|
||
txn.date, eintrag)
|
||
_pruefe(db, user, konto, ziel, exakt, trotzdem)
|
||
_pruefe_gegenseite(db, user, konto, ziel, txn.type, ziel_betrag, txn.date,
|
||
exakt, trotzdem)
|
||
|
||
# Die vorhandene Buchung umschreiben statt eine zweite anzulegen: Alles, was
|
||
# auf sie zeigt – Eingangs-Eintrag, Verknuepfungen – bleibt damit gueltig.
|
||
if signiert < 0:
|
||
von, nach = konto, ziel
|
||
txn.amount_cents, txn.to_amount_cents = abs(signiert), ziel_betrag
|
||
else:
|
||
von, nach = ziel, konto
|
||
txn.amount_cents = ziel_betrag if ziel_betrag is not None else abs(signiert)
|
||
txn.to_amount_cents = abs(signiert)
|
||
txn.type = TxnType.transfer
|
||
txn.account_id = von.id
|
||
txn.to_account_id = nach.id
|
||
txn.category_id = None
|
||
if kurs:
|
||
txn.fx_rate = kurs # der Kurs der Bank schlaegt die EZB
|
||
buchung.setze_betraege(db, txn, von, nach)
|
||
db.flush()
|
||
|
||
_gegenseite(db, user, txn, ziel, signiert,
|
||
ziel_betrag if ziel_betrag is not None else abs(signiert), txn.date)
|
||
return txn
|