from datetime import datetime from typing import Optional from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel from sqlmodel import Session, col, select from app.auth import get_current_user from app.db import get_session from app.models.models import ( InstallmentPlan, InstallmentPlanCreate, InstallmentPlanUpdate, Transaction, TransactionRead, TransactionSource, TransactionType, ) from app.services.installments import ( compute_installment_amounts, create_plan, delete_plan, regenerate_cuotas, ) from app.timeutil import utcnow router = APIRouter(prefix="/installment-plans", tags=["installments"]) MIN_INSTALLMENTS = 2 MAX_INSTALLMENTS = 48 class InstallmentPlanRead(BaseModel): id: int anchor_transaction_id: int merchant: str purchase_date: datetime num_installments: int first_installment_date: datetime total_amount: float installment_amount: float last_installment_amount: float currency: str notes: Optional[str] = None cuotas_billed: int paid_amount: float remaining_amount: float next_cuota_date: Optional[datetime] = None is_completed: bool class InstallmentPlanDetail(InstallmentPlanRead): cuotas: list[TransactionRead] = [] class InstallmentPlanListResponse(BaseModel): plans: list[InstallmentPlanRead] total_remaining: float def _validate_num_installments(n: int) -> None: if not MIN_INSTALLMENTS <= n <= MAX_INSTALLMENTS: raise HTTPException( status_code=400, detail=f"num_installments must be {MIN_INSTALLMENTS}-{MAX_INSTALLMENTS}", ) def _get_plan_and_anchor( session: Session, plan_id: int ) -> tuple[InstallmentPlan, Transaction]: plan = session.get(InstallmentPlan, plan_id) if not plan: raise HTTPException(status_code=404, detail="Installment plan not found") anchor = session.get(Transaction, plan.anchor_transaction_id) if not anchor: raise HTTPException(status_code=404, detail="Anchor transaction not found") return plan, anchor def _to_read( plan: InstallmentPlan, anchor: Transaction, cuotas: list[Transaction] ) -> InstallmentPlanRead: now = utcnow() billed = [c for c in cuotas if c.date <= now] future = [c for c in cuotas if c.date > now] paid = float(sum(c.amount for c in billed)) total = float(plan.total_amount) amounts = compute_installment_amounts( plan.total_amount, plan.num_installments ) return InstallmentPlanRead( id=plan.id, anchor_transaction_id=plan.anchor_transaction_id, merchant=anchor.merchant, purchase_date=anchor.date, num_installments=plan.num_installments, first_installment_date=plan.first_installment_date, total_amount=total, installment_amount=float(plan.installment_amount), last_installment_amount=float(amounts[-1]), currency=plan.currency.value, notes=plan.notes, cuotas_billed=len(billed), paid_amount=paid, remaining_amount=total - paid, next_cuota_date=min((c.date for c in future), default=None), is_completed=len(billed) == plan.num_installments, ) def _cuotas_by_plan( session: Session, plan_ids: list[int] ) -> dict[int, list[Transaction]]: grouped: dict[int, list[Transaction]] = {pid: [] for pid in plan_ids} if plan_ids: rows = session.exec( select(Transaction) .where(col(Transaction.installment_plan_id).in_(plan_ids)) .order_by(col(Transaction.date)) ).all() for row in rows: grouped[row.installment_plan_id].append(row) return grouped @router.get("/", response_model=InstallmentPlanListResponse) def list_installment_plans( session: Session = Depends(get_session), _user: str = Depends(get_current_user), ): plans = session.exec(select(InstallmentPlan)).all() anchors = { t.id: t for t in session.exec( select(Transaction).where( col(Transaction.id).in_([p.anchor_transaction_id for p in plans]) ) ).all() } if plans else {} cuotas = _cuotas_by_plan(session, [p.id for p in plans]) reads = [ _to_read(p, anchors[p.anchor_transaction_id], cuotas[p.id]) for p in plans if p.anchor_transaction_id in anchors ] reads.sort(key=lambda r: r.purchase_date, reverse=True) return InstallmentPlanListResponse( plans=reads, total_remaining=sum(r.remaining_amount for r in reads), ) @router.get("/{plan_id}", response_model=InstallmentPlanDetail) def get_installment_plan( plan_id: int, session: Session = Depends(get_session), _user: str = Depends(get_current_user), ): plan, anchor = _get_plan_and_anchor(session, plan_id) cuotas = _cuotas_by_plan(session, [plan.id])[plan.id] read = _to_read(plan, anchor, cuotas) return InstallmentPlanDetail(**read.model_dump(), cuotas=cuotas) @router.post("/", response_model=InstallmentPlanRead, status_code=201) def create_installment_plan( data: InstallmentPlanCreate, session: Session = Depends(get_session), _user: str = Depends(get_current_user), ): """Convert an existing credit-card purchase into a Tasa Cero plan.""" _validate_num_installments(data.num_installments) anchor = session.get(Transaction, data.transaction_id) if not anchor: raise HTTPException(status_code=404, detail="Transaction not found") if anchor.source != TransactionSource.CREDIT_CARD: raise HTTPException( status_code=400, detail="Only credit card transactions can be financed" ) if anchor.transaction_type != TransactionType.COMPRA: raise HTTPException( status_code=400, detail="Only COMPRA transactions can be financed" ) if anchor.installment_plan_id: raise HTTPException( status_code=400, detail="A cuota cannot be converted to a plan" ) if anchor.is_installment_anchor: raise HTTPException( status_code=409, detail="Transaction already has an installment plan" ) plan = create_plan( session, anchor, data.num_installments, data.first_installment_date ) session.commit() session.refresh(plan) cuotas = _cuotas_by_plan(session, [plan.id])[plan.id] return _to_read(plan, anchor, cuotas) @router.patch("/{plan_id}", response_model=InstallmentPlanRead) def update_installment_plan( plan_id: int, data: InstallmentPlanUpdate, session: Session = Depends(get_session), _user: str = Depends(get_current_user), ): """Edit a plan; cuota rows are regenerated (per-cuota edits are lost).""" plan, anchor = _get_plan_and_anchor(session, plan_id) if data.num_installments is not None: _validate_num_installments(data.num_installments) if data.notes is not None: plan.notes = data.notes regenerate_cuotas( session, plan, anchor, data.num_installments, data.first_installment_date ) session.commit() session.refresh(plan) cuotas = _cuotas_by_plan(session, [plan.id])[plan.id] return _to_read(plan, anchor, cuotas) @router.delete("/{plan_id}", status_code=204) def delete_installment_plan( plan_id: int, session: Session = Depends(get_session), _user: str = Depends(get_current_user), ): """Unconvert: remove the plan and its cuotas; the anchor becomes a normal transaction again.""" plan, anchor = _get_plan_and_anchor(session, plan_id) delete_plan(session, plan, anchor) session.commit()