diff --git a/backend/app/deps.py b/backend/app/deps.py index c9f5fa1..7d6fd1c 100644 --- a/backend/app/deps.py +++ b/backend/app/deps.py @@ -10,15 +10,18 @@ einrichten). Bisher wurde das nur im Frontend umgeleitet – wer die API direkt ansprach, kam daran vorbei. """ +from datetime import datetime, timezone + from fastapi import Depends, HTTPException, Request, status from sqlalchemy.orm import Session from .config import get_settings from .database import get_db -from .models import Role, User +from .models import ApiToken, Role, TokenScope, User from .schemas import UserOut from .services import policy as policy_service from .services import sessions +from .security import hash_session_token _credentials_exc = HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, @@ -90,6 +93,44 @@ def enforce_account_state(request: Request, db: Session = Depends(get_db)) -> No raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=grund) +_token_exc = HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Ungültige oder fehlende Gerätemarke", +) + + +def require_quick_token(request: Request, db: Session = Depends(get_db)) -> ApiToken: + """Prüft die Gerätemarke im Header ``X-Certain-Token``. + + Bewusst eine EIGENE Abhängigkeit statt eines Sonderfalls in + ``get_current_user``: So gilt eine Marke ausschliesslich an den Endpunkten, + die sie ausdrücklich verlangen. Auf allen übrigen Wegen ist sie schlicht + keine Anmeldung – wer sie erbeutet, kommt an keine Daten. + """ + roh = (request.headers.get("X-Certain-Token") or "").strip() + if not roh: + raise _token_exc + marke = ( + db.query(ApiToken) + .filter(ApiToken.token_hash == hash_session_token(roh), + ApiToken.scope == TokenScope.quick_add) + .first() + ) + if marke is None: + raise _token_exc + besitzer = db.get(User, marke.owner_id) + if besitzer is None or not besitzer.is_active: + raise _token_exc + # Wer sein Passwort wechseln muss, soll auch nicht per Kurzbefehl buchen. + grund = pending_reason(db, besitzer) + if grund is not None: + raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=grund) + marke.last_used_at = datetime.now(timezone.utc) + db.commit() + request.state.token_user = besitzer + return marke + + def require_admin(current_user: User = Depends(get_current_user)) -> User: if current_user.role != Role.admin: raise HTTPException( diff --git a/backend/app/main.py b/backend/app/main.py index 8ce3500..3656207 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -29,6 +29,8 @@ from .routers import ( documents, imports, policy, + inbox, + quick, rates, settings as settings_router, transactions, @@ -116,6 +118,8 @@ api.include_router(imports.router) api.include_router(documents.router) api.include_router(backup.router) api.include_router(rates.router) +api.include_router(quick.router) +api.include_router(inbox.router) app.include_router(api) diff --git a/backend/app/models.py b/backend/app/models.py index a076775..eede165 100644 --- a/backend/app/models.py +++ b/backend/app/models.py @@ -589,3 +589,98 @@ class CategorizationRule(Base): UniqueConstraint("owner_id", "match_field", "pattern", "kind", "amount_cents", name="uq_rule_owner_field_pattern_kind_amount"), ) + + +class TokenScope(str, enum.Enum): + """Was eine Gerätemarke darf. Bewusst eng gehalten. + + Eine Marke liegt auf dem Telefon, oft in einem Kurzbefehl, den man teilt. + Sie darf deshalb nur erfassen, niemals lesen: Geht das Gerät verloren, ist + kein Datenabfluss möglich. + """ + + quick_add = "quick_add" + + +class ApiToken(Base): + """Zugangsmerkmal für Geräte, die sich nicht anmelden können (Kurzbefehle).""" + + __tablename__ = "api_tokens" + + id: Mapped[int] = mapped_column(Integer, primary_key=True) + owner_id: Mapped[int] = mapped_column( + ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True + ) + name: Mapped[str] = mapped_column(String(80), nullable=False) + # Nur der Hash wird abgelegt; den Klartext sieht man einmal beim Anlegen. + token_hash: Mapped[str] = mapped_column(String(64), nullable=False, unique=True, index=True) + scope: Mapped[TokenScope] = mapped_column( + _enum(TokenScope), nullable=False, default=TokenScope.quick_add + ) + # Vorbelegtes Konto, damit der Kurzbefehl ohne Kontoauswahl auskommt. + default_account_id: Mapped[int | None] = mapped_column( + ForeignKey("accounts.id", ondelete="SET NULL"), nullable=True + ) + last_used_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now) + + +class InboxSource(str, enum.Enum): + shortcut = "shortcut" # per Kurzbefehl vom Telefon erfasst + mail = "mail" # aus einer Benachrichtigungs-Mail gelesen + + +class InboxStatus(str, enum.Enum): + booked = "booked" # als Buchung übernommen + pending = "pending" # wartet auf Bestätigung + unclear = "unclear" # Erkennung unsicher, erst ansehen + discarded = "discarded" # keine Zahlung / verworfen + + +class InboxItem(Base): + """Eine von aussen erfasste Zahlung, bevor der Kontoauszug sie bestätigt. + + Der Kontoauszug kommt erst am Monatsende. Wer eine Zahlung gleich erfasst, + weiss noch, wofür sie war – auf dem Auszug steht später oft nur ein + nichtssagender Betreibername. Der Eintrag bleibt nach dem Buchen erhalten, + damit der spätere CSV-Import ihn wiedererkennt und die Zahlung nicht ein + zweites Mal verbucht. + """ + + __tablename__ = "inbox_items" + + id: Mapped[int] = mapped_column(Integer, primary_key=True) + owner_id: Mapped[int] = mapped_column( + ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True + ) + source: Mapped[InboxSource] = mapped_column(_enum(InboxSource), nullable=False) + status: Mapped[InboxStatus] = mapped_column( + _enum(InboxStatus), nullable=False, default=InboxStatus.pending + ) + account_id: Mapped[int | None] = mapped_column( + ForeignKey("accounts.id", ondelete="CASCADE"), nullable=True, index=True + ) + date: Mapped[date_t] = mapped_column(Date, nullable=False, index=True) + amount_cents: Mapped[int] = mapped_column(Integer, nullable=False) + currency: Mapped[str] = mapped_column(String(3), nullable=False, default="EUR") + payee: Mapped[str | None] = mapped_column(String(255), nullable=True) + note: Mapped[str | None] = mapped_column(Text, nullable=True) + category_id: Mapped[int | None] = mapped_column( + ForeignKey("categories.id", ondelete="SET NULL"), nullable=True + ) + # Die erzeugte Buchung. SET NULL, damit das Löschen einer Buchung den + # Eingangs-Eintrag nicht mitreisst – sonst käme die Zahlung beim nächsten + # Auszug unbemerkt ein zweites Mal herein. + transaction_id: Mapped[int | None] = mapped_column( + ForeignKey("transactions.id", ondelete="SET NULL"), nullable=True + ) + # Kennung der Quelle (Message-ID der Mail, Kennung des Kurzbefehls): + # schützt davor, dieselbe Meldung zweimal zu erfassen. + external_id: Mapped[str | None] = mapped_column(String(255), nullable=True) + raw_text: Mapped[str | None] = mapped_column(Text, nullable=True) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now) + updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now, onupdate=_now) + + __table_args__ = ( + UniqueConstraint("owner_id", "source", "external_id", name="uq_inbox_owner_source_external"), + ) diff --git a/backend/app/routers/inbox.py b/backend/app/routers/inbox.py new file mode 100644 index 0000000..0403782 --- /dev/null +++ b/backend/app/routers/inbox.py @@ -0,0 +1,117 @@ +"""Eingang extern erfasster Zahlungen und die Gerätemarken dafür. + +Der Eingang zeigt, was per Kurzbefehl (später auch per Mail) hereinkam. Die +Einträge bleiben nach dem Buchen bestehen: Der spätere CSV-Import gleicht +gegen sie ab, damit dieselbe Zahlung nicht ein zweites Mal verbucht wird. +""" + +from fastapi import APIRouter, Depends, HTTPException, status +from sqlalchemy.orm import Session + +from ..database import get_db +from ..deps import get_current_user, get_owned +from ..models import Account, ApiToken, InboxItem, InboxStatus, Transaction, User +from ..schemas import ApiTokenCreate, ApiTokenCreated, ApiTokenOut, InboxItemOut +from ..security import generate_session_token, hash_session_token + +router = APIRouter(prefix="/inbox", tags=["inbox"]) + + +# --- Gerätemarken ------------------------------------------------------------ + + +@router.get("/tokens", response_model=list[ApiTokenOut]) +def list_tokens(db: Session = Depends(get_db), user: User = Depends(get_current_user)): + return ( + db.query(ApiToken) + .filter(ApiToken.owner_id == user.id) + .order_by(ApiToken.created_at.desc()) + .all() + ) + + +@router.post("/tokens", response_model=ApiTokenCreated, status_code=status.HTTP_201_CREATED) +def create_token( + payload: ApiTokenCreate, + db: Session = Depends(get_db), + user: User = Depends(get_current_user), +): + """Legt eine Marke an. Der Klartext ist NUR in dieser Antwort zu sehen.""" + if payload.default_account_id is not None: + get_owned(db, Account, payload.default_account_id, user) + klartext = generate_session_token() + marke = ApiToken( + owner_id=user.id, + name=payload.name.strip(), + token_hash=hash_session_token(klartext), + default_account_id=payload.default_account_id, + ) + db.add(marke) + db.commit() + db.refresh(marke) + return ApiTokenCreated(**ApiTokenOut.model_validate(marke).model_dump(), token=klartext) + + +@router.delete("/tokens/{token_id}", status_code=status.HTTP_204_NO_CONTENT) +def delete_token(token_id: int, db: Session = Depends(get_db), user: User = Depends(get_current_user)): + marke = get_owned(db, ApiToken, token_id, user) + db.delete(marke) + db.commit() + + +# --- Eingang ----------------------------------------------------------------- + + +@router.get("", response_model=list[InboxItemOut]) +def list_inbox( + limit: int = 200, + db: Session = Depends(get_db), + user: User = Depends(get_current_user), +): + return ( + db.query(InboxItem) + .filter(InboxItem.owner_id == user.id, InboxItem.status != InboxStatus.discarded) + .order_by(InboxItem.date.desc(), InboxItem.id.desc()) + .limit(min(limit, 1000)) + .all() + ) + + +@router.delete("/{item_id}", status_code=status.HTTP_204_NO_CONTENT) +def discard_item(item_id: int, db: Session = Depends(get_db), user: User = Depends(get_current_user)): + """Eintrag verwerfen – samt der Buchung, die daraus entstanden ist. + + Der Eintrag selbst bleibt als ``discarded`` stehen. Sonst wäre die Zahlung + beim nächsten Auszug wieder neu, obwohl der Nutzer sie bewusst weggeworfen + hat; so kann der Abgleich sie weiterhin erkennen. + """ + eintrag = get_owned(db, InboxItem, item_id, user) + if eintrag.transaction_id is not None: + txn = db.get(Transaction, eintrag.transaction_id) + if txn is not None and txn.owner_id == user.id: + db.delete(txn) + eintrag.status = InboxStatus.discarded + eintrag.transaction_id = None + db.commit() + + +@router.post("/{item_id}/category/{category_id}", response_model=InboxItemOut) +def set_category( + item_id: int, + category_id: int, + db: Session = Depends(get_db), + user: User = Depends(get_current_user), +): + """Kategorie setzen – auf dem Eintrag UND auf der erzeugten Buchung.""" + from ..models import Category + + eintrag = get_owned(db, InboxItem, item_id, user) + kategorie = get_owned(db, Category, category_id, user) + eintrag.category_id = kategorie.id + if eintrag.transaction_id is not None: + txn = db.get(Transaction, eintrag.transaction_id) + if txn is not None and txn.owner_id == user.id: + txn.category_id = kategorie.id + db.commit() + db.refresh(eintrag) + return eintrag diff --git a/backend/app/routers/quick.py b/backend/app/routers/quick.py new file mode 100644 index 0000000..98d40b4 --- /dev/null +++ b/backend/app/routers/quick.py @@ -0,0 +1,146 @@ +"""Schnellerfassung für Geräte ohne Anmeldung (iOS-Kurzbefehle). + +Der Kontoauszug kommt erst am Monatsende. Wer im Laden bezahlt hat, weiss in +dem Moment noch, wofür – auf dem Auszug steht später oft nur ein +nichtssagender Betreibername. Diese Endpunkte nehmen so eine Zahlung entgegen. + +Die Anmeldung läuft NICHT über die Sitzung, sondern über eine Gerätemarke +(``require_quick_token``). Deren Geltungsbereich ist eng: erfassen ja, lesen +nein. Die einzige Ausnahme ist die Kontoliste, die der Kurzbefehl für seine +Auswahl braucht – sie liefert nur Name, Symbol und Farbe, keine Salden. +""" + +from datetime import date as date_t +from decimal import ROUND_HALF_UP, Decimal + +from fastapi import APIRouter, Depends, HTTPException, status +from sqlalchemy.orm import Session + +from ..database import get_db +from ..deps import require_quick_token +from ..models import ( + Account, + ApiToken, + InboxItem, + InboxSource, + InboxStatus, + Transaction, + TxnType, + User, +) +from ..schemas import InboxItemOut, QuickBookingIn, QuickBookingOut +from ..services import csv_import, currency + +router = APIRouter(prefix="/quick", tags=["quick"]) + + +def _konto(db: Session, marke: ApiToken, gewuenscht: int | None) -> Account: + """Zielkonto bestimmen: ausdrücklich genannt, sonst das der Marke.""" + konto_id = gewuenscht or marke.default_account_id + if konto_id is None: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="Kein Konto angegeben und die Marke hat kein Standardkonto", + ) + konto = db.get(Account, konto_id) + if konto is None or konto.owner_id != marke.owner_id: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Konto nicht gefunden") + return konto + + +@router.get("/accounts") +def quick_accounts( + marke: ApiToken = Depends(require_quick_token), db: Session = Depends(get_db) +) -> list[dict]: + """Konten für die Auswahlliste im Kurzbefehl – ohne Salden.""" + konten = ( + db.query(Account) + .filter(Account.owner_id == marke.owner_id, Account.is_archived.is_(False)) + .order_by(Account.position, Account.name) + .all() + ) + return [ + {"id": a.id, "name": a.name, "currency": a.currency, "icon": a.icon, "color": a.color, + "default": a.id == marke.default_account_id} + for a in konten + ] + + +@router.post("/booking", response_model=QuickBookingOut, status_code=status.HTTP_201_CREATED) +def quick_booking( + payload: QuickBookingIn, + marke: ApiToken = Depends(require_quick_token), + db: Session = Depends(get_db), +) -> QuickBookingOut: + konto = _konto(db, marke, payload.account_id) + user = db.get(User, marke.owner_id) + tag = payload.date or date_t.today() + + betrag = int( + (payload.amount.copy_abs() * 100).quantize(Decimal("1"), rounding=ROUND_HALF_UP) + ) + if betrag == 0: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Betrag ist null") + + # Ein doppelt ausgelöster Kurzbefehl (schlechtes Netz, Doppeltipp) darf + # nicht zweimal buchen. Die bestehende Erfassung wird zurückgemeldet. + if payload.external_id: + schon = ( + db.query(InboxItem) + .filter(InboxItem.owner_id == user.id, InboxItem.source == InboxSource.shortcut, + InboxItem.external_id == payload.external_id) + .first() + ) + if schon is not None: + return QuickBookingOut( + message="Bereits erfasst", item=InboxItemOut.model_validate(schon) + ) + + # Durch die normale Erkennung schicken, damit Schlagwort- und Betragsregeln + # die Kategorie vorschlagen – dieselbe Strecke wie beim CSV-Import. + vorzeichen = -betrag if payload.kind == "expense" else betrag + erkannt = csv_import.detect_rows(db, user, konto, [{ + "date": tag.isoformat(), + "amount_cents": vorzeichen, + "payee": payload.payee, + "purpose": payload.note, + }]) + kategorie_id = erkannt[0].get("category_id") if erkannt else None + + txn = Transaction( + owner_id=user.id, + type=TxnType.expense if payload.kind == "expense" else TxnType.income, + date=tag, + account_id=konto.id, + amount_cents=betrag, + amount_eur_cents=currency.compute_amount_eur_cents(db, betrag, konto.currency, tag), + category_id=kategorie_id, + payee=(payload.payee or None), + description=(payload.note or None), + ) + db.add(txn) + db.flush() + + eintrag = InboxItem( + owner_id=user.id, + source=InboxSource.shortcut, + status=InboxStatus.booked, + account_id=konto.id, + date=tag, + amount_cents=vorzeichen, + currency=konto.currency, + payee=(payload.payee or None), + note=(payload.note or None), + category_id=kategorie_id, + transaction_id=txn.id, + external_id=(payload.external_id or None), + ) + db.add(eintrag) + db.commit() + db.refresh(eintrag) + + wohin = f" · {payload.payee}" if payload.payee else "" + return QuickBookingOut( + message=f"{betrag / 100:.2f} {konto.currency} auf {konto.name}{wohin}", + item=InboxItemOut.model_validate(eintrag), + ) diff --git a/backend/app/schemas.py b/backend/app/schemas.py index c4364d2..d9ecb72 100644 --- a/backend/app/schemas.py +++ b/backend/app/schemas.py @@ -1,6 +1,8 @@ """Pydantic-Schemas für Anfragen/Antworten (Phase 0: Auth, Benutzer, Einstellungen).""" from datetime import date as date_t, datetime, time as time_t +from decimal import Decimal +from typing import Literal from pydantic import BaseModel, ConfigDict, Field, field_validator @@ -475,3 +477,66 @@ class FlaggedKeywordOut(BaseModel): word: str applied_count: int overridden_count: int + + +# --- Gerätemarken und Eingang ------------------------------------------------ + + +class ApiTokenCreate(BaseModel): + name: str = Field(min_length=1, max_length=80) + default_account_id: int | None = None + + +class ApiTokenOut(BaseModel): + id: int + name: str + default_account_id: int | None + last_used_at: datetime | None + created_at: datetime | None + + model_config = ConfigDict(from_attributes=True) + + +class ApiTokenCreated(ApiTokenOut): + """Antwort beim Anlegen – der Klartext ist hier das einzige Mal zu sehen.""" + + token: str + + +class QuickBookingIn(BaseModel): + """Was ein Kurzbefehl schickt. Bewusst genügsam: Betrag reicht.""" + + amount: Decimal + payee: str | None = Field(default=None, max_length=255) + date: date_t | None = None + account_id: int | None = None + note: str | None = None + # Kennung des Kurzbefehls, damit ein doppelt ausgelöster Aufruf nicht + # zweimal bucht (schlechtes Netz, versehentlicher Doppeltipp). + external_id: str | None = Field(default=None, max_length=255) + kind: Literal["expense", "income"] = "expense" + + +class InboxItemOut(BaseModel): + id: int + source: str + status: str + account_id: int | None + date: date_t + amount_cents: int + currency: str + payee: str | None + note: str | None + category_id: int | None + transaction_id: int | None + created_at: datetime | None + + model_config = ConfigDict(from_attributes=True) + + +class QuickBookingOut(BaseModel): + """Kurze Rückmeldung – der Kurzbefehl zeigt sie als Mitteilung an.""" + + ok: bool = True + message: str + item: InboxItemOut diff --git a/backend/app/services/csv_import.py b/backend/app/services/csv_import.py index a420d00..d704274 100644 --- a/backend/app/services/csv_import.py +++ b/backend/app/services/csv_import.py @@ -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, diff --git a/backend/migrations/versions/0015_inbox_and_api_tokens.py b/backend/migrations/versions/0015_inbox_and_api_tokens.py new file mode 100644 index 0000000..f1ab0a7 --- /dev/null +++ b/backend/migrations/versions/0015_inbox_and_api_tokens.py @@ -0,0 +1,77 @@ +"""Eingang fuer extern erfasste Zahlungen und Geraetemarken. + +Der Kontoauszug kommt erst am Monatsende; wer eine Zahlung gleich erfasst, +weiss noch, wofuer sie war. ``inbox_items`` haelt solche Zahlungen fest und +bleibt auch nach dem Buchen bestehen, damit der spaetere CSV-Import sie +wiedererkennt und nicht ein zweites Mal verbucht. + +``api_tokens`` gibt Geraeten ein Zugangsmerkmal, die sich nicht anmelden +koennen (iOS-Kurzbefehle). Der Geltungsbereich ist bewusst eng: nur erfassen, +nicht lesen. + +Revision ID: 0015 +Revises: 0014 +""" + +import sqlalchemy as sa +from alembic import op + +revision = "0015" +down_revision = "0014" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "api_tokens", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("owner_id", sa.Integer(), nullable=False), + sa.Column("name", sa.String(length=80), nullable=False), + sa.Column("token_hash", sa.String(length=64), nullable=False), + sa.Column("scope", sa.String(length=32), nullable=False, server_default="quick_add"), + sa.Column("default_account_id", sa.Integer(), nullable=True), + sa.Column("last_used_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=True), + sa.ForeignKeyConstraint(["owner_id"], ["users.id"], ondelete="CASCADE"), + sa.ForeignKeyConstraint(["default_account_id"], ["accounts.id"], ondelete="SET NULL"), + ) + op.create_index("ix_api_tokens_owner_id", "api_tokens", ["owner_id"]) + op.create_index("ix_api_tokens_token_hash", "api_tokens", ["token_hash"], unique=True) + + op.create_table( + "inbox_items", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("owner_id", sa.Integer(), nullable=False), + sa.Column("source", sa.String(length=32), nullable=False), + sa.Column("status", sa.String(length=32), nullable=False, server_default="pending"), + sa.Column("account_id", sa.Integer(), nullable=True), + sa.Column("date", sa.Date(), nullable=False), + sa.Column("amount_cents", sa.Integer(), nullable=False), + sa.Column("currency", sa.String(length=3), nullable=False, server_default="EUR"), + sa.Column("payee", sa.String(length=255), nullable=True), + sa.Column("note", sa.Text(), nullable=True), + sa.Column("category_id", sa.Integer(), nullable=True), + sa.Column("transaction_id", sa.Integer(), nullable=True), + sa.Column("external_id", sa.String(length=255), nullable=True), + sa.Column("raw_text", sa.Text(), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=True), + sa.ForeignKeyConstraint(["owner_id"], ["users.id"], ondelete="CASCADE"), + sa.ForeignKeyConstraint(["account_id"], ["accounts.id"], ondelete="CASCADE"), + sa.ForeignKeyConstraint(["category_id"], ["categories.id"], ondelete="SET NULL"), + # SET NULL, nicht CASCADE: Wird die Buchung geloescht, muss der + # Eingangs-Eintrag bleiben – sonst kaeme die Zahlung beim naechsten + # Auszug unbemerkt ein zweites Mal herein. + sa.ForeignKeyConstraint(["transaction_id"], ["transactions.id"], ondelete="SET NULL"), + sa.UniqueConstraint("owner_id", "source", "external_id", + name="uq_inbox_owner_source_external"), + ) + op.create_index("ix_inbox_items_owner_id", "inbox_items", ["owner_id"]) + op.create_index("ix_inbox_items_account_id", "inbox_items", ["account_id"]) + op.create_index("ix_inbox_items_date", "inbox_items", ["date"]) + + +def downgrade() -> None: + op.drop_table("inbox_items") + op.drop_table("api_tokens") diff --git a/backend/tests/test_quick.py b/backend/tests/test_quick.py new file mode 100644 index 0000000..804a095 --- /dev/null +++ b/backend/tests/test_quick.py @@ -0,0 +1,243 @@ +"""Schnellerfassung per Gerätemarke (iOS-Kurzbefehl).""" + +from fastapi.testclient import TestClient + + +def _fremd(): + """Client OHNE Sitzung. Die Fixture ``client`` ist derselbe Client wie + ``admin`` und traegt dessen Anmeldung – damit liesse sich nicht pruefen, + ob eine Marke fuer sich allein etwas darf.""" + from app.main import app + + return TestClient(app) + + +def _konto(admin, **kw): + body = {"name": "Girokonto", "type": "girokonto", "currency": "CHF", + "opening_balance_cents": 100000} + body.update(kw) + r = admin.post("/api/accounts", json=body) + assert r.status_code == 201, r.text + return r.json() + + +def _marke(admin, konto_id=None, name="iPhone"): + r = admin.post("/api/inbox/tokens", json={"name": name, "default_account_id": konto_id}) + assert r.status_code == 201, r.text + return r.json() + + +def _kategorie(admin, name): + for g in admin.get("/api/categories").json(): + for c in g["categories"]: + if c["name"] == name: + return c["id"] + raise AssertionError(f"{name} fehlt") + + +def test_marke_bucht_und_taucht_im_eingang_auf(admin): + client = _fremd() + konto = _konto(admin) + marke = _marke(admin, konto["id"]) + + r = client.post("/api/quick/booking", json={"amount": "12.50", "payee": "Kantine SV"}, + headers={"X-Certain-Token": marke["token"]}) + assert r.status_code == 201, r.text + assert "12.50 CHF" in r.json()["message"] + + assert admin.get(f"/api/accounts/{konto['id']}").json()["balance_cents"] == 100000 - 1250 + meine = [e for e in admin.get("/api/inbox").json() if e["payee"] == "Kantine SV"] + assert len(meine) == 1 + assert meine[0]["status"] == "booked" + + +def test_marke_darf_nur_erfassen_nichts_lesen(admin): + """Die Marke liegt auf dem Telefon. Sie darf keine Daten herausgeben.""" + client = _fremd() + konto = _konto(admin) + marke = _marke(admin, konto["id"]) + kopf = {"X-Certain-Token": marke["token"]} + + for pfad in ("/api/transactions", "/api/accounts", "/api/inbox", "/api/export/all.json", "/api/backup/db", "/api/categories"): + r = client.get(pfad, headers=kopf) + assert r.status_code == 401, f"{pfad} gab {r.status_code} statt 401" + + +def test_unbekannte_oder_fehlende_marke_wird_abgewiesen(admin): + client = _fremd() + _konto(admin) + assert client.post("/api/quick/booking", json={"amount": "5"}).status_code == 401 + r = client.post("/api/quick/booking", json={"amount": "5"}, + headers={"X-Certain-Token": "ausgedacht"}) + assert r.status_code == 401 + + +def test_widerrufene_marke_bucht_nicht_mehr(admin): + client = _fremd() + konto = _konto(admin) + marke = _marke(admin, konto["id"]) + kopf = {"X-Certain-Token": marke["token"]} + assert client.post("/api/quick/booking", json={"amount": "1"}, headers=kopf).status_code == 201 + + assert admin.delete(f"/api/inbox/tokens/{marke['id']}").status_code == 204 + assert client.post("/api/quick/booking", json={"amount": "1"}, headers=kopf).status_code == 401 + + +def test_klartext_der_marke_ist_nur_einmal_zu_sehen(admin): + konto = _konto(admin) + marke = _marke(admin, konto["id"]) + assert marke["token"] + liste = admin.get("/api/inbox/tokens").json() + meine = [t for t in liste if t["id"] == marke["id"]] + assert len(meine) == 1 + assert "token" not in meine[0], "Der Klartext darf nie wieder ausgeliefert werden" + + +def test_doppelt_ausgeloester_kurzbefehl_bucht_nur_einmal(admin): + """Schlechtes Netz oder Doppeltipp dürfen nicht zweimal abbuchen.""" + client = _fremd() + konto = _konto(admin) + marke = _marke(admin, konto["id"]) + kopf = {"X-Certain-Token": marke["token"]} + daten = {"amount": "9.90", "payee": "Kiosk", "external_id": "abc-123"} + + a = client.post("/api/quick/booking", json=daten, headers=kopf) + b = client.post("/api/quick/booking", json=daten, headers=kopf) + assert a.status_code == 201 and b.status_code == 201 + assert b.json()["message"] == "Bereits erfasst" + assert a.json()["item"]["id"] == b.json()["item"]["id"] + assert admin.get(f"/api/accounts/{konto['id']}").json()["balance_cents"] == 100000 - 990 + + +def test_schlagwort_schlaegt_die_kategorie_vor(admin): + """Was der CSV-Import kann, kann der Kurzbefehl auch.""" + client = _fremd() + konto = _konto(admin) + leb = _kategorie(admin, "Lebensmittel") + assert admin.post(f"/api/categories/{leb}/keywords", json={"word": "coop"}).status_code == 201 + + marke = _marke(admin, konto["id"]) + r = client.post("/api/quick/booking", json={"amount": "23.40", "payee": "COOP Bahnhof"}, + headers={"X-Certain-Token": marke["token"]}) + assert r.status_code == 201, r.text + assert r.json()["item"]["category_id"] == leb + + +def test_einnahme_und_fremdes_konto(admin): + client = _fremd() + konto = _konto(admin) + zweites = _konto(admin, name="Sparkonto", opening_balance_cents=0) + marke = _marke(admin, konto["id"]) + kopf = {"X-Certain-Token": marke["token"]} + + r = client.post("/api/quick/booking", + json={"amount": "50", "kind": "income", "account_id": zweites["id"]}, + headers=kopf) + assert r.status_code == 201, r.text + assert admin.get(f"/api/accounts/{zweites['id']}").json()["balance_cents"] == 5000 + # Das Standardkonto bleibt unberührt. + assert admin.get(f"/api/accounts/{konto['id']}").json()["balance_cents"] == 100000 + + +def test_verwerfen_loescht_die_buchung_behaelt_aber_die_spur(admin): + client = _fremd() + konto = _konto(admin) + marke = _marke(admin, konto["id"]) + client.post("/api/quick/booking", json={"amount": "7.00", "payee": "Irrtum"}, + headers={"X-Certain-Token": marke["token"]}) + eintrag = admin.get("/api/inbox").json()[0] + + assert admin.delete(f"/api/inbox/{eintrag['id']}").status_code == 204 + assert admin.get(f"/api/accounts/{konto['id']}").json()["balance_cents"] == 100000 + assert not [e for e in admin.get("/api/inbox").json() if e["id"] == eintrag["id"]] + + +def _import(admin, konto_id, csv_text): + import io + import json + return admin.post( + "/api/imports/preview", + files={"file": ("a.csv", io.BytesIO(csv_text.encode("utf-8")), "text/csv")}, + data={"account_id": str(konto_id), "delimiter": ";", "decimal_sep": ",", + "date_format": "%d.%m.%Y", "encoding": "utf-8", "has_header": "true", + "skip_rows": "0", "amount_mode": "signed", + "column_map": json.dumps({"date": 0, "amount": 1, "payee": 2, "purpose": 3})}, + ).json() + + +KOPF = "Datum;Betrag;Empfaenger;Zweck\n" + + +def test_auszug_erkennt_die_bereits_per_kurzbefehl_erfasste_zahlung(admin): + """Der Kern: dieselbe Zahlung darf nicht zweimal in den Saldo laufen. + + Der Import-Schlüssel greift hier nicht – auf dem Auszug steht ein anderer + Text als das, was man am Telefon eingetippt hat, und die Bank bucht ein + bis zwei Tage später. + """ + client = _fremd() + konto = _konto(admin) + marke = _marke(admin, konto["id"]) + client.post("/api/quick/booking", + json={"amount": "12.50", "payee": "Kantine", "date": "2026-05-04"}, + headers={"X-Certain-Token": marke["token"]}) + assert admin.get(f"/api/accounts/{konto['id']}").json()["balance_cents"] == 100000 - 1250 + + # Der Auszug nennt sie anders und datiert sie zwei Tage später. + v = _import(admin, konto["id"], KOPF + "06.05.2026;-12,50;SV GROUP SCHWEIZ AG;Kartenzahlung\n") + zeile = v["rows"][0] + assert zeile["is_duplicate"] is True + assert zeile["duplicate_reason"] == "inbox" + assert zeile["include"] is False + + admin.post("/api/imports/commit", json={ + "account_id": konto["id"], "filename": "mai.csv", + "rows": [r for r in v["rows"] if r["include"]]}) + assert admin.get(f"/api/accounts/{konto['id']}").json()["balance_cents"] == 100000 - 1250 + + +def test_abweichender_betrag_gilt_nicht_als_dieselbe_zahlung(admin): + """Bei Geld ist Toleranz gefährlich: nur der exakte Betrag zählt.""" + client = _fremd() + konto = _konto(admin) + marke = _marke(admin, konto["id"]) + client.post("/api/quick/booking", + json={"amount": "12.50", "payee": "Kantine", "date": "2026-06-04"}, + headers={"X-Certain-Token": marke["token"]}) + + v = _import(admin, konto["id"], KOPF + "05.06.2026;-12,60;SV GROUP;Kartenzahlung\n") + assert v["rows"][0]["is_duplicate"] is False + + +def test_zu_weit_entferntes_datum_gilt_nicht(admin): + client = _fremd() + konto = _konto(admin) + marke = _marke(admin, konto["id"]) + client.post("/api/quick/booking", + json={"amount": "31.00", "payee": "Baumarkt", "date": "2026-07-01"}, + headers={"X-Certain-Token": marke["token"]}) + + v = _import(admin, konto["id"], KOPF + "20.07.2026;-31,00;OBI;Einkauf\n") + assert v["rows"][0]["is_duplicate"] is False + + +def test_zwei_gleiche_zahlungen_verbrauchen_je_einen_eintrag(admin): + """Zweimal derselbe Kaffee am selben Tag: beide Seiten müssen aufgehen.""" + client = _fremd() + konto = _konto(admin) + marke = _marke(admin, konto["id"]) + kopf = {"X-Certain-Token": marke["token"]} + for lauf in ("a", "b"): + client.post("/api/quick/booking", + json={"amount": "4.20", "payee": "Kaffee", "date": "2026-09-02", + "external_id": f"kaffee-{lauf}"}, headers=kopf) + assert admin.get(f"/api/accounts/{konto['id']}").json()["balance_cents"] == 100000 - 840 + + v = _import(admin, konto["id"], KOPF + + "02.09.2026;-4,20;CAFE BAR;Karte\n" + + "02.09.2026;-4,20;CAFE BAR;Karte\n" + + "02.09.2026;-4,20;CAFE BAR;Karte\n") + gruende = [r["duplicate_reason"] for r in v["rows"]] + assert gruende.count("inbox") == 2, gruende + # Die dritte ist echt und wird gebucht. + assert sum(1 for r in v["rows"] if r["include"]) == 1