146 lines
5.3 KiB
Python
146 lines
5.3 KiB
Python
"""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 import select, union
|
||
from sqlalchemy.orm import Session
|
||
|
||
from ..models import (
|
||
Account,
|
||
CategorizationRule,
|
||
Category,
|
||
CategoryGroup,
|
||
CategoryKind,
|
||
Transaction,
|
||
TransferLink,
|
||
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)
|
||
}
|
||
# Verknuepfte Buchungen sind erledigt – genau wie in der Arbeitsliste.
|
||
# Ohne diesen Ausschluss passen sie weiter auf die Uebertrags-Regel, werden
|
||
# abgelehnt („schon verbunden“) und stehen bei JEDEM Lauf erneut unter
|
||
# „uebersprungen“. Dort verdecken sie die wirklich offenen Buchungen und
|
||
# fuellen obendrein das Limit.
|
||
verknuepft = union(
|
||
select(TransferLink.from_txn_id).where(TransferLink.owner_id == user.id),
|
||
select(TransferLink.to_txn_id).where(TransferLink.owner_id == user.id),
|
||
)
|
||
offen = (
|
||
db.query(Transaction)
|
||
.filter(Transaction.owner_id == user.id,
|
||
Transaction.category_id.is_(None),
|
||
Transaction.type != TxnType.transfer,
|
||
Transaction.id.notin_(verknuepft))
|
||
.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)}
|