"""/api/demand/* router. Routers stay thin: parse query params, call a service, return the result. No SQL, no business metrics — both belong one layer down. """ from __future__ import annotations from typing import Optional from fastapi import APIRouter, Body, Depends, File, Form, HTTPException, Path, Query, UploadFile from fastapi.responses import StreamingResponse from sqlalchemy import text from sqlalchemy.orm import Session from backend.api.middleware.auth import CurrentUser, get_current_user, require_ingest_or_user, require_role, require_section from backend.models.database import get_db from backend.schemas.demand import ( ConsensusDetail, ConsensusDiff, ConsensusListResponse, DemandPlanResponse, FASummary, FATabResponse, FABreakdownResponse, ForecastPrereqsResponse, ForecastRunRow, InputTemplateResponse, InputUserRow, BuyerAliasUpsertRequest, BuyerAliasUpsertResponse, BuyerCellEditRequest, BuyerCellEditResponse, BuyerCellsBatchRequest, BuyerCellsBatchResponse, BuyerFAResponse, BuyerPlannerResponse, PartnerSearchRow, WholesaleAvailResponse, DemandReviewNote, DemandReviewNoteCreate, DemandReviewNoteResolveRequest, DemandReviewNotesPayload, DemandReviewResponse, KAMBuyerFAResponse, KAMFAResponse, KamControlResponse, OnTopChangeRow, OnTopChangesResponse, SaveInputsV2Request, SaveInputsV2Response, PlanningViewResponse, RevenueForecastResponse, RevenueSummary, SavePlannerFactorsRequest, RunForecastRequest, RunForecastResponse, SalesFiltersApplied, SalesWeeklyResponse, SaveInputsRequest, SaveInputsResponse, SkuDetailResponse, SkuInfo, SkuWeeklySalesResponse, SopMeetingPackage, SubmissionStatusResponse, StockUploadResponse, UploadResponse, WatchlistResponse, WholesaleInputTemplate, SaveWholesaleInputRequest, SaveWholesaleInputResponse, BuyerListingUploadResponse, KamUploadParseResponse, KamUploadApplyResponse, ) from backend.services.demand_service import DemandService from backend.services.forecast_service import ForecastService from backend.services.upload_service import UploadService from backend.services.upload_stock_service import process_stock_upload router = APIRouter(prefix="/demand", tags=["demand"]) @router.get("/sales-daily") def get_sales_daily( date_from: str = Query(..., description="YYYY-MM-DD"), date_to: str = Query(..., description="YYYY-MM-DD"), category: list[str] = Query(default=[], description="Filter by category name (repeat for multi)"), db: Session = Depends(get_db), ) -> dict: """Daily sales by channel — reads erp_transactions directly (preserves day-level granularity from Rekapitulacija uploads). Returns one row per (date × channel) with qty, net revenue, RUC, and a total row per date. Useful when the weekly aggregation drops detail you need (Mon vs Sun, retail spike days, etc.).""" from sqlalchemy import text where = ["et.transaction_date BETWEEN :df AND :dt", "et.quantity > 0"] params = {"df": date_from, "dt": date_to} if category: where.append("c.name = ANY(:cats)") params["cats"] = category sql = text(f""" SELECT et.transaction_date::text AS day, cm.channel, SUM(et.quantity)::float AS qty, SUM(COALESCE(NULLIF(et.tax_base,0), et.total_value * 0.80))::float AS revenue_eur, SUM(et.ruc_eur)::float AS ruc_eur, COUNT(*) AS n_lines FROM erp_transactions et JOIN lookup_channel_map cm ON cm.id = et.channel_map_id JOIN dim_products p ON p.id = et.product_id LEFT JOIN dim_categories c ON c.id = p.category_id WHERE {' AND '.join(where)} GROUP BY et.transaction_date, cm.channel ORDER BY et.transaction_date, cm.channel """) rows = [dict(r) for r in db.execute(sql, params).mappings()] # Daily totals (all channels) by_day: dict[str, dict] = {} for r in rows: d = r["day"] b = by_day.setdefault(d, {"day": d, "qty": 0.0, "revenue_eur": 0.0, "ruc_eur": 0.0, "n_lines": 0}) b["qty"] += float(r["qty"] or 0) b["revenue_eur"] += float(r["revenue_eur"] or 0) b["ruc_eur"] += float(r["ruc_eur"] or 0) b["n_lines"] += int(r["n_lines"] or 0) grand = { "qty": sum(b["qty"] for b in by_day.values()), "revenue_eur": sum(b["revenue_eur"] for b in by_day.values()), "ruc_eur": sum(b["ruc_eur"] for b in by_day.values()), "n_lines": sum(b["n_lines"] for b in by_day.values()), "n_days": len(by_day), } return { "date_from": date_from, "date_to": date_to, "rows": rows, "daily_totals": sorted(by_day.values(), key=lambda b: b["day"]), "grand_total": grand, } @router.get("/sku-weekly-sales", response_model=SkuWeeklySalesResponse) def get_sku_weekly_sales( sku: str = Query(..., min_length=1), n_weeks: int = Query(13, ge=1, le=52), db: Session = Depends(get_db), ) -> SkuWeeklySalesResponse: """Last N ISO weeks of one SKU's sales: channel split, promo flag per week, and wholesale qty broken down by buyer (for the hover).""" return DemandService(db).get_sku_weekly_sales(sku, n_weeks=n_weeks) @router.get("/sales-weekly", response_model=SalesWeeklyResponse) def get_sales_weekly( tier: list[str] = Query(default=[], description="GOLD / SILVER / BRONZE (repeat for multi)"), xyz: list[str] = Query(default=[], description="X / Y / Z (repeat for multi)"), category: list[str] = Query(default=[], description="Category name (exact, repeat for multi)"), year: Optional[int] = Query(None, description="ISO year"), week_from: Optional[int] = Query(None, ge=1, le=53), week_to: Optional[int] = Query(None, ge=1, le=53), sort_by: Optional[str] = Query(None, description="One of: sku, name, category, year, week, qty_retail, qty_webshop, qty_wholesale, qty_total, tier, xyz"), sort_dir: str = Query("asc", pattern="^(asc|desc)$"), page: int = Query(1, ge=1), page_size: int = Query(50, ge=1, le=1000), db: Session = Depends(get_db), ) -> SalesWeeklyResponse: data = DemandService(db).get_sales_data( tier=tier or None, xyz=xyz or None, category=category or None, year=year, week_from=week_from, week_to=week_to, sort_by=sort_by, sort_dir=sort_dir, page=page, page_size=page_size, ) return SalesWeeklyResponse( **data, filters_applied=SalesFiltersApplied( tier=tier or None, xyz=xyz or None, category=category or None, year=year, week_from=week_from, week_to=week_to, sort_by=sort_by, sort_dir=sort_dir, page=page, page_size=page_size, ), ) # Note: The forecast-projection revenue page (matching Streamlit's page_revenue) will be added # when forecasts table is populated — see prompt 4A @router.get("/sales-history", response_model=RevenueSummary) def get_sales_history( year: int = Query(..., description="ISO year"), db: Session = Depends(get_db), ) -> RevenueSummary: return DemandService(db).get_revenue_data(year=year) # Backward-compat alias — kept so any existing client calls still work @router.get("/revenue", response_model=RevenueSummary, include_in_schema=False) def get_revenue( year: int = Query(..., description="ISO year"), db: Session = Depends(get_db), ) -> RevenueSummary: return DemandService(db).get_revenue_data(year=year) @router.get("/categories", response_model=list[str]) def get_categories(db: Session = Depends(get_db)) -> list[str]: return DemandService(db).get_categories() @router.get("/forecast-accuracy", response_model=FASummary) def get_forecast_accuracy( tier: Optional[str] = Query(None), db: Session = Depends(get_db), ) -> FASummary: """Legacy single-tab FA endpoint (kept for backward-compat with any client that still hits it). The 4-tab UI uses /forecast-accuracy/* below.""" return DemandService(db).get_fa_summary(tier=tier) # ----- 4-tab Forecast Accuracy -------------------------------------------- # Each tab returns the same FATabResponse shape; only the data source differs. # Filters (tier / xyz / category / date range) are identical across all four. # KAM·CM has its own shape (KAMFAResponse) because the rollup dimension is # people, not tier/xyz. def _parse_yw(value: Optional[str]) -> Optional[int]: """Accepts YYYYWW (e.g. 202610) or YYYY-WW (e.g. 2026-10). Returns year*100 + week or None.""" if value is None: return None s = str(value).replace("-", "").replace("W", "").replace("w", "").strip() if not s.isdigit() or len(s) != 6: return None return int(s) @router.get("/forecast-accuracy/backtest", response_model=FATabResponse) def get_fa_backtest( tier: list[str] = Query(default=[]), xyz: list[str] = Query(default=[]), category: list[str] = Query(default=[]), date_from: Optional[str] = Query(None, description="YYYYWW or YYYY-WW"), date_to: Optional[str] = Query(None, description="YYYYWW or YYYY-WW"), top_n: int = Query(10, ge=1, le=50), db: Session = Depends(get_db), ) -> FATabResponse: """Backtest FA — backtest_results vs replayed actuals. Largest source (~3.8k rows currently). This is the 'Global FA' tab in the UI.""" return DemandService(db).get_fa_tab( mode="backtest", tier=tier or None, xyz=xyz or None, category=category or None, date_from=_parse_yw(date_from), date_to=_parse_yw(date_to), top_n=top_n, ) @router.get("/forecast-accuracy/live", response_model=FATabResponse) def get_fa_live( tier: list[str] = Query(default=[]), xyz: list[str] = Query(default=[]), category: list[str] = Query(default=[]), date_from: Optional[str] = Query(None), date_to: Optional[str] = Query(None), top_n: int = Query(10, ge=1, le=50), db: Session = Depends(get_db), ) -> FATabResponse: """Live FA — first-forecast-per-week from `forecasts` table vs actuals. Empty until the forecast pipeline starts writing rows (prompt 4A).""" return DemandService(db).get_fa_tab( mode="live", tier=tier or None, xyz=xyz or None, category=category or None, date_from=_parse_yw(date_from), date_to=_parse_yw(date_to), top_n=top_n, ) @router.get("/forecast-accuracy/model", response_model=FATabResponse) def get_fa_model_only( tier: list[str] = Query(default=[]), xyz: list[str] = Query(default=[]), category: list[str] = Query(default=[]), date_from: Optional[str] = Query(None), date_to: Optional[str] = Query(None), top_n: int = Query(10, ge=1, le=50), db: Session = Depends(get_db), ) -> FATabResponse: """Model-only FA — backtest_results minus wholesale channel-mode rows. Once on_top_inputs is populated, will also subtract planner commitments to match Streamlit's `_build_fa_dataset('model_only')` strip-channel rule.""" return DemandService(db).get_fa_tab( mode="model_only", tier=tier or None, xyz=xyz or None, category=category or None, date_from=_parse_yw(date_from), date_to=_parse_yw(date_to), top_n=top_n, ) @router.get("/forecast-accuracy/breakdown", response_model=FABreakdownResponse) def get_fa_breakdown( weeks_back: int = Query(4, ge=1, le=26), top_n: int = Query(12, ge=1, le=50), date_from: Optional[str] = Query(None), date_to: Optional[str] = Query(None), db: Session = Depends(get_db), ) -> FABreakdownResponse: """Channel summary (forecast baseline+on-top vs actual, per channel: Retail / Webshop / Wholesale) + worst-miss SKUs with forecast source and actual source. Window = [date_from, date_to] if given, else last `weeks_back` weeks.""" return DemandService(db).get_fa_breakdown( # type: ignore[return-value] weeks_back=weeks_back, top_n=top_n, date_from=_parse_yw(date_from), date_to=_parse_yw(date_to), ) @router.get("/forecast-accuracy/kam", response_model=KAMFAResponse) def get_fa_kam( tier: list[str] = Query(default=[]), xyz: list[str] = Query(default=[]), category: list[str] = Query(default=[]), date_from: Optional[str] = Query(None), date_to: Optional[str] = Query(None), db: Session = Depends(get_db), ) -> KAMFAResponse: """KAM·CM FA — on_top_inputs per submitter vs channel actuals. Empty until KAM/CM templates are uploaded through the S&OP cycle.""" return DemandService(db).get_kam_fa_summary( tier=tier or None, xyz=xyz or None, category=category or None, date_from=_parse_yw(date_from), date_to=_parse_yw(date_to), ) @router.get("/sku-list", response_model=list[SkuInfo]) def get_sku_list(db: Session = Depends(get_db)) -> list[SkuInfo]: return DemandService(db).get_sku_list() # SKU detail — full profile assembled from many sources. The {sku} path # parameter is the human-readable SKU code (string, e.g. "POL12876"), not the # numeric product_id; this keeps URLs shareable and matches the column users # see in Streamlit / Excel exports. @router.get("/sku/{sku}", response_model=SkuDetailResponse) def get_sku_detail( sku: str = Path(..., min_length=1, max_length=64, description="SKU code (string, e.g. POL12876)"), db: Session = Depends(get_db), ) -> SkuDetailResponse: data = DemandService(db).get_sku_profile(sku) if data is None: raise HTTPException(status_code=404, detail=f"SKU {sku!r} not found") return data @router.get("/watchlist", response_model=WatchlistResponse) def get_watchlist( n: int = Query(30, ge=1, le=200, description="Top N to return"), sort: str = Query( "volume", description="volume / fa / coverage / revenue", pattern="^(volume|fa|coverage|revenue)$", ), db: Session = Depends(get_db), ) -> WatchlistResponse: return DemandService(db).get_watchlist_data(n=n, sort_by=sort) # Consensus snapshots — list / detail / diff. Snapshots are read-only; # creating one happens at cycle close (separate workflow, not in this router). @router.get("/consensus", response_model=ConsensusListResponse) def get_consensus_history( db: Session = Depends(get_db), ) -> ConsensusListResponse: return DemandService(db).get_consensus_history() @router.get("/consensus/compare", response_model=ConsensusDiff) def compare_consensus( a: int = Query(..., description="ID of snapshot A (baseline)"), b: int = Query(..., description="ID of snapshot B (comparison)"), db: Session = Depends(get_db), ) -> ConsensusDiff: """Note: this route is registered BEFORE /consensus/{id} so the literal 'compare' doesn't collide with the int path-param parser.""" if a == b: raise HTTPException(status_code=400, detail="a and b must differ") data = DemandService(db).compare_snapshots(a, b) if data is None: raise HTTPException(status_code=404, detail=f"Snapshot {a} or {b} not found") return data # ── Monthly plan snapshots (May 2026 upgrade) ──────────────────────── # # Active counterpart to the legacy consensus_snapshots archive. Instead of # storing whole-cycle JSONB blobs, this saves a per-SKU plan for a single # calendar month and is consumed by Margin Bridge / Revenue Forecast / # NPL Report. Snapshot is immutable once created (DELETE-then-redo is the # only way to redo). @router.get("/sop-monthly") def get_sop_monthly_review( year: int = Query(..., ge=2020, le=2100), month: int = Query(..., ge=1, le=12), db: Session = Depends(get_db), ) -> dict: """S&OP monthly review payload — plan vs actuals, top movers, monthly FA with reason tagging, lost sales for the month, and (if month is the current calendar month) actual-to-date + projection-to-EOM.""" from backend.services import sop_monthly_service as sop return sop.build_sop_monthly_review(db, year, month) @router.get("/monthly-plan/snapshots") def list_monthly_snapshots(db: Session = Depends(get_db)) -> dict: from backend.services import monthly_plan_service as mp return {"snapshots": mp.list_snapshots(db)} @router.get("/monthly-plan/snapshots/{month_key}") def get_monthly_snapshot( month_key: int = Path(..., description="YYYY*100 + MM, e.g. 202604"), db: Session = Depends(get_db), ) -> dict: from backend.services import monthly_plan_service as mp snap = mp.get_snapshot(db, month_key) if snap is None: raise HTTPException(status_code=404, detail=f"No snapshot for month {month_key}") return snap @router.get("/monthly-plan/preview") def preview_monthly_snapshot( year: int = Query(..., ge=2020, le=2100), month: int = Query(..., ge=1, le=12), db: Session = Depends(get_db), ) -> dict: """Compute (but don't save) a snapshot for the given month. Lets the user see what would be locked before clicking lock.""" from backend.services import monthly_plan_service as mp snap = mp.build_monthly_snapshot(db, year, month) # Strip per_sku_data from preview (heavy) — keep aggregates + notes return {k: v for k, v in snap.items() if k != "per_sku_data"} | { "per_sku_count": len(snap["per_sku_data"]), } @router.post("/monthly-plan/lock") def lock_monthly_plan( year: int = Query(..., ge=2020, le=2100), month: int = Query(..., ge=1, le=12), label: Optional[str] = Query(default=None), db: Session = Depends(get_db), user: CurrentUser = Depends(get_current_user), ) -> dict: """Build and persist a monthly plan snapshot. Snapshots are immutable once locked — to redo, delete via DELETE /monthly-plan/snapshots/{id} first.""" from backend.services import monthly_plan_service as mp try: snap = mp.build_monthly_snapshot(db, year, month) snapshot_id = mp.save_snapshot( db, snap, locked_by_id=user.user_id, label=label or f"{year}-{month:02d}", ) return { "ok": True, "snapshot_id": snapshot_id, "month_key": snap["month_key"], "n_skus": snap["n_skus"], "total_ruc_eur": snap["total_ruc_eur"], "total_revenue_eur": snap["total_revenue_eur"], } except ValueError as e: raise HTTPException(status_code=409, detail=str(e)) @router.delete("/monthly-plan/snapshots/{month_key}") def delete_monthly_snapshot( month_key: int = Path(..., description="YYYY*100 + MM"), db: Session = Depends(get_db), user: CurrentUser = Depends(require_role("Admin")), ) -> dict: """Admin-only: delete a snapshot. Use this to redo a snapshot after fixing upstream data.""" n = db.execute(text( "DELETE FROM monthly_plan_snapshots WHERE month_key = :mk" ), {"mk": month_key}).rowcount db.commit() return {"ok": True, "deleted": n} @router.get("/consensus/{snapshot_id}", response_model=ConsensusDetail) def get_consensus_snapshot( snapshot_id: int = Path(..., ge=1), db: Session = Depends(get_db), ) -> ConsensusDetail: data = DemandService(db).get_consensus_snapshot(snapshot_id) if data is None: raise HTTPException(status_code=404, detail=f"Snapshot {snapshot_id} not found") return data @router.get("/sop-meeting", response_model=SopMeetingPackage) def get_sop_meeting(db: Session = Depends(get_db)) -> SopMeetingPackage: """Complete S&OP review package: exceptions, KPIs, upcoming promos, suggested actions. Read-only — action checkoff is client-side for now; persistence lands when consensus_snapshots gains an `actions` column.""" return DemandService(db).get_sop_meeting_data() # ── Demand Review > Consensus forecast sign-off ─────────────────────── # 13-week rolling overview. Phase 1 returns only the decomposition # waterfall (baseline + planner factor + on-tops + promo = consensus); # headline / drivers / R&O / sign-off blocks come in later phases. @router.get("/review/overview", response_model=DemandReviewResponse) def get_demand_review_overview( db: Session = Depends(get_db), _: CurrentUser = Depends(require_section("demand_review")), ) -> DemandReviewResponse: from backend.services.demand_review_service import report_demand_review return report_demand_review(db) # Risks & Opportunities log — CRUD. Same permission gate as the overview; # resolve/delete enforce author-or-admin in the service layer. @router.get("/review/notes", response_model=DemandReviewNotesPayload) def list_demand_review_notes( db: Session = Depends(get_db), _: CurrentUser = Depends(require_section("demand_review")), ) -> DemandReviewNotesPayload: from backend.services.demand_review_service import list_notes return list_notes(db) @router.post("/review/notes", response_model=DemandReviewNote) def create_demand_review_note( body: DemandReviewNoteCreate, db: Session = Depends(get_db), user: CurrentUser = Depends(require_section("demand_review")), ) -> DemandReviewNote: from backend.services.demand_review_service import create_note, list_notes try: create_note(db, user_id=int(user.user_id), kind=body.kind, severity=body.severity, note=body.note) except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) # Return the freshly-inserted row by reading it back (cheaper than # building the full DemandReviewNote shape from create_note's lean # output; list_notes already joins author + cycle_label). fresh = list_notes(db) bucket = (fresh["risks"] + fresh["opportunities"] + fresh["decisions"]) bucket = sorted(bucket, key=lambda r: r["id"], reverse=True) return bucket[0] @router.patch("/review/notes/{note_id}", response_model=DemandReviewNote) def resolve_demand_review_note( note_id: int, body: DemandReviewNoteResolveRequest, db: Session = Depends(get_db), user: CurrentUser = Depends(require_section("demand_review")), ) -> DemandReviewNote: from backend.services.demand_review_service import ( list_notes, update_note_resolved, ) update_note_resolved(db, note_id=note_id, user_id=int(user.user_id), resolved=bool(body.resolved)) fresh = list_notes(db) all_rows = fresh["risks"] + fresh["opportunities"] + fresh["decisions"] match = next((r for r in all_rows if r["id"] == note_id), None) if not match: raise HTTPException(status_code=404, detail="note not found after update") return match @router.delete("/review/notes/{note_id}") def delete_demand_review_note( note_id: int, db: Session = Depends(get_db), user: CurrentUser = Depends(require_section("demand_review")), ) -> dict: from backend.services.demand_review_service import delete_note is_admin = (getattr(user, "role", "") or "").lower() == "admin" ok = delete_note(db, note_id=note_id, user_id=int(user.user_id), is_admin=is_admin) if not ok: # Either not found, or non-admin trying to delete someone else's row. raise HTTPException( status_code=403, detail="note not found or not owned by you", ) return {"ok": True, "id": note_id} # ── Wholesale Review > Control Module ───────────────────────────────── # KAM commit-vs-pull reconciliation. Surfaces buyer→partner coverage gaps # and classifies every on-top wholesale commit in the last N closed # weeks as first-buy vs incremental, then HIT / PARTIAL / MISS against # the baseline trailing-13w rate of that (partner, SKU) pair. @router.get("/wholesale-review/kam-control", response_model=KamControlResponse) def get_kam_commit_control( n_weeks: int = Query(4, ge=1, le=13), db: Session = Depends(get_db), _: CurrentUser = Depends(require_section("wholesale_review")), ) -> KamControlResponse: from backend.services.wholesale_review_service import report_kam_commit_control return report_kam_commit_control(db, n_weeks=n_weeks) # Buyer Planner — KAM → buyer → SKU×week grid with OOS-risk flags. # Without `buyer` returns just the nav tree; with it, the per-SKU grid. @router.get("/wholesale-review/buyer-planner", response_model=BuyerPlannerResponse) def get_buyer_planner_view( kam: Optional[str] = Query(None), buyer: Optional[str] = Query(None), db: Session = Depends(get_db), _: CurrentUser = Depends(require_section("wholesale_review")), ) -> BuyerPlannerResponse: from backend.services.wholesale_review_service import get_buyer_planner return get_buyer_planner(db, kam=kam, buyer=buyer) @router.patch("/wholesale-review/buyer-planner/cell", response_model=BuyerCellEditResponse) def edit_buyer_planner_cell( body: BuyerCellEditRequest, db: Session = Depends(get_db), user: CurrentUser = Depends(require_section("wholesale_review")), ) -> BuyerCellEditResponse: from backend.services.wholesale_review_service import edit_buyer_cell try: return edit_buyer_cell( db, product_id=body.product_id, buyer=body.buyer, year_week=body.year_week, new_quantity=body.new_quantity, new_year_week=body.new_year_week, editor_user_id=int(user.user_id), ) except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) @router.post("/wholesale-review/buyer-planner/cells", response_model=BuyerCellsBatchResponse) def edit_buyer_planner_cells_batch( body: BuyerCellsBatchRequest, db: Session = Depends(get_db), user: CurrentUser = Depends(require_section("wholesale_review")), ) -> BuyerCellsBatchResponse: from backend.services.wholesale_review_service import edit_buyer_cells_batch try: return edit_buyer_cells_batch( db, [e.model_dump() for e in body.edits], editor_user_id=int(user.user_id), ) except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) @router.get("/wholesale-review/buyer-fa", response_model=BuyerFAResponse) def get_buyer_fa( kam: Optional[str] = Query(None), buyer: Optional[str] = Query(None), db: Session = Depends(get_db), _: CurrentUser = Depends(require_section("wholesale_review")), ) -> BuyerFAResponse: from backend.services.wholesale_review_service import report_buyer_fa return report_buyer_fa(db, kam=kam, buyer=buyer) # ── Availability (wholesale) — available-to-promise per listed buyer ── @router.get("/wholesale-availability", response_model=WholesaleAvailResponse) def get_wholesale_availability( kam: Optional[str] = Query(None), buyer: Optional[str] = Query(None), exclude_on_top: bool = Query(False, description="Strip the on-top portion from buyer projection; show only regular-increase"), commit_filter: str = Query("all", pattern="^(all|on_top|reg|any)$", description="Restrict listed SKUs by commit type: all / on_top / reg / any (union)"), db: Session = Depends(get_db), _: CurrentUser = Depends(require_section("wholesale_review")), ) -> WholesaleAvailResponse: from backend.services.wholesale_review_service import report_wholesale_availability return report_wholesale_availability( db, kam=kam, buyer=buyer, exclude_on_top=exclude_on_top, commit_filter=commit_filter, ) @router.get("/wholesale-availability/export") def export_wholesale_availability( kam: Optional[str] = Query(None), buyer: Optional[str] = Query(None), exclude_on_top: bool = Query(False), commit_filter: str = Query("all", pattern="^(all|on_top|reg|any)$"), only_gaps: bool = Query(False, description="Export only SKUs with at least one short week"), db: Session = Depends(get_db), _: CurrentUser = Depends(require_section("wholesale_review")), ) -> StreamingResponse: """Excel of the availability grid (ATP per SKU per week) for the selected buyer.""" import io from backend.services.wholesale_review_service import export_wholesale_availability_xlsx data = export_wholesale_availability_xlsx( db, kam=kam, buyer=buyer, exclude_on_top=exclude_on_top, commit_filter=commit_filter, only_gaps=only_gaps, ) safe = (buyer or "all").replace("/", "-").replace(" ", "_")[:40] fname = f"availability_ws_{safe}.xlsx" return StreamingResponse( io.BytesIO(data), media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", headers={"Content-Disposition": f'attachment; filename="{fname}"'}, ) @router.get("/wholesale-availability/export-summary") def export_wholesale_availability_summary( kam: Optional[str] = Query(None), buyer: Optional[str] = Query(None), exclude_on_top: bool = Query(False), commit_filter: str = Query("all", pattern="^(all|on_top|reg|any)$"), only_gaps: bool = Query(False, description="Export only at-risk SKUs"), db: Session = Depends(get_db), _: CurrentUser = Depends(require_section("wholesale_review")), ) -> StreamingResponse: """Excel of the simple per-KAM summary: one row per listed SKU, with the earliest short week, units short, committed on-top, and affected buyers.""" import io from backend.services.wholesale_review_service import export_wholesale_availability_summary_xlsx data = export_wholesale_availability_summary_xlsx( db, kam=kam, buyer=buyer, exclude_on_top=exclude_on_top, commit_filter=commit_filter, only_gaps=only_gaps, ) safe = (kam or buyer or "all").replace("/", "-").replace(" ", "_")[:40] fname = f"availability_summary_{safe}.xlsx" return StreamingResponse( io.BytesIO(data), media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", headers={"Content-Disposition": f'attachment; filename="{fname}"'}, ) @router.get("/wholesale-review/partners", response_model=list[PartnerSearchRow]) def search_wholesale_partners( q: str = Query(..., min_length=1), db: Session = Depends(get_db), _: CurrentUser = Depends(require_section("wholesale_review")), ) -> list[PartnerSearchRow]: from backend.services.wholesale_review_service import search_partners return search_partners(db, q) @router.put("/wholesale-review/alias", response_model=BuyerAliasUpsertResponse) def upsert_wholesale_alias( body: BuyerAliasUpsertRequest, db: Session = Depends(get_db), user: CurrentUser = Depends(require_section("wholesale_review")), ) -> BuyerAliasUpsertResponse: from backend.services.wholesale_review_service import upsert_buyer_alias return upsert_buyer_alias(db, buyer_lc=body.buyer_lc, partner_id=body.partner_id, editor_user_id=int(user.user_id)) # Demand Planning grid — read-only. Editable planner_factor adjustment comes # in a later prompt (see CLAUDE.md migration plan). @router.get("/plan", response_model=DemandPlanResponse) def get_demand_plan( tier: list[str] = Query(default=[]), xyz: list[str] = Query(default=[]), category: list[str] = Query(default=[]), db: Session = Depends(get_db), ) -> DemandPlanResponse: return DemandService(db).get_demand_plan_data( tier=tier or None, xyz=xyz or None, category=category or None, ) # Demand Planning page (Streamlit page_demand_planning parity) --------- # Returns a single-SKU-or-aggregate VIEW with chart + numbers data, # unlike /plan which returns an all-SKU pivot grid. @router.get("/planning-view", response_model=PlanningViewResponse) def get_planning_view( category: Optional[str] = Query(default=None), oznaka: Optional[str] = Query(default=None, description="ABC tier — e.g. '01 GOLD'. Single value, ILIKE substring match."), xyz: list[str] = Query(default=[]), sku: Optional[str] = Query(default=None, description="Optional SKU drilldown — when set, returns just that SKU."), date_from: Optional[str] = Query(default=None, description="YYYY-MM-DD. Narrows historical actuals; forecast always full."), date_to: Optional[str] = Query(default=None), db: Session = Depends(get_db), ) -> PlanningViewResponse: from datetime import date as _date def _yw(s: Optional[str]) -> Optional[int]: if not s: return None try: d = _date.fromisoformat(s) iso = d.isocalendar() return iso[0] * 100 + iso[1] except ValueError: return None return DemandService(db).get_planning_view( category=category, oznaka=oznaka, xyz=xyz or None, sku=sku, yw_from=_yw(date_from), yw_to=_yw(date_to), ) @router.post("/planner-factors") def save_planner_factors( body: SavePlannerFactorsRequest, db: Session = Depends(get_db), ) -> dict: return DemandService(db).save_planner_factors( sku=body.sku, factors=[{"cw_label": f.cw_label, "factor": f.factor} for f in body.factors], ) # Revenue forecast (Streamlit page_revenue parity) -------------------- @router.get("/revenue-forecast", response_model=RevenueForecastResponse) def get_revenue_forecast( view: str = Query(default="revenue", description="revenue | ruc"), category: Optional[str] = Query(default=None), source: str = Query(default="all", description="all | vp | mp"), months: list[str] = Query(default=[]), include_nonplanned: bool = Query(default=False), db: Session = Depends(get_db), ) -> RevenueForecastResponse: return DemandService(db).get_revenue_forecast( view=view, category=category, source=source, months_filter=months or None, include_nonplanned=include_nonplanned, ) # ====================================================================== # Upload sales (Rekapitulacija Excel → erp_transactions) # ====================================================================== @router.post("/upload-sales", response_model=UploadResponse) async def upload_sales( files: list[UploadFile] = File(..., description="1–3 Rekapitulacija .xlsx files (one per country)"), db: Session = Depends(get_db), _: CurrentUser = Depends(require_ingest_or_user), ) -> UploadResponse: """Upload up to 3 Rekapitulacija files, parse with the same column detection as update_sales.py, auto-create new SKUs in dim_products, bulk-insert into erp_transactions, and refresh v_sales_weekly_full.""" if not files: raise HTTPException(status_code=400, detail="No files supplied") if len(files) > 3: raise HTTPException(status_code=400, detail="At most 3 files allowed") payload: list[tuple[str, bytes]] = [] for f in files: name = f.filename or "upload.xlsx" if not name.lower().endswith((".xlsx", ".xls", ".csv")): raise HTTPException(status_code=400, detail=f"File '{name}' must be .xlsx, .xls or .csv") data = await f.read() if not data: raise HTTPException(status_code=400, detail=f"File '{name}' is empty") payload.append((name, data)) result = UploadService(db).process_weekly_update(payload) return UploadResponse( rows_added=result["rows_added"], n_files=result["n_files"], new_products=result["new_products"], new_partners=result["new_partners"], new_stores=result["new_stores"], warnings=result["warnings"], view_refreshed=result["view_refreshed"], date_range=result["date_range"], ) # ====================================================================== # Stock upload — 1-3 Excel files (HR / SI / AT) → refresh erp_stock_current # ====================================================================== @router.post("/upload-stock", response_model=StockUploadResponse) async def upload_stock( hr_file: Optional[UploadFile] = File(None), si_file: Optional[UploadFile] = File(None), at_file: Optional[UploadFile] = File(None), db: Session = Depends(get_db), _: CurrentUser = Depends(require_ingest_or_user), ) -> StockUploadResponse: """Stock snapshot upload — one file per country (HR/SI/AT). HR file is expected to carry a per-location column ('Jedin.' / 'Naziv odjela'); rows where the location code is '01' / '1' / 'Warehouse' are written to data/stock.csv (central WH), everything else is summed into data/stock_stores.csv. SI / AT are aggregated into stock_stores_slo.csv / stock_stores_at.csv. Any country left blank keeps its existing CSV (partial uploads are safe). After CSVs are written, the existing _load_stock_current loader truncates erp_stock_current and re-inserts from all four files. """ payload: list[tuple[str, str, bytes]] = [] for tag, f in (("HR", hr_file), ("SI", si_file), ("AT", at_file)): if f is None: continue name = f.filename or f"{tag.lower()}.xlsx" if not name.lower().endswith((".xlsx", ".xls", ".csv")): raise HTTPException( status_code=400, detail=f"{tag} file '{name}' must be .xlsx, .xls or .csv", ) data = await f.read() if not data: raise HTTPException(status_code=400, detail=f"{tag} file is empty") payload.append((tag, name, data)) if not payload: raise HTTPException(status_code=400, detail="No files supplied") result = process_stock_upload(db, payload) return StockUploadResponse(**result) # ====================================================================== # Run forecast (wraps forecast_engine.py) # ====================================================================== @router.get("/forecast-prereqs", response_model=ForecastPrereqsResponse) def get_forecast_prereqs(db: Session = Depends(get_db)) -> ForecastPrereqsResponse: """Pre-flight: is the workbook present? sales_clean.csv? forecast_engine.py? Frontend calls this before showing the Run buttons so the planner sees upfront what's missing rather than waiting 15 minutes for a failure.""" return ForecastService.check_prereqs() # type: ignore[return-value] @router.post("/run-forecast", response_model=RunForecastResponse) def run_forecast( payload: RunForecastRequest, db: Session = Depends(get_db), ) -> RunForecastResponse: """Synchronously run forecast_engine.py as a subprocess. Takes 10–15 minutes for the full SKU set. After the engine exits, ingests data/forecast_log.csv into the forecasts table (filtered to the latest run_id) and data/factor_history.csv into the factor_history table (filtered to current ISO year+week). Always returns HTTP 200 — caller inspects `success` flag. On engine failure, `stdout_tail` carries the last 20 lines of engine output for diagnosis. Frontend should call GET /forecast-prereqs first and set a client timeout ≥ 25 min.""" if payload.run_type not in ("baseline", "with_ontop"): raise HTTPException(status_code=400, detail="run_type must be 'baseline' or 'with_ontop'") return ForecastService(db).run_forecast( # type: ignore[return-value] run_type=payload.run_type, run_by_id=payload.run_by_id, ) @router.get("/forecast-runs", response_model=list[ForecastRunRow]) def list_forecast_runs( limit: int = Query(20, ge=1, le=100), db: Session = Depends(get_db), ) -> list[ForecastRunRow]: """Recent forecast runs — populates the history pane on the RunForecast page.""" return ForecastService(db).get_recent_runs(limit=limit) # ====================================================================== # Plan download / upload — round-trip planner corrections through Excel # ====================================================================== @router.get("/download-plan") def download_plan( run_id: Optional[int] = Query(None, description="forecast_runs.id, defaults to latest"), ) -> StreamingResponse: """Generate Polleo_Demand_Plan-style xlsx from the forecasts table. The xlsx contains a single 'Demand Planning' sheet with: - per-SKU forecast columns (one per CW in the horizon) - planner_factor (editable) - on_top_vp / on_top_mp (editable) - total (sum of CW columns) Planners download, edit, re-upload via POST /upload-plan. Round-trip preserves the run_id via a hidden _meta sheet.""" import io try: data, resolved_run_id = ForecastService.generate_plan_xlsx(run_id=run_id) except Exception as exc: raise HTTPException(status_code=400, detail=str(exc)) # Pull cw_label for filename from datetime import datetime iso = datetime.now().isocalendar() cw = int(iso[1]) fname = f"Polleo_Demand_Plan_CW{cw:02d}_run{resolved_run_id}.xlsx" return StreamingResponse( io.BytesIO(data), media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", headers={"Content-Disposition": f'attachment; filename="{fname}"'}, ) @router.post("/upload-plan") async def upload_plan( file: UploadFile = File(..., description="Edited Polleo_Demand_Plan.xlsx"), db: Session = Depends(get_db), ) -> dict: """Parse an edited plan xlsx and apply planner_factor / on_top changes back into the forecasts table. The xlsx must have been generated by GET /download-plan so it carries the run_id in its _meta sheet — otherwise we apply changes to the latest run with a warning.""" if not file.filename or not file.filename.lower().endswith((".xlsx", ".xls")): raise HTTPException(status_code=400, detail="Expected .xlsx file") data = await file.read() if not data: raise HTTPException(status_code=400, detail="Uploaded file is empty") try: result = ForecastService.apply_plan_upload(data) return result except Exception as exc: raise HTTPException(status_code=400, detail=str(exc)) # ── KAM / CM input ──────────────────────────────────────────────────────── @router.get("/input-template", response_model=InputTemplateResponse) def get_input_template( user_id: int = Query(..., description="User ID (temporary until JWT auth)"), db: Session = Depends(get_db), ): """Return the on-top input grid for a KAM (wholesale) or CM (retail) user. VP/KAM users receive all planning SKUs grouped by their buyer list. MP/CM users receive only the SKUs from their assigned categories. Each SKU row includes baseline forecast reference + prior submission.""" try: svc = DemandService(db) data = svc.get_kam_template(user_id) return data except ValueError as exc: raise HTTPException(status_code=404, detail=str(exc)) except Exception as exc: raise HTTPException(status_code=500, detail=str(exc)) @router.post("/inputs", response_model=SaveInputsResponse) def save_inputs( user_id: int = Query(..., description="User ID (temporary until JWT auth)"), body: SaveInputsRequest = Body(...), db: Session = Depends(get_db), ): """Legacy full-replace save. Kept for backward compatibility — the new wizard uses POST /inputs/v2 (audit + lock-window aware).""" try: svc = DemandService(db) result = svc.save_kam_inputs(user_id, [i.model_dump() for i in body.inputs]) return result except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) except Exception as exc: raise HTTPException(status_code=500, detail=str(exc)) # ─── KAM/CM wizard v2 (audit + lock window) ────────────────────────────── @router.get("/kam-input/users", response_model=list[InputUserRow]) def list_input_users_v2( db: Session = Depends(get_db), user: CurrentUser = Depends(get_current_user), ) -> list[InputUserRow]: """Step-1 dropdown. Admin sees every VP+MP user; non-admin sees only themselves (and only if they have role VP or MP).""" is_admin = (user.role or "").lower() == "admin" return DemandService(db).list_kam_cm_users( requester_id=user.user_id, requester_is_admin=is_admin, ) @router.get("/kam-input/buyers", response_model=list[str]) def list_user_buyers_v2( user_id: int = Query(..., description="Whose past buyers to list"), db: Session = Depends(get_db), _: CurrentUser = Depends(get_current_user), ) -> list[str]: """Step-2 dropdown for KAMs. Returns distinct buyer names the given user has submitted on-top inputs for in the past.""" return DemandService(db).list_user_buyers(user_id) @router.post("/inputs/v2", response_model=SaveInputsV2Response) def save_inputs_v2( body: SaveInputsV2Request = Body(...), db: Session = Depends(get_db), user: CurrentUser = Depends(get_current_user), ) -> SaveInputsV2Response: """Audit-aware save for the KAM/CM wizard. Diffs against the existing rows in (cycle, user, buyer) scope, enforces the 4-week lock window (cur_yw through cur_yw+3 inclusive), and writes one on_top_changes row per delta.""" try: # Non-admin users can only submit as themselves — silently rewrite # to prevent UI tampering. is_admin = (user.role or "").lower() == "admin" acting_as = None target_user = body.user_id if not is_admin and target_user != user.user_id: target_user = user.user_id elif is_admin and body.user_id != user.user_id: acting_as = user.user_id # log who actually did it result = DemandService(db).save_kam_inputs_v2( user_id=target_user, buyer=body.buyer, inputs=[i.model_dump() for i in body.inputs], acting_as_id=acting_as, ) return result except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) # ─── Wholesale buyer-listings input (reg-increase / on-top per listed SKU) ── @router.get("/wholesale-input/buyers", response_model=list[str]) def list_wholesale_input_buyers( user_id: int = Query(..., description="KAM user id"), db: Session = Depends(get_db), _: CurrentUser = Depends(get_current_user), ) -> list[str]: """Buyers with a listed assortment under this KAM (from wholesale_listings).""" try: return DemandService(db).list_wholesale_input_buyers(user_id) except ValueError as exc: raise HTTPException(status_code=404, detail=str(exc)) @router.get("/wholesale-input/template", response_model=WholesaleInputTemplate) def get_wholesale_input_template( user_id: int = Query(..., description="KAM user id"), buyer: str = Query(..., description="Buyer name (as listed)"), db: Session = Depends(get_db), _: CurrentUser = Depends(get_current_user), ) -> WholesaleInputTemplate: """Grid of the SKUs listed for (KAM, buyer), with existing regular-increase and on-top portions pre-filled per week.""" try: return DemandService(db).get_wholesale_input_template(user_id, buyer) except ValueError as exc: raise HTTPException(status_code=404, detail=str(exc)) @router.post("/wholesale-input/save", response_model=SaveWholesaleInputResponse) def save_wholesale_input( body: SaveWholesaleInputRequest = Body(...), db: Session = Depends(get_db), user: CurrentUser = Depends(get_current_user), ) -> SaveWholesaleInputResponse: """Save one portion (regular-increase or on-top) of the buyer-listings grid and recompute the affected wholesale forecast cells immediately.""" try: is_admin = (user.role or "").lower() == "admin" acting_as = None target_user = body.user_id if not is_admin and target_user != user.user_id: target_user = user.user_id elif is_admin and body.user_id != user.user_id: acting_as = user.user_id return DemandService(db).save_wholesale_input( user_id=target_user, buyer=body.buyer, portion=body.portion, inputs=[i.model_dump() for i in body.inputs], acting_as_id=acting_as, ) except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) @router.post("/wholesale-listings/upload", response_model=BuyerListingUploadResponse) async def upload_buyer_listings( file: UploadFile = File(..., description="Excel of the buyer's listed SKUs"), kam: str = Form(...), buyer: str = Form(...), dry_run: bool = Form(False, description="Preview only — parse + match, don't write"), db: Session = Depends(get_db), _: CurrentUser = Depends(get_current_user), ) -> BuyerListingUploadResponse: """Parse a buyer's listed-assortment Excel and (unless dry_run) replace that (KAM, buyer)'s rows in wholesale_listings — per-buyer, no truncate.""" name = file.filename or "upload.xlsx" if not name.lower().endswith((".xlsx", ".xls")): raise HTTPException(status_code=400, detail=f"'{name}' must be .xlsx or .xls") data = await file.read() if not data: raise HTTPException(status_code=400, detail="File is empty") try: return DemandService(db).upload_buyer_listings( kam=kam, buyer=buyer, file_bytes=data, filename=name, dry_run=dry_run, ) except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) @router.post("/kam-upload/parse", response_model=KamUploadParseResponse) async def parse_kam_upload( file: UploadFile = File(..., description="Weekly KAM/CM DEMAND INPUT template"), user_id: int = Form(..., description="Selected KAM/CM user id"), db: Session = Depends(get_db), _: CurrentUser = Depends(get_current_user), ) -> KamUploadParseResponse: """Parse a weekly template and return detected sheets + suggested canonical buyers (for the mapping screen). No writes.""" name = file.filename or "upload.xlsx" if not name.lower().endswith((".xlsx", ".xls")): raise HTTPException(status_code=400, detail=f"'{name}' must be .xlsx or .xls") data = await file.read() if not data: raise HTTPException(status_code=400, detail="File is empty") try: return DemandService(db).parse_kam_upload(kam_user_id=user_id, file_bytes=data) except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) @router.post("/kam-upload/apply", response_model=KamUploadApplyResponse) async def apply_kam_upload( file: UploadFile = File(...), user_id: int = Form(...), mapping: str = Form("{}", description="JSON {sheet_name: canonical_buyer}"), seed_listings: str = Form("[]", description="JSON [sheet_name, ...] to seed listing from"), dry_run: bool = Form(False), db: Session = Depends(get_db), user: CurrentUser = Depends(get_current_user), ) -> KamUploadApplyResponse: """Apply a weekly template: diff vs current state → write deltas under the 4-week lock → recompute the forecast. Returns the per-sheet review.""" import json name = file.filename or "upload.xlsx" if not name.lower().endswith((".xlsx", ".xls")): raise HTTPException(status_code=400, detail=f"'{name}' must be .xlsx or .xls") data = await file.read() if not data: raise HTTPException(status_code=400, detail="File is empty") try: mapping_d = json.loads(mapping or "{}") seed_l = json.loads(seed_listings or "[]") except json.JSONDecodeError as exc: raise HTTPException(status_code=400, detail=f"Bad mapping/seed JSON: {exc}") is_admin = (user.role or "").lower() == "admin" acting_as = None target_user = user_id if not is_admin and target_user != user.user_id: target_user = user.user_id elif is_admin and user_id != user.user_id: acting_as = user.user_id try: return DemandService(db).apply_kam_upload( kam_user_id=target_user, file_bytes=data, mapping=mapping_d, seed_listings=seed_l, acting_as_id=acting_as, dry_run=dry_run, ) except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) @router.get("/admin/on-top-changes", response_model=OnTopChangesResponse) def get_on_top_changes( user_id: Optional[int] = Query(None, description="Filter by submitter"), from_yw: Optional[int] = Query(None, description="ISO yw, inclusive"), to_yw: Optional[int] = Query(None, description="ISO yw, inclusive"), weeks_back: Optional[int] = Query(4, description="Only edits made in the last N weeks; 0 = all history"), limit: int = Query(500, ge=1, le=5000), db: Session = Depends(get_db), _: CurrentUser = Depends(require_section("on_top_changes_audit")), ) -> OnTopChangesResponse: """Per-KAM change-control feed for on-top inputs: what they added, removed, re-quantified, or moved between weeks. Delete+insert pairs from one save collapse into a single 'move' row. Defaults to edits made in the last 4 weeks. Gated by the on_top_changes_audit section (Admin + Uprava).""" rows = DemandService(db).get_on_top_changes( user_id=user_id, from_yw=from_yw, to_yw=to_yw, weeks_back=weeks_back, limit=limit, ) # Coerce timestamp to str for the schema for r in rows: ts = r.get("changed_at") if ts is not None: r["changed_at"] = ts.isoformat() if hasattr(ts, "isoformat") else str(ts) return {"rows": [OnTopChangeRow(**r) for r in rows]} @router.get("/forecast-accuracy/kam-buyer", response_model=KAMBuyerFAResponse) def get_kam_buyer_fa( tier: list[str] = Query(default=[]), xyz: list[str] = Query(default=[]), category: list[str] = Query(default=[]), date_from: Optional[int] = Query(None, description="YYYYWW lower bound"), date_to: Optional[int] = Query(None, description="YYYYWW upper bound"), db: Session = Depends(get_db), ): """Buyer-level FA for VP/KAM wholesale on-top inputs. Matches on_top_inputs buyer names to ERP partners via ILIKE and computes per-week-average FA comparing commitments against erp_transactions actuals. Only covers wholesale (VP) — CM retail has no buyer-level tracking.""" svc = DemandService(db) return svc.get_kam_buyer_fa( tier=tier or None, xyz=xyz or None, category=category or None, date_from=date_from, date_to=date_to, ) @router.get("/input-status", response_model=SubmissionStatusResponse) def get_input_status( cycle_id: Optional[int] = Query(None, description="SOP cycle ID; defaults to active cycle"), db: Session = Depends(get_db), ): """Submission status for all VP/MP users in the given (or active) cycle. Used by the demand planner to see who has and hasn't submitted.""" svc = DemandService(db) return svc.get_submission_overview(cycle_id)