from datetime import date from decimal import Decimal from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel, ConfigDict from sqlalchemy import func, or_, select from sqlalchemy.orm import Session from app.auth import require_auth from app.db import get_session from app.models.tables import Account, Category, Transaction from app.parsers.validate import dedup_hash from app.services.categorize import apply_rules router = APIRouter(prefix="/api/transactions", tags=["transactions"], dependencies=[Depends(require_auth)]) class TransactionIn(BaseModel): account_id: int booking_date: date value_date: date | None = None amount: Decimal purpose: str = "" counterparty: str = "" force: bool = False class TransactionOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int account_id: int statement_id: int | None = None booking_date: date value_date: date | None = None amount: Decimal purpose: str counterparty: str category_id: int | None = None status: str dedup_hash: str is_duplicate: bool class TransactionPatch(BaseModel): category_id: int | None = None def _apply_transaction_filters( stmt, account_id: int | None = None, date_from: date | None = None, date_to: date | None = None, category_id: int | None = None, q: str | None = None, status: str = "confirmed", ): """Wendet die Buchungs-Filter auf ein select()-Statement an. Wird sowohl fuer die Liste als auch fuer die dazugehoerige Gesamtanzahl (Pagination) genutzt, damit beide garantiert dieselben Ergebnisse zaehlen/zeigen.""" if status: stmt = stmt.where(Transaction.status == status) if account_id is not None: stmt = stmt.where(Transaction.account_id == account_id) if date_from is not None: stmt = stmt.where(Transaction.booking_date >= date_from) if date_to is not None: stmt = stmt.where(Transaction.booking_date <= date_to) if category_id is not None: stmt = stmt.where(Transaction.category_id == category_id) if q: like = f"%{q}%" stmt = stmt.where(or_(Transaction.purpose.ilike(like), Transaction.counterparty.ilike(like))) return stmt # Serverseitige Sortierung der Buchungsliste (Ausbaustufe 11): Whitelist # URL-Schlüssel -> Sortierausdruck. Textspalten über lower(), damit SQLite # (Tests) und Postgres gleich (case-insensitiv) sortieren. SORT_COLUMNS = { "datum": Transaction.booking_date, "konto": func.lower(Account.name), "zweck": func.lower(Transaction.purpose), "empfaenger": func.lower(Transaction.counterparty), "betrag": Transaction.amount, "kategorie": func.lower(Category.name), } def normalize_sort(sort: str | None, dir: str | None) -> tuple[str, str]: """Unbekannte Werte fallen still auf den Standard (datum/asc) zurück - kein 422 für manipulierte oder veraltete Links.""" return (sort if sort in SORT_COLUMNS else "datum", dir if dir in ("asc", "desc") else "asc") def _apply_sort(stmt, sort: str | None, dir: str | None): sort, dir = normalize_sort(sort, dir) if sort == "konto": stmt = stmt.outerjoin(Account, Account.id == Transaction.account_id) elif sort == "kategorie": stmt = stmt.outerjoin(Category, Category.id == Transaction.category_id) col = SORT_COLUMNS[sort] if dir == "desc": return stmt.order_by(col.desc().nulls_last(), Transaction.id.desc()) return stmt.order_by(col.asc().nulls_last(), Transaction.id.asc()) @router.get("", response_model=list[TransactionOut]) def list_transactions( account_id: int | None = None, date_from: date | None = None, date_to: date | None = None, category_id: int | None = None, q: str | None = None, status: str = "confirmed", sort: str = "datum", dir: str = "asc", limit: int = 200, offset: int = 0, session: Session = Depends(get_session), ): stmt = _apply_transaction_filters( select(Transaction), account_id=account_id, date_from=date_from, date_to=date_to, category_id=category_id, q=q, status=status, ) stmt = _apply_sort(stmt, sort, dir).offset(offset).limit(limit) return [TransactionOut.model_validate(t) for t in session.execute(stmt).scalars()] def count_transactions( account_id: int | None = None, date_from: date | None = None, date_to: date | None = None, category_id: int | None = None, q: str | None = None, status: str = "confirmed", session: Session = Depends(get_session), ) -> int: """Gesamtanzahl der Buchungen fuer die gleichen Filter wie list_transactions (ohne limit/offset) - Grundlage fuer die Seitenzahl in der GUI-Pagination.""" stmt = _apply_transaction_filters( select(func.count(Transaction.id)), account_id=account_id, date_from=date_from, date_to=date_to, category_id=category_id, q=q, status=status, ) return session.execute(stmt).scalar_one() @router.post("", response_model=TransactionOut, status_code=201) def create_transaction(data: TransactionIn, session: Session = Depends(get_session)): if session.get(Account, data.account_id) is None: raise HTTPException(404, "Konto nicht gefunden") h = dedup_hash(data.account_id, data.booking_date, data.amount, data.purpose) existing = session.execute( select(Transaction).where(Transaction.dedup_hash == h)).scalar() if existing is not None and not data.force: raise HTTPException(409, "Buchung existiert bereits (Duplikat)") tx = Transaction( account_id=data.account_id, booking_date=data.booking_date, value_date=data.value_date, amount=data.amount, purpose=data.purpose, counterparty=data.counterparty, status="confirmed", dedup_hash=h, is_duplicate=existing is not None, ) session.add(tx) session.commit() session.refresh(tx) apply_rules(session, [tx]) session.commit() session.refresh(tx) return TransactionOut.model_validate(tx) @router.patch("/{tx_id}", response_model=TransactionOut) def patch_transaction(tx_id: int, data: TransactionPatch, session: Session = Depends(get_session)): tx = session.get(Transaction, tx_id) if tx is None: raise HTTPException(404, "Buchung nicht gefunden") if "category_id" in data.model_fields_set: if data.category_id is not None and session.get(Category, data.category_id) is None: raise HTTPException(404, "Kategorie nicht gefunden") tx.category_id = data.category_id session.commit() session.refresh(tx) return TransactionOut.model_validate(tx)