"""Abos und Verträge: wiederkehrende Zahlungen mit bekanntem Betrag. Netflix, Handyvertrag, Krankenkasse, Halbtax – dieselbe Form, deshalb derselbe Bereich. Hier wird nur verzeichnet und gerechnet; erkannt und gebucht wird (noch) nichts. Der Monatswert entsteht in ``Subscription.monatlich_cents``: Betrag geteilt durch Intervall. Für die Summe am Kopf wird er zusätzlich in die Hauptwährung umgerechnet – ein Vertrag in Euro und einer in Franken lassen sich sonst nicht addieren. """ from datetime import date from fastapi import APIRouter, Depends, File, HTTPException, Response, UploadFile, status from sqlalchemy.orm import Session from ..database import get_db from ..deps import get_current_user, get_owned from ..models import ( Account, Category, CategoryKind, Subscription, SubscriptionLogo, Transaction, User, ) from ..schemas import ( SubscriptionIn, SubscriptionListOut, SubscriptionOut, SubscriptionSummary, SubscriptionUpdate, ) from ..services import abo_erkennung, currency, logo as logo_service from .settings import BASE_CURRENCY_KEY, get_user_setting router = APIRouter(prefix="/subscriptions", tags=["subscriptions"]) def _base(db: Session, user: User) -> str: return get_user_setting(db, user, BASE_CURRENCY_KEY) or "EUR" def _pruefe_verweise(db: Session, user: User, daten: dict) -> None: """Konten und Kategorie müssen dem Nutzer gehören. Die Kategorie muss ausserdem eine AUSGABE sein: Ein Vertrag kostet Geld, und gegen eine Einnahme-Kategorie liesse er sich später nicht budgetieren. """ for feld in ("account_id", "savings_account_id"): if daten.get(feld) is not None: get_owned(db, Account, daten[feld], user) if daten.get("category_id") is not None: kategorie = get_owned(db, Category, daten["category_id"], user) if kategorie.type != CategoryKind.expense: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=(f"'{kategorie.name}' ist keine Ausgabe-Kategorie. Ein " f"Vertrag kostet Geld."), ) def _logo_stand(db: Session, sub_id: int) -> int | None: """Wann das Logo zuletzt geaendert wurde – oder None, wenn es keins gibt. Bewusst NICHT das Bild: Die Liste laedt es einzeln nach, sonst haenge sie voller Base64. Der Zeitstempel wandert in die Bild-Adresse und ueberholt damit den Zwischenspeicher des Browsers; ohne ihn zeigte er nach dem Austauschen weiter das alte Logo. """ row = ( db.query(SubscriptionLogo.updated_at) .filter(SubscriptionLogo.subscription_id == sub_id) .first() ) if row is None: return None return int(row[0].timestamp()) if row[0] else 0 def to_out(db: Session, s: Subscription, base: str, heute: date) -> SubscriptionOut: stand = _logo_stand(db, s.id) monatlich = s.monatlich_cents umgerechnet = ( monatlich if s.currency == base else currency.convert(db, monatlich, s.currency, base, heute) ) return SubscriptionOut( **{k: getattr(s, k) for k in ( "id", "name", "icon", "color", "amount_cents", "currency", "interval_months", "account_id", "category_id", "next_due", "payee_pattern", "savings_account_id", "is_active", "cancelled_at", "notes", "position")}, monatlich_cents=monatlich, monatlich_base_cents=umgerechnet, tage_bis_faellig=(s.next_due - heute).days if s.next_due else None, has_logo=stand is not None, logo_version=stand, ) @router.get("", response_model=SubscriptionListOut) def list_subscriptions(db: Session = Depends(get_db), user: User = Depends(get_current_user)): base = _base(db, user) heute = date.today() zeilen = [ to_out(db, s, base, heute) for s in db.query(Subscription) .filter(Subscription.owner_id == user.id) .order_by(Subscription.position, Subscription.name) ] # Gekündigtes zählt nicht mit – es steht nur noch für den Rückblick da. aktiv = [z for z in zeilen if z.is_active] summe = sum(z.monatlich_base_cents or 0 for z in aktiv) return SubscriptionListOut( subscriptions=zeilen, summary=SubscriptionSummary( currency=base, monatlich_cents=summe, # Nicht monatlich × 12: Bei einem Jahresvertrag hat das Runden des # Monatswerts sonst bis zu elf Rappen Fehler. jaehrlich_cents=summe * 12, anzahl_aktiv=len(aktiv), ohne_kurs=sorted({ z.currency for z in aktiv if z.monatlich_base_cents is None }), ), ) @router.post("", response_model=SubscriptionOut, status_code=status.HTTP_201_CREATED) def create_subscription(payload: SubscriptionIn, db: Session = Depends(get_db), user: User = Depends(get_current_user)): daten = payload.model_dump() _pruefe_verweise(db, user, daten) max_pos = ( db.query(Subscription.position) .filter(Subscription.owner_id == user.id) .order_by(Subscription.position.desc()) .first() ) s = Subscription(owner_id=user.id, position=(max_pos[0] + 1) if max_pos else 0, **daten) s.name = s.name.strip() # Ohne eigenes Muster der Name: Er ist der beste erste Versuch, und wer es # besser weiss, traegt es ein. if not s.payee_pattern: s.payee_pattern = s.name db.add(s) db.commit() db.refresh(s) return to_out(db, s, _base(db, user), date.today()) # --- Zuordnung: welche Buchung gehoert zu welchem Vertrag -------------------- @router.post("/zuordnen") def zuordnen(db: Session = Depends(get_db), user: User = Depends(get_current_user)) -> dict: """Buchungen ihren Vertraegen zuordnen. Eindeutig Sicheres wird zugeordnet, der Rest kommt als Vorschlag zurueck. Laeuft ueber ALLE noch nicht zugeordneten Buchungen, auch alte – wer seine Vertraege nachtraegt, bekommt damit rueckwirkend seinen Verlauf. """ return abo_erkennung.finde(db, user) @router.get("/vorschlaege") def vorschlaege(db: Session = Depends(get_db), user: User = Depends(get_current_user)) -> dict: """Nur fragen, nichts veraendern – fuer eine Ansicht, die bloss hinsieht.""" return abo_erkennung.finde(db, user, nur_vorschlaege=True) @router.post("/{sub_id}/buchungen/{txn_id}", response_model=SubscriptionOut) def zuordnen_von_hand(sub_id: int, txn_id: int, db: Session = Depends(get_db), user: User = Depends(get_current_user)): """Eine Buchung diesem Vertrag zuordnen – einen Vorschlag annehmen.""" s = get_owned(db, Subscription, sub_id, user) txn = get_owned(db, Transaction, txn_id, user) txn.subscription_id = s.id if txn.date is not None: abo_erkennung._ruecke_faelligkeit(s, txn.date) db.commit() db.refresh(s) return to_out(db, s, _base(db, user), date.today()) @router.delete("/buchungen/{txn_id}", status_code=status.HTTP_204_NO_CONTENT) def zuordnung_loesen(txn_id: int, db: Session = Depends(get_db), user: User = Depends(get_current_user)) -> None: """Das Etikett abziehen. Die Buchung bleibt, wie sie ist.""" txn = get_owned(db, Transaction, txn_id, user) txn.subscription_id = None db.commit() @router.get("/{sub_id}/verlauf") def verlauf(sub_id: int, db: Session = Depends(get_db), user: User = Depends(get_current_user)) -> dict: """Die erkannten Abbuchungen dieses Vertrags. Beantwortet "was habe ich dafuer insgesamt bezahlt" – und zeigt, wann der Preis gestiegen ist. """ s = get_owned(db, Subscription, sub_id, user) zeilen = abo_erkennung.verlauf(db, user, s) return { "buchungen": zeilen, "anzahl": len(zeilen), "summe_cents": sum(z["amount_cents"] for z in zeilen), "currency": zeilen[0]["currency"] if zeilen else s.currency, } @router.post("/{sub_id}/preis", response_model=SubscriptionOut) def preis_uebernehmen(sub_id: int, amount_cents: int, db: Session = Depends(get_db), user: User = Depends(get_current_user)): """Einen geaenderten Preis uebernehmen. Eigener Schritt und nicht automatisch: Eine Preiserhoehung ist eine Nachricht an den Nutzer, kein Wartungsvorgang. Wer sie stillschweigend nachzieht, nimmt ihm die einzige Gelegenheit, sie zu bemerken. """ if amount_cents < 0: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Betrag darf nicht negativ sein") s = get_owned(db, Subscription, sub_id, user) s.amount_cents = amount_cents db.commit() db.refresh(s) return to_out(db, s, _base(db, user), date.today()) @router.get("/{sub_id}", response_model=SubscriptionOut) def get_subscription(sub_id: int, db: Session = Depends(get_db), user: User = Depends(get_current_user)): return to_out(db, get_owned(db, Subscription, sub_id, user), _base(db, user), date.today()) @router.patch("/{sub_id}", response_model=SubscriptionOut) def update_subscription(sub_id: int, payload: SubscriptionUpdate, db: Session = Depends(get_db), user: User = Depends(get_current_user)): s = get_owned(db, Subscription, sub_id, user) daten = payload.model_dump(exclude_unset=True) _pruefe_verweise(db, user, daten) for key, value in daten.items(): # Diese Felder muessen sich LEEREN lassen – sonst wird man ein einmal # gesetztes Sparkonto oder Faelligkeitsdatum nie wieder los. leerbar = key in ("account_id", "category_id", "next_due", "savings_account_id", "cancelled_at", "notes", "payee_pattern") if value is not None or leerbar: setattr(s, key, value) if s.name: s.name = s.name.strip() db.commit() db.refresh(s) return to_out(db, s, _base(db, user), date.today()) @router.delete("/{sub_id}", status_code=status.HTTP_204_NO_CONTENT) def delete_subscription(sub_id: int, db: Session = Depends(get_db), user: User = Depends(get_current_user)): db.delete(get_owned(db, Subscription, sub_id, user)) db.commit() # --- Markenlogo ------------------------------------------------------------- @router.put("/{sub_id}/logo", response_model=SubscriptionOut) async def upload_logo(sub_id: int, file: UploadFile = File(...), db: Session = Depends(get_db), user: User = Depends(get_current_user)): """Ein Logo hochladen. Ersetzt ein vorhandenes. Am Logo findet man den Vertrag schneller als am Namen – Netflix erkennt man am roten N, nicht am Wort. """ s = get_owned(db, Subscription, sub_id, user) try: klein = logo_service.mach_logo(await file.read()) except logo_service.LogoFehler as e: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) from e vorhanden = db.get(SubscriptionLogo, s.id) if vorhanden is None: db.add(SubscriptionLogo(subscription_id=s.id, content=klein, size_bytes=len(klein))) else: vorhanden.content = klein vorhanden.size_bytes = len(klein) db.commit() return to_out(db, s, _base(db, user), date.today()) @router.get("/{sub_id}/logo") def get_logo(sub_id: int, db: Session = Depends(get_db), user: User = Depends(get_current_user)): # Ueber get_owned, nicht ueber die Logo-Tabelle direkt: Sonst laege das # Bild eines fremden Vertrags hinter einer geratenen Nummer offen. s = get_owned(db, Subscription, sub_id, user) row = db.get(SubscriptionLogo, s.id) if row is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Kein Logo") return Response( content=row.content, media_type=logo_service.MIME, headers={ # Kurz, aber nicht null: Die Liste fragt es je Zeile ab. Beim # Austauschen haengt die Fassung in der Adresse (?v=...), das # ueberholt den Zwischenspeicher. "Cache-Control": "private, max-age=300", }, ) @router.delete("/{sub_id}/logo", response_model=SubscriptionOut) def delete_logo(sub_id: int, db: Session = Depends(get_db), user: User = Depends(get_current_user)): s = get_owned(db, Subscription, sub_id, user) row = db.get(SubscriptionLogo, s.id) if row is not None: db.delete(row) db.commit() return to_out(db, s, _base(db, user), date.today())