"""/api/finance/* — CFO/S&OP reports as live JSON + downloadable Excels. Five reports + dashboard rollup. Each report has a JSON endpoint for the UI and an /export sibling that streams an Excel workbook. Access: finance_dashboard / finance_lost_sales / etc. permission keys (Admin + Uprava + Nabava for the operational reports; Admin + Uprava for the bridge). """ from __future__ import annotations from datetime import datetime from typing import Optional from fastapi import APIRouter, Depends, Query, Response from sqlalchemy.orm import Session from backend.api.middleware.auth import CurrentUser, require_section from backend.models.database import get_db from backend.schemas.finance import ( BridgeResponse, ContestRiskResponse, FinanceDashboard, LockedCashResponse, LostSalesMonthlyResponse, LostSalesResponse, SlowMoverResponse, ) from backend.services import finance_service as fin router = APIRouter(prefix="/finance", tags=["finance"]) _REQ_DASH = Depends(require_section("finance_dashboard")) _REQ_LOST = Depends(require_section("finance_lost_sales")) _REQ_LOCKED = Depends(require_section("finance_locked_cash")) _REQ_CONTEST = Depends(require_section("finance_contest_risk")) _REQ_SLOW = Depends(require_section("finance_slow_movers")) _REQ_BRIDGE = Depends(require_section("finance_bridge")) def _xlsx_response(data: bytes, filename: str) -> Response: stamp = datetime.now().strftime("%Y%m%d_%H%M") return Response( content=data, media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", headers={"Content-Disposition": f'attachment; filename="{filename}_{stamp}.xlsx"'}, ) # ── Dashboard ────────────────────────────────────────────────────────── @router.get("/dashboard", response_model=FinanceDashboard) def get_dashboard(db: Session = Depends(get_db), _: CurrentUser = _REQ_DASH) -> FinanceDashboard: ctx = fin.FinanceContext(db) return fin.dashboard(ctx, db) # ── Lost sales ───────────────────────────────────────────────────────── @router.get("/lost-sales", response_model=LostSalesResponse) def get_lost_sales(db: Session = Depends(get_db), _: CurrentUser = _REQ_LOST) -> LostSalesResponse: ctx = fin.FinanceContext(db) r = fin.report_lost_sales(ctx) return { "summary": r["summary"], "rows": r["df"].to_dict("records"), "by_category": r["by_category"].rename(columns={"lost_revenue_eur":"lost_revenue_eur"}).to_dict("records"), "by_tier": r["by_tier"].to_dict("records"), "top5": r["top5"], } @router.get("/lost-sales/export") def export_lost_sales(db: Session = Depends(get_db), _: CurrentUser = _REQ_LOST) -> Response: ctx = fin.FinanceContext(db) r = fin.report_lost_sales(ctx) return _xlsx_response(fin.export_lost_sales_xlsx(r), "lost_sales") # ── Lost sales — monthly past-looking (RUC, Gold/Silver/Bronze only) ── @router.get("/lost-sales-monthly", response_model=LostSalesMonthlyResponse) def get_lost_sales_monthly( year: int = Query(..., ge=2000, le=2100), month: int = Query(..., ge=1, le=12), tiers: Optional[list[str]] = Query(None), db: Session = Depends(get_db), _: CurrentUser = _REQ_LOST, ) -> LostSalesMonthlyResponse: return fin.report_lost_sales_monthly(db, year, month, tiers) # ── Locked cash ──────────────────────────────────────────────────────── @router.get("/locked-cash", response_model=LockedCashResponse) def get_locked_cash(db: Session = Depends(get_db), _: CurrentUser = _REQ_LOCKED) -> LockedCashResponse: ctx = fin.FinanceContext(db) r = fin.report_locked_cash(ctx) return { "summary": r["summary"], "rows": r["df"].to_dict("records"), "by_category": r["by_category"].to_dict("records"), "by_supplier": r["by_supplier"].to_dict("records"), } @router.get("/locked-cash/export") def export_locked_cash(db: Session = Depends(get_db), _: CurrentUser = _REQ_LOCKED) -> Response: ctx = fin.FinanceContext(db) r = fin.report_locked_cash(ctx) return _xlsx_response(fin.export_locked_cash_xlsx(r), "locked_cash") # ── Contest risk ─────────────────────────────────────────────────────── @router.get("/contest-risk/months") def get_contest_months(db: Session = Depends(get_db), _: CurrentUser = _REQ_CONTEST) -> dict: """Upcoming contest months available for selection (current month + later that exist in contest_monthly_plan). Powers the month dropdown.""" ctx = fin.FinanceContext(db) return {"months": fin.list_contest_months(ctx)} @router.get("/contest-risk", response_model=ContestRiskResponse) def get_contest_risk( month_key: Optional[int] = Query(None, description="YYYYMM — pin to a specific contest month; omit for next upcoming"), db: Session = Depends(get_db), _: CurrentUser = _REQ_CONTEST, ) -> ContestRiskResponse: ctx = fin.FinanceContext(db) r = fin.report_contest_risk(ctx, month_key=month_key) return { "summary": r["summary"], "rows": r["df"].to_dict("records"), "top10_at_risk": r["top10_at_risk"], } @router.get("/contest-risk/export") def export_contest_risk( month_key: Optional[int] = Query(None, description="YYYYMM — pin to a specific contest month"), db: Session = Depends(get_db), _: CurrentUser = _REQ_CONTEST, ) -> Response: ctx = fin.FinanceContext(db) r = fin.report_contest_risk(ctx, month_key=month_key) mk = r["summary"].get("contest_month_key") or "contest" return _xlsx_response(fin.export_contest_risk_xlsx(r), f"contest_risk_{mk}") # ── Slow movers ──────────────────────────────────────────────────────── @router.get("/slow-movers", response_model=SlowMoverResponse) def get_slow_movers(db: Session = Depends(get_db), _: CurrentUser = _REQ_SLOW) -> SlowMoverResponse: ctx = fin.FinanceContext(db) r = fin.report_slow_movers(ctx) def _rename_group(df, key): if df is None or df.empty: return [] df = df.rename(columns={key: "group_value"}) return df.to_dict("records") return { "summary": r["summary"], "rows": r["df"].to_dict("records"), "by_category": _rename_group(r["by_category"], "category"), "by_supplier": _rename_group(r["by_supplier"], "supplier"), } @router.get("/slow-movers/export") def export_slow_movers(db: Session = Depends(get_db), _: CurrentUser = _REQ_SLOW) -> Response: ctx = fin.FinanceContext(db) r = fin.report_slow_movers(ctx) return _xlsx_response(fin.export_slow_movers_xlsx(r), "slow_movers") # ── Bridge ───────────────────────────────────────────────────────────── @router.get("/bridge", response_model=BridgeResponse) def get_bridge(db: Session = Depends(get_db), _: CurrentUser = _REQ_BRIDGE) -> BridgeResponse: ctx = fin.FinanceContext(db) r = fin.report_bridge(ctx, db) return { "available": r["available"], "type": r["type"], "summary": r["summary"] if r["summary"] else { "window_weeks": [], "n_skus": 0, "plan_margin": 0, "actual_margin": 0, "volume_effect": 0, "price_effect": 0, "mix_effect": 0, "cogs_effect": 0, "total_variance": 0, }, "rows": r["df"].to_dict("records") if not r["df"].empty else [], "by_category": r["by_category"].rename(columns={"name":"category"}).to_dict("records") if not r["by_category"].empty else [], "waterfall": r["waterfall"], } @router.get("/bridge/export") def export_bridge(db: Session = Depends(get_db), _: CurrentUser = _REQ_BRIDGE) -> Response: ctx = fin.FinanceContext(db) r = fin.report_bridge(ctx, db) return _xlsx_response(fin.export_bridge_xlsx(r), "bridge_analysis") # ── Bridge MONTHLY — full V/P/M/COGS with month selector + tier ────── @router.get("/bridge/monthly") def get_bridge_monthly(db: Session = Depends(get_db), _: CurrentUser = _REQ_BRIDGE) -> dict: ctx = fin.FinanceContext(db) return fin.report_bridge_monthly(ctx, db) @router.get("/bridge/monthly/export") def export_bridge_monthly(db: Session = Depends(get_db), _: CurrentUser = _REQ_BRIDGE) -> Response: ctx = fin.FinanceContext(db) r = fin.report_bridge_monthly(ctx, db) return _xlsx_response(fin.export_bridge_monthly_xlsx(r), "bridge_monthly")