feat: Schnellerfassung per Kurzbefehl (Etappe A)

Der Kontoauszug kommt erst am Monatsende; wer im Laden bezahlt hat, weiss
in dem Moment noch, wofuer. iOS laesst keine App fremde Mitteilungen
mitlesen, also fuehrt der Weg ueber einen Kurzbefehl, der die Zahlung an
Certain schickt.

Gebaut:
- Geraetemarken mit eigener Abhaengigkeit statt Sonderfall in
  get_current_user. Eine Marke gilt NUR an /quick – auf allen anderen
  Wegen ist sie keine Anmeldung. Geht das Telefon verloren, ist kein
  Datenabfluss moeglich; Tests halten das fuer Buchungen, Konten,
  Kategorien, Export und Datenbanksicherung fest.
- POST /quick/booking laeuft durch detect_rows, damit Schlagwort- und
  Betragsregeln die Kategorie schon vorschlagen. external_id verhindert,
  dass ein doppelt ausgeloester Kurzbefehl zweimal bucht.
- inbox_items merkt sich die Zahlung ueber das Buchen hinaus. Beim
  spaeteren CSV-Import greift der Import-Schluessel nicht: Der Text vom
  Telefon ist selten wortgleich mit dem des Auszugs und die Bank bucht ein
  bis zwei Tage spaeter. detect_rows vergleicht deshalb zusaetzlich
  unscharf – exakter Betrag, Datum +/- 3 Tage, je Eintrag nur einmal.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Guido
2026-08-22 19:34:14 +02:00
co-authored by Claude Opus 5
parent 9d6a53a224
commit 02260d7cd2
9 changed files with 841 additions and 1 deletions
+52
View File
@@ -14,6 +14,8 @@ from sqlalchemy.orm import Session
from ..models import (
Account,
InboxItem,
InboxStatus,
AmountMode,
BudgetTreatment,
CategorizationRule,
@@ -367,6 +369,31 @@ def parse_file_to_candidates(data: bytes, template: ImportTemplate) -> list[dict
return out
INBOX_TAGE = 3 # Bank bucht oft ein bis zwei Tage nach der Zahlung
def _find_inbox_match(
eingang: list[tuple], verbraucht: set[int], d: date, amount: int
) -> int | None:
"""Passenden, noch nicht verbrauchten Eingangs-Eintrag suchen.
Verglichen wird der Betrag exakt – bei Geld ist Toleranz gefährlich – und
das Datum mit ein paar Tagen Spielraum. Der zeitlich nächste gewinnt, damit
bei wiederkehrenden Beträgen (jeden Tag derselbe Kaffee) die richtige Zeile
getroffen wird.
"""
beste: tuple[int, int] | None = None # (Abstand in Tagen, ID)
for eid, edatum, ebetrag in eingang:
if eid in verbraucht or ebetrag != amount:
continue
abstand = abs((d - edatum).days)
if abstand > INBOX_TAGE:
continue
if beste is None or abstand < beste[0]:
beste = (abstand, eid)
return beste[1] if beste else None
def detect_rows(db: Session, user: User, account: Account, candidates: list[dict]) -> list[dict]:
"""Reichert Kandidaten an: Typ, Auto-Kategorie, Duplikat.
@@ -400,6 +427,22 @@ def detect_rows(db: Session, user: User, account: Account, candidates: list[dict
neutrale_kategorie = next(
(cid for cid, k in cat_kind.items() if k is None), None
)
# Zahlungen, die schon von aussen erfasst wurden (Kurzbefehl, später Mail).
# Der Import-Schlüssel greift hier NICHT: Der Text vom Telefon ist selten
# wortgleich mit dem des Auszugs, und das Buchungsdatum der Bank liegt oft
# ein bis zwei Tage später. Deshalb wird zusätzlich unscharf verglichen –
# gleicher Betrag, Datum in der Nähe.
eingang = [
(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 == account.id,
InboxItem.status.in_((InboxStatus.booked, InboxStatus.discarded)),
)
]
eingang_verbraucht: set[int] = set()
seen: Counter = Counter()
out: list[dict] = []
@@ -445,6 +488,14 @@ def detect_rows(db: Session, user: User, account: Account, candidates: list[dict
h = import_hash(account.id, d, amount, payee)
is_dup = seen[h] < existing[h] # nur so viele Duplikate wie schon in der DB
seen[h] += 1
dup_grund = "hash" if is_dup else None
if not is_dup and d is not None and amount:
treffer = _find_inbox_match(eingang, eingang_verbraucht, d, amount)
if treffer is not None:
eingang_verbraucht.add(treffer)
is_dup = True
dup_grund = "inbox"
is_counterpart = False
braucht_ziel = False
@@ -470,6 +521,7 @@ def detect_rows(db: Session, user: User, account: Account, candidates: list[dict
"to_account_id": to_account_id,
"import_hash": h,
"is_duplicate": is_dup,
"duplicate_reason": dup_grund,
"is_counterpart": is_counterpart,
# Damit commit_import die geschätzte Gegenseite auf den echten Wert setzt.
"counterpart_txn_id": None,