certain/backend/app/services/eingang.py

"""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 abo_erkennung, uebertraege

    verknuepft = 0
    try:
        verknuepft = uebertraege.finde(db, user).get("verknuepft", 0)
    except Exception:                      # pragma: no cover - Sicherheitsnetz
        log.exception("Uebertragssuche nach dem Buchen fehlgeschlagen")
        db.rollback()

    # Danach nach Vertraegen sehen. Ein Abo, das ueber die Bank hereinkommt,
    # soll sich von selbst seinem Vertrag zuordnen – sonst muesste man nach
    # jedem Abruf von Hand nachsehen.
    try:
        abo_erkennung.finde(db, user)
    except Exception:                      # pragma: no cover - Sicherheitsnetz
        log.exception("Vertrags-Zuordnung nach dem Buchen fehlgeschlagen")
        db.rollback()
    return verknuepft