"""Demand Review — the S&OP consensus-forecast sign-off surface. This is the page where Demand Planning aligns on the unconstrained number that goes into Supply Review. It is intentionally a DIFFERENT page from Wholesale Review (KAM commit control), Revenue Forecast (number itself), and Forecast Accuracy (retrospective look) — it puts those threads on the same page in the form an S&OP demand step needs. Blocks the page exposes (each can be built incrementally): headline — week-by-week (rolling 13w): this cycle's forecast vs prior cycle's forecast for the same week vs PY actuals. Shows revision direction. decomposition — week-by-week: stat baseline + planner factor adjustment + on-top WS commits + on-top MP commits + promo uplift = consensus. Reveals what drives the headline number. drivers — promo calendar bands + NPD launches overlaid on horizon. Qualitative context for the quantitative blocks above. ro_notes — persistent risks/opportunities/decisions per S&OP cycle. signoff — reviewer audit trail (records, doesn't lock plan — that stays on Consensus Plan). Phase 1 (this revision) only populates `decomposition`. The other keys are returned as `null` so the frontend shell can render placeholders without needing a separate endpoint per block. """ from __future__ import annotations from datetime import date, timedelta from typing import Optional import pandas as pd from sqlalchemy import text from sqlalchemy.orm import Session HORIZON_WEEKS = 13 def _horizon(today: Optional[date] = None) -> list[tuple[int, int]]: """Rolling 13-week horizon starting from the CURRENT ISO week (today's week is included as the first slice — partial actuals are OK for the decomposition view since we're showing forecast components, not actuals).""" today = today or date.today() iso = today.isocalendar() out: list[tuple[int, int]] = [] cur = today - timedelta(days=today.weekday()) # Monday of this week for _ in range(HORIZON_WEEKS): i = cur.isocalendar() out.append((int(i[0]), int(i[1]))) cur += timedelta(days=7) return out def _current_cycle(db: Session) -> dict: """Return a thin summary of the latest sop_cycles row — id, label. Used to header the page. Falls back to a synthetic entry when no cycles exist yet. sop_cycles stores the kick-off (year, week) as a single packed `year_week` int (e.g. 202622 = CW22/2026).""" row = db.execute(text(""" SELECT id, year_week, status, started_at, closed_at FROM sop_cycles ORDER BY id DESC LIMIT 1 """)).mappings().first() iso_now = date.today().isocalendar() if not row: return {"id": None, "label": f"Ad-hoc (CW{int(iso_now[1]):02d}/{int(iso_now[0])})", "year": int(iso_now[0]), "week": int(iso_now[1]), "status": None, "started_at": None} yw = int(row["year_week"]) year, week = yw // 100, yw % 100 return { "id": int(row["id"]), "label": f"CW{week:02d}/{year}", "year": year, "week": week, "status": row.get("status"), "started_at": str(row["started_at"]) if row.get("started_at") else None, } def _decomposition(db: Session, horizon: list[tuple[int, int]]) -> dict: """Per-week breakdown of the consensus forecast into its components. For each week in the 13-week horizon we compute (across all planned SKUs) the qty contribution of: baseline_qty = SUM(baseline) — raw model output planner_lift_qty = SUM(baseline × (planner_factor − 1)) — what the planner's factor moved the engine by on_top_ws_qty = SUM(on_top_wholesale) — KAM commits on_top_mp_qty = SUM(on_top_retail) — CM commits promo_uplift_qty = SUM(promo_uplift) — engine promo lift consensus_qty = baseline + planner_lift + on_top_ws + on_top_mp + promo_uplift (= SUM(total) from the forecasts table, by construction) Each component also surfaces an € version, valued at the SKU's `erp_prices.avg_sell_price` (channel-blended) so the page can show the revenue-side waterfall at the same granularity. Latest run wins per (product_id, year, week) via DISTINCT ON — same pattern Revenue Forecast and monthly_plan_service use. """ if not horizon: return {"rows": [], "totals": {}} yws = [y * 100 + w for (y, w) in horizon] df = pd.read_sql(text(""" WITH latest_fc AS ( SELECT DISTINCT ON (product_id, year, week) product_id, year, week, COALESCE(baseline, 0)::float AS baseline, COALESCE(planner_factor, 1)::float AS planner_factor, COALESCE(on_top_wholesale, 0)::float AS on_top_ws, COALESCE(on_top_retail, 0)::float AS on_top_mp, COALESCE(promo_uplift, 0)::float AS promo_uplift, COALESCE(total, 0)::float AS total FROM forecasts WHERE (year * 100 + week) = ANY(:yws) ORDER BY product_id, year, week, run_id DESC ) SELECT f.product_id, f.year, f.week, f.baseline, f.planner_factor, f.on_top_ws, f.on_top_mp, f.promo_uplift, f.total, COALESCE(ep.avg_sell_price, 0)::float AS price FROM latest_fc f LEFT JOIN erp_prices ep ON ep.product_id = f.product_id """), db.bind, params={"yws": yws}) if df.empty: return {"rows": [{ "year": y, "week": w, "cw_label": f"CW{w:02d}", "baseline_qty": 0.0, "baseline_eur": 0.0, "planner_lift_qty": 0.0, "planner_lift_eur": 0.0, "on_top_ws_qty": 0.0, "on_top_ws_eur": 0.0, "on_top_mp_qty": 0.0, "on_top_mp_eur": 0.0, "promo_uplift_qty": 0.0,"promo_uplift_eur": 0.0, "consensus_qty": 0.0, "consensus_eur": 0.0, } for (y, w) in horizon], "totals": {}} # Per-row component qty derivations df["planner_lift_qty"] = df["baseline"] * (df["planner_factor"] - 1.0) df["consensus_qty"] = df["total"] # Valuation: each component × the SKU's avg sell price. avg_sell_price # is a blended (retail+webshop+wholesale) realized rate — pragmatic # single-number proxy here. A channel-aware valuation could split # by ws_share but the decomposition is about *forecast components*, # not channel mix, so the simpler blend keeps the chart readable. for col in ("baseline", "planner_lift_qty", "on_top_ws", "on_top_mp", "promo_uplift", "consensus_qty"): df[f"{col}_eur"] = df[col] * df["price"] # Aggregate per (year, week) agg = df.groupby(["year", "week"], as_index=False).agg( baseline_qty =("baseline", "sum"), baseline_eur =("baseline_eur", "sum"), planner_lift_qty =("planner_lift_qty", "sum"), planner_lift_eur =("planner_lift_qty_eur", "sum"), on_top_ws_qty =("on_top_ws", "sum"), on_top_ws_eur =("on_top_ws_eur", "sum"), on_top_mp_qty =("on_top_mp", "sum"), on_top_mp_eur =("on_top_mp_eur", "sum"), promo_uplift_qty =("promo_uplift", "sum"), promo_uplift_eur =("promo_uplift_eur", "sum"), consensus_qty =("consensus_qty", "sum"), consensus_eur =("consensus_qty_eur", "sum"), ) # Reindex by horizon to ensure every week appears, even with zeros horizon_df = pd.DataFrame(horizon, columns=["year", "week"]) out = horizon_df.merge(agg, on=["year", "week"], how="left").fillna(0) out["cw_label"] = out["week"].apply(lambda w: f"CW{int(w):02d}") rows = out.round(2).to_dict("records") totals_cols = [c for c in out.columns if c not in ("year", "week", "cw_label")] totals = {c: float(out[c].sum()) for c in totals_cols} return {"rows": rows, "totals": {k: round(v, 2) for k, v in totals.items()}} def report_demand_review(db: Session) -> dict: """Top-level assembly for the Demand Review page. Phase 1: only `decomposition` is populated. Other blocks return null so the frontend can render placeholders and signal "coming soon" without needing a separate endpoint per block. """ cycle = _current_cycle(db) horiz = _horizon() return { "cycle": cycle, "horizon": { "weeks": [f"CW{w:02d}" for (_, w) in horiz], "year_week_keys": [y * 100 + w for (y, w) in horiz], }, "headline": None, "decomposition": _decomposition(db, horiz), "drivers": None, # ro_notes is folded into the overview payload so the page can # render the full Demand Review with a single endpoint hit. The # notes CRUD endpoints handle add/resolve/delete; this surface # is read-through. "ro_notes": list_notes(db), "signoff": None, } # ───────────────────────────────────────────────────────────────────── # Risks & Opportunities — persistent log # ───────────────────────────────────────────────────────────────────── # Notes are keyed to the sop_cycle they were raised in, but unresolved # items show up on EVERY subsequent cycle's review page until explicitly # marked done. That carryover is what turns the log into a "promises" # tracker — risks don't disappear just because the calendar flipped. VALID_KINDS = {"RISK", "OPPORTUNITY", "DECISION"} VALID_SEVERITIES = {"HIGH", "MED", "LOW"} def _current_cycle_id(db: Session) -> Optional[int]: row = db.execute(text( "SELECT id FROM sop_cycles ORDER BY id DESC LIMIT 1" )).first() return int(row[0]) if row else None def list_notes(db: Session) -> dict: """Return the page's notes payload split by kind. Current-cycle notes + carryover notes (any unresolved row from a PRIOR cycle) are returned together. `is_carryover=True` flags the latter so the UI can render a small chip with the originating cycle. """ cur = _current_cycle_id(db) if cur is None: return {"current_cycle_id": None, "risks": [], "opportunities": [], "decisions": []} rows = db.execute(text(""" SELECT n.id, n.cycle_id, n.kind, n.severity, n.note, n.resolved, n.resolved_at, n.resolved_by, COALESCE(NULLIF(ru.display_name, ''), ru.username) AS resolved_by_name, n.author_id, COALESCE(NULLIF(au.display_name, ''), au.username, '(unknown)') AS author_name, n.created_at, sc.year_week AS cycle_year_week FROM demand_review_notes n LEFT JOIN users au ON au.id = n.author_id LEFT JOIN users ru ON ru.id = n.resolved_by LEFT JOIN sop_cycles sc ON sc.id = n.cycle_id WHERE n.cycle_id = :cur OR n.resolved = FALSE ORDER BY n.resolved ASC, n.created_at DESC """), {"cur": cur}).mappings().all() def _row(r: dict) -> dict: yw = int(r["cycle_year_week"]) if r.get("cycle_year_week") else None cycle_label = f"CW{yw % 100:02d}/{yw // 100}" if yw else "?" return { "id": int(r["id"]), "cycle_id": int(r["cycle_id"]), "cycle_label": cycle_label, "is_carryover": int(r["cycle_id"]) != cur, "kind": str(r["kind"]), "severity": r.get("severity"), "note": str(r["note"]), "resolved": bool(r["resolved"]), "resolved_at": str(r["resolved_at"]) if r.get("resolved_at") else None, "resolved_by_id": int(r["resolved_by"]) if r.get("resolved_by") else None, "resolved_by_name": r.get("resolved_by_name"), "author_id": int(r["author_id"]), "author_name": str(r["author_name"]), "created_at": str(r["created_at"]), } enriched = [_row(dict(r)) for r in rows] risks = [r for r in enriched if r["kind"] == "RISK"] opportunities = [r for r in enriched if r["kind"] == "OPPORTUNITY"] decisions = [r for r in enriched if r["kind"] == "DECISION"] return { "current_cycle_id": cur, "risks": risks, "opportunities": opportunities, "decisions": decisions, } def create_note(db: Session, *, user_id: int, kind: str, severity: Optional[str], note: str) -> dict: """Insert a new note against the current S&OP cycle.""" kind = (kind or "").upper().strip() if kind not in VALID_KINDS: raise ValueError(f"kind must be one of {sorted(VALID_KINDS)}") sev = (severity or "").upper().strip() or None if sev is not None and sev not in VALID_SEVERITIES: raise ValueError(f"severity must be one of {sorted(VALID_SEVERITIES)} or omitted") note = (note or "").strip() if not note: raise ValueError("note text is required") cur = _current_cycle_id(db) if cur is None: raise ValueError("no active sop_cycles row — start a cycle first") new_id = db.execute(text(""" INSERT INTO demand_review_notes (cycle_id, kind, severity, note, author_id) VALUES (:cycle_id, :kind, :sev, :note, :author) RETURNING id """), {"cycle_id": cur, "kind": kind, "sev": sev, "note": note, "author": user_id}).scalar() db.commit() return {"id": int(new_id), "cycle_id": cur, "kind": kind, "severity": sev, "note": note} def update_note_resolved(db: Session, *, note_id: int, user_id: int, resolved: bool) -> dict: """Toggle the resolved flag. Records who closed it + when so the audit trail survives carryover semantics.""" if resolved: db.execute(text(""" UPDATE demand_review_notes SET resolved = TRUE, resolved_at = now(), resolved_by = :uid WHERE id = :nid """), {"nid": note_id, "uid": user_id}) else: db.execute(text(""" UPDATE demand_review_notes SET resolved = FALSE, resolved_at = NULL, resolved_by = NULL WHERE id = :nid """), {"nid": note_id}) db.commit() return {"id": note_id, "resolved": resolved} def delete_note(db: Session, *, note_id: int, user_id: int, is_admin: bool) -> bool: """Author-or-admin only. Soft policy is enforced in the router layer because it has access to the requesting CurrentUser; this function trusts the caller. Returns True if a row was removed.""" if not is_admin: # When not admin, restrict to author. Single UPDATE-or-DELETE # combo so we don't expose ownership state to the caller. result = db.execute(text(""" DELETE FROM demand_review_notes WHERE id = :nid AND author_id = :uid """), {"nid": note_id, "uid": user_id}) else: result = db.execute(text( "DELETE FROM demand_review_notes WHERE id = :nid" ), {"nid": note_id}) db.commit() return bool(result.rowcount)