certain/backend/app/services/regeln.py

"""Regeln nachträglich auf Buchungen ohne Kategorie anwenden.

Regeln greifen beim Import. Alles, was vorher hereinkam – oder hereinkam,
bevor es die Regel gab –, bleibt liegen. Genau das passiert bei einer
Bankanbindung dauernd: Die Umsätze kommen von selbst, die Regel entsteht
später.

Übertrags-Regeln laufen durch denselben Dienst wie jeder andere Übertrag.
Er lehnt ab, wo die Gegenseite schon gebucht ist; solche Buchungen bleiben
unangetastet und werden gemeldet. Geraten wird nichts.
"""

from __future__ import annotations

from sqlalchemy.orm import Session

from ..models import (
    Account,
    CategorizationRule,
    Category,
    CategoryGroup,
    CategoryKind,
    Transaction,
    TxnType,
    User,
)
from . import csv_import, uebertrag_buchen
from .uebertrag_buchen import UebertragFehler

# Mehr als das ist keine Nachbesserung mehr, sondern ein Umbau – und der
# gehoert nicht hinter einen Knopf.
MAX_BUCHUNGEN = 2000


def _ziel(db: Session, user: User, regel: CategorizationRule,
          txn: Transaction) -> Account | None:
    """Zielkonto einer Übertrags-Regel: fest gewählt oder aus dem Text erraten.

    Erraten wird an Wortgrenzen über die Erkennungs-Texte der Konten – dieselbe
    Prüfung wie beim Verbinden. „Euro" darf nicht „PayPal Europe" treffen.
    """
    if regel.target_account_id:
        konto = db.get(Account, regel.target_account_id)
        return konto if konto and konto.owner_id == user.id else None

    from .uebertraege import _kontobezeichner, _trifft_als_wort

    from .csv_import import normalize_text

    # Reihenfolge beachten: Gesucht wird das WORT im Text, nicht umgekehrt.
    text = normalize_text(f"{txn.payee or ''} {txn.description or ''}")
    treffer = [
        konto_id
        for konto_id, woerter in _kontobezeichner(db, user).items()
        if konto_id != txn.account_id
        and any(_trifft_als_wort(w, text) for w in woerter)
    ]
    # Nur bei GENAU einem Treffer. Zwei Möglichkeiten heisst raten.
    if len(treffer) != 1:
        return None
    return db.get(Account, treffer[0])


def wende_an(db: Session, user: User, nur_vorschau: bool = True) -> dict:
    """Regeln auf alle Buchungen ohne Kategorie anwenden.

    Mit ``nur_vorschau`` wird nichts geändert – dann steht nur da, was
    passieren würde.
    """
    regeln = sorted(
        db.query(CategorizationRule).filter(CategorizationRule.owner_id == user.id),
        key=csv_import._rule_sort_key,
        reverse=True,
    )
    if not regeln:
        return {"kategorien": 0, "uebertraege": 0, "uebersprungen": [], "offen": 0}

    art_je_kategorie = {
        cid: kind
        for cid, kind in db.query(Category.id, CategoryGroup.kind)
        .join(CategoryGroup, CategoryGroup.id == Category.group_id)
        .filter(Category.owner_id == user.id)
    }
    offen = (
        db.query(Transaction)
        .filter(Transaction.owner_id == user.id,
                Transaction.category_id.is_(None),
                Transaction.type != TxnType.transfer)
        .order_by(Transaction.date, Transaction.id)
        .limit(MAX_BUCHUNGEN)
        .all()
    )

    kategorien = uebertraege = 0
    uebersprungen: list[dict] = []
    for txn in offen:
        art = (CategoryKind.income if txn.type == TxnType.income
               else CategoryKind.expense)
        regel = csv_import._match_rule(
            regeln, txn.payee or "", txn.description or "", txn.amount_cents,
            art, art_je_kategorie)
        if regel is None:
            continue

        text = txn.payee or txn.description or "ohne Text"
        if regel.is_transfer:
            ziel = _ziel(db, user, regel, txn)
            if ziel is None:
                uebersprungen.append({
                    "text": text, "datum": txn.date.isoformat(),
                    "grund": "kein eindeutiges Zielkonto"})
                continue
            if nur_vorschau:
                uebertraege += 1
                continue
            # Je Uebertrag ein eigener Rettungspunkt: Ein einfaches rollback()
            # verwuerfe auch alles, was in diesem Durchlauf schon gelungen ist.
            try:
                with db.begin_nested():
                    uebertrag_buchen.aus_buchung(db, user, txn, ziel)
                uebertraege += 1
            except UebertragFehler as e:
                uebersprungen.append({"text": text, "datum": txn.date.isoformat(),
                                      "grund": str(e)})
        elif regel.category_id:
            if not nur_vorschau:
                txn.category_id = regel.category_id
            kategorien += 1

    if not nur_vorschau:
        db.commit()
    return {"kategorien": kategorien, "uebertraege": uebertraege,
            "uebersprungen": uebersprungen, "offen": len(offen)}