"""Demand Plan workbook (v4) — the planner round-trip Excel. Rebuilds the rich legacy "Polleo_Demand_Plan" layout from the v4 `forecasts` table + `on_top_inputs` + actual sales, with four sheets: • Demand Planning — multi-row per SKU (Baseline FC / Planner Factor / Adjusted FC / VP on-top / VP regular / MP on-top / MP regular / On-top Total / TOTAL DEMAND). RR-25..RR-0 = the article's ACTUAL sales in the trailing weeks AND, on the on-top/regular rows, what was actually entered as on-top / regular-increase in those past weeks (per VP/MP channel). CW.. = the forward forecast. Planner Factor (per week) is the editable cell — re-uploaded via parse_plan_workbook + apply_plan_factors. • Forecast Detail — one row per SKU: model + the forward baseline. • Demand Output - Total — one row per SKU: TOTAL forecast per week. • Demand Output - On Top — one row per SKU: on-top per week. A hidden `_meta` sheet carries the run_id so re-upload pins the right run. """ from __future__ import annotations import io from datetime import date, datetime, timedelta from typing import Optional from sqlalchemy import text from backend.models.database import engine _TRAILING = 26 # RR-25 .. RR-0 _RETAIL_CHANNELS = ("retail", "food retail") # Demand Planning per-SKU row labels, in order. _LABELS = [ "Baseline FC", "Planner Factor", "Adjusted FC", "VP on-top", "VP regular", "MP on-top", "MP regular", "On-top Total", "TOTAL DEMAND", ] def _iso(d: date) -> tuple[int, int]: c = d.isocalendar() return int(c[0]), int(c[1]) def _monday(year: int, week: int) -> date: return date.fromisocalendar(year, week, 1) def _trailing_weeks(first_fwd: tuple[int, int], n: int = _TRAILING) -> list[tuple[int, int]]: """The n ISO weeks immediately BEFORE the first forecast week, oldest first (so the list maps to RR-(n-1) .. RR-0).""" mon = _monday(*first_fwd) out = [] for k in range(n, 0, -1): out.append(_iso(mon - timedelta(weeks=k))) return out # ── data load ──────────────────────────────────────────────────────────── def _load(run_id: Optional[int]): with engine.connect() as c: if run_id is None: run_id = c.execute(text( "SELECT MAX(run_id) FROM forecasts")).scalar() if run_id is None: raise RuntimeError("No forecast runs found — run a forecast first") run_id = int(run_id) rows = c.execute(text(""" SELECT f.product_id, p.sku, COALESCE(p.name, p.sku) AS name, COALESCE(cat.name, '—') AS category, COALESCE(sp.tier, '') AS tier, f.year, f.week, COALESCE(f.baseline, 0)::float AS baseline, COALESCE(f.planner_factor, 1)::float AS planner_factor, COALESCE(f.on_top_wholesale, 0)::float AS on_top_ws, COALESCE(f.on_top_retail, 0)::float AS on_top_mp, COALESCE(f.total, 0)::float AS total, COALESCE(f.model_used, '') AS model FROM forecasts f JOIN dim_products p ON p.id = f.product_id LEFT JOIN dim_categories cat ON cat.id = p.category_id LEFT JOIN sku_planning sp ON sp.product_id = f.product_id WHERE f.run_id = :rid ORDER BY p.sku, f.year, f.week """), {"rid": run_id}).mappings().all() if not rows: raise RuntimeError(f"Run {run_id} has no forecast rows") pids = sorted({int(r["product_id"]) for r in rows}) fwd = sorted({(int(r["year"]), int(r["week"])) for r in rows}) trail = _trailing_weeks(fwd[0]) all_yws = [y * 100 + w for (y, w) in (trail + fwd)] # on-top inputs (past AND forward) split into VP/MP × on-top/regular ot = c.execute(text(f""" SELECT product_id, year_week, SUM(CASE WHEN channel='wholesale' THEN GREATEST(COALESCE(quantity,0)-COALESCE(regular_increase_qty,0),0) ELSE 0 END)::float AS vp_ot, SUM(CASE WHEN channel='wholesale' THEN COALESCE(regular_increase_qty,0) ELSE 0 END)::float AS vp_reg, SUM(CASE WHEN channel IN :rc THEN GREATEST(COALESCE(quantity,0)-COALESCE(regular_increase_qty,0),0) ELSE 0 END)::float AS mp_ot, SUM(CASE WHEN channel IN :rc THEN COALESCE(regular_increase_qty,0) ELSE 0 END)::float AS mp_reg FROM on_top_inputs WHERE product_id = ANY(:pids) AND year_week = ANY(:yws) GROUP BY product_id, year_week """), {"pids": pids, "yws": all_yws, "rc": _RETAIL_CHANNELS}).mappings().all() otmap = {(int(r["product_id"]), int(r["year_week"])): (r["vp_ot"], r["vp_reg"], r["mp_ot"], r["mp_reg"]) for r in ot} # actual sales for the trailing weeks sales = {} if trail: srows = c.execute(text(""" SELECT product_id, year, week, COALESCE(qty_total,0)::float AS q FROM v_sales_weekly_full WHERE product_id = ANY(:pids) AND (year*100+week) = ANY(:yws) """), {"pids": pids, "yws": [y * 100 + w for (y, w) in trail]}).mappings().all() sales = {(int(r["product_id"]), int(r["year"]) * 100 + int(r["week"])): r["q"] for r in srows} # group forecast rows per SKU per_sku: dict = {} meta: dict = {} for r in rows: pid = int(r["product_id"]) d = per_sku.setdefault(pid, {}) d[(int(r["year"]), int(r["week"]))] = r meta.setdefault(pid, {"sku": r["sku"], "name": r["name"], "category": r["category"], "tier": r["tier"], "model": r["model"]}) order = sorted(per_sku, key=lambda p: meta[p]["sku"]) return {"run_id": run_id, "fwd": fwd, "trail": trail, "pids": order, "per_sku": per_sku, "meta": meta, "otmap": otmap, "sales": sales} # ── build ────────────────────────────────────────────────────────────────── def build_plan_workbook(run_id: Optional[int] = None, generated_at: Optional[str] = None) -> tuple[bytes, int]: from openpyxl import Workbook from openpyxl.styles import Font, PatternFill, Alignment D = _load(run_id) fwd, trail = D["fwd"], D["trail"] cw_labels = [f"CW{w:02d}" for (_, w) in fwd] rr_labels = [f"RR-{i}" for i in range(len(trail) - 1, -1, -1)] # RR-25..RR-0 ts = generated_at or datetime.now().strftime("%Y-%m-%d %H:%M") from openpyxl.utils import get_column_letter HEAD = Font(bold=True) TITLE = Font(bold=True, size=13, color="1B2A4A") GREY = Font(color="888888") WHITEB = Font(bold=True, color="FFFFFF") RIGHT = Alignment(horizontal="right") # header fills (past vs forward visually distinct, like legacy) H_ID = PatternFill("solid", fgColor="2F5496") # SKU..Label (dark blue) H_RR = PatternFill("solid", fgColor="EDF2F9") # RR cols (light) H_CW = PatternFill("solid", fgColor="1F3864") # CW cols (darker blue) # per-row value fills GREEN = PatternFill("solid", fgColor="E2EFDA") # Baseline / Adjusted PINK = PatternFill("solid", fgColor="FFF0F0") # Planner Factor (editable) AMBER = PatternFill("solid", fgColor="FFF2CC") # on-top rows GREYF = PatternFill("solid", fgColor="F2F2F2") # TOTAL DEMAND BAND = PatternFill("solid", fgColor="D6DCE4") # per-SKU band row ROW_FILL = {"Baseline FC": GREEN, "Adjusted FC": GREEN, "Planner Factor": PINK, "VP on-top": AMBER, "MP on-top": AMBER, "On-top Total": AMBER, "TOTAL DEMAND": GREYF} ROW_BOLD = {"Baseline FC", "Planner Factor", "Adjusted FC", "On-top Total", "TOTAL DEMAND"} wb = Workbook() # ---- Sheet 1: Demand Planning ---- ws = wb.active ws.title = "Demand Planning" ws.append(["DEMAND PLANNING"]); ws["A1"].font = TITLE ws.append([f"Generated {ts} | Engine v4"]); ws["A2"].font = GREY ws.append(["Planner Factor: 1.1 = +10%, 0.9 = -10% (edit per week, then re-upload)"]) ws["A3"].font = GREY ws.append([]) header = ["SKU", "Artikl", "Grupacija", "Oznaka", "Label", *rr_labels, *cw_labels, "Review", "Promo"] ws.append(header) rr_off = 5 # first RR column (0-based) in the row list cw_off = rr_off + len(rr_labels) n_cols = len(header) for ci in range(1, n_cols + 1): c = ws.cell(row=5, column=ci) if ci <= rr_off: c.fill, c.font = H_ID, WHITEB elif ci <= cw_off: c.fill, c.font = H_RR, GREY else: c.fill, c.font = H_CW, WHITEB c.alignment = RIGHT if ci > rr_off else Alignment(horizontal="left") def _vals(pid, label): """Return (rr_values, cw_values) for one label row of one SKU.""" meta = D["meta"][pid]; per = D["per_sku"][pid] rr, cw = [], [] # RR (past) for (y, w) in trail: yw = y * 100 + w ot = D["otmap"].get((pid, yw), (0, 0, 0, 0)) vp_ot, vp_reg, mp_ot, mp_reg = ot if label == "Baseline FC": rr.append(D["sales"].get((pid, yw), 0)) elif label == "VP on-top": rr.append(vp_ot) elif label == "VP regular": rr.append(vp_reg) elif label == "MP on-top": rr.append(mp_ot) elif label == "MP regular": rr.append(mp_reg) elif label == "On-top Total": rr.append(vp_ot + vp_reg + mp_ot + mp_reg) elif label == "TOTAL DEMAND": rr.append(D["sales"].get((pid, yw), 0)) else: rr.append(None) # Planner Factor / Adjusted FC: no past # CW (forward) for (y, w) in fwd: r = per.get((y, w)) yw = y * 100 + w ot = D["otmap"].get((pid, yw), (0, 0, 0, 0)) vp_ot, vp_reg, mp_ot, mp_reg = ot if r is None: cw.append(0); continue if label == "Baseline FC": cw.append(round(r["baseline"])) elif label == "Planner Factor": cw.append(round(r["planner_factor"], 4)) elif label == "Adjusted FC": cw.append(round(r["baseline"] * r["planner_factor"])) elif label == "VP on-top": cw.append(round(vp_ot)) elif label == "VP regular": cw.append(round(vp_reg)) elif label == "MP on-top": cw.append(round(mp_ot)) elif label == "MP regular": cw.append(round(mp_reg)) elif label == "On-top Total": cw.append(round(r["on_top_ws"] + r["on_top_mp"])) elif label == "TOTAL DEMAND": cw.append(round(r["total"])) else: cw.append(None) return rr, cw for pid in D["pids"]: m = D["meta"][pid] # per-SKU band row (groups the 9 label rows visually) ws.append([m["sku"], m["name"], m["category"], m["tier"]]) bidx = ws.max_row for ci in range(1, n_cols + 1): bc = ws.cell(row=bidx, column=ci) bc.fill = BAND bc.font = HEAD for label in _LABELS: rr, cw = _vals(pid, label) row = [m["sku"], m["name"], m["category"], m["tier"], label, *rr, *cw, "", ""] ws.append(row) ridx = ws.max_row fill = ROW_FILL.get(label) bold = label in ROW_BOLD if bold: ws.cell(row=ridx, column=5).font = HEAD for j in range(len(rr) + len(cw)): cell = ws.cell(row=ridx, column=rr_off + 1 + j) cell.alignment = RIGHT if bold: cell.font = HEAD if fill is not None: cell.fill = fill if label == "Planner Factor": cell.number_format = "0.00" ws.append([]) # spacer between SKUs widths = {"A": 14, "B": 38, "C": 20, "D": 10, "E": 16} for col, w in widths.items(): ws.column_dimensions[col].width = w for ci in range(rr_off + 1, cw_off + 1): # RR value cols ws.column_dimensions[get_column_letter(ci)].width = 7 for ci in range(cw_off + 1, cw_off + 1 + len(cw_labels)): # CW value cols ws.column_dimensions[get_column_letter(ci)].width = 9 ws.auto_filter.ref = f"A5:{get_column_letter(n_cols)}{ws.max_row}" # filter by SKU/Grupacija/Oznaka/Label ws.freeze_panes = "F6" # Collapse the past (RR) columns by default so the editable forward (CW) # factors are visible immediately on open; the planner clicks the "+" # outline button to expand the sales / past-on-top history when needed. for ci in range(rr_off + 1, cw_off + 1): # each RR (past) column cd = ws.column_dimensions[get_column_letter(ci)] cd.outline_level = 1 cd.hidden = True ws.sheet_properties.outlinePr.summaryRight = False # +/- button on the LEFT of the group # ---- Sheet 2: Forecast Detail ---- fd = wb.create_sheet("Forecast Detail") fd.append(["SKU", "Name", "Grupacija", "Oznaka", "Model", "Pattern", "Uplift", "Cann%", *cw_labels]) for cell in fd[1]: cell.font = HEAD for pid in D["pids"]: m = D["meta"][pid]; per = D["per_sku"][pid] cw = [round(per[(y, w)]["baseline"]) if (y, w) in per else 0 for (y, w) in fwd] fd.append([m["sku"], m["name"], m["category"], m["tier"], m["model"], "", "", "", *cw]) fd.freeze_panes = "I2" # ---- Sheets 3 & 4: Demand Output Total / On-Top ---- def _output_sheet(title, kind): sh = wb.create_sheet(title) sh.append([title]); sh["A1"].font = TITLE sh.append([]); sh.append([]) sh.append(["SKU", "Artikl", "Grupacija", "OZNAKA", *cw_labels]) for cell in sh[4]: cell.font = HEAD for pid in D["pids"]: m = D["meta"][pid]; per = D["per_sku"][pid] vals = [] for (y, w) in fwd: r = per.get((y, w)) if r is None: vals.append(0) elif kind == "total": vals.append(round(r["total"])) else: vals.append(round(r["on_top_ws"] + r["on_top_mp"])) sh.append([m["sku"], m["name"], m["category"], m["tier"], *vals]) sh.freeze_panes = "E5" return sh _output_sheet("Demand Output - Total", "total") _output_sheet("Demand Output - On Top", "ontop") # ---- hidden meta for round-trip ---- meta_ws = wb.create_sheet("_meta") meta_ws.append(["run_id", D["run_id"]]) meta_ws.append(["fwd_year_weeks", ",".join(f"{y}{w:02d}" for (y, w) in fwd)]) meta_ws.sheet_state = "hidden" bio = io.BytesIO() wb.save(bio) return bio.getvalue(), D["run_id"] # ── parse + apply (round-trip: per-week Planner Factor) ───────────────────── def parse_plan_workbook(file_bytes: bytes) -> dict: """Read the edited Demand Planning sheet → per-SKU per-week planner factors. Returns {run_id, updates: [{sku, factors: {(year,week): factor}}], warnings}.""" from openpyxl import load_workbook warnings: list[str] = [] wb = load_workbook(io.BytesIO(file_bytes), data_only=True, read_only=True) run_id = None fwd_yw: list[int] = [] if "_meta" in wb.sheetnames: for row in wb["_meta"].iter_rows(values_only=True): if not row: continue if row[0] == "run_id" and row[1] is not None: run_id = int(row[1]) elif row[0] == "fwd_year_weeks" and row[1]: fwd_yw = [int(x) for x in str(row[1]).split(",") if x] if "Demand Planning" not in wb.sheetnames: return {"run_id": run_id, "updates": [], "warnings": ["No 'Demand Planning' sheet in upload"]} ws = wb["Demand Planning"] rows = list(ws.iter_rows(values_only=True)) # locate the header row (SKU in col0, 'Label' in col4) hdr_i = next((i for i, r in enumerate(rows) if r and str(r[0]).strip() == "SKU" and len(r) > 4 and str(r[4]).strip() == "Label"), None) if hdr_i is None: return {"run_id": run_id, "updates": [], "warnings": ["Could not find the SKU/Label header row"]} header = rows[hdr_i] # map each column index → year_week for the CW columns col_yw: dict[int, int] = {} for ci, name in enumerate(header): s = str(name).strip() if name is not None else "" if s.startswith("CW"): try: wk = int(s[2:]) except ValueError: continue yw = next((y for y in fwd_yw if y % 100 == wk), None) if yw is not None: col_yw[ci] = yw updates: list[dict] = [] for r in rows[hdr_i + 1:]: if not r or len(r) <= 4: continue if str(r[4]).strip() != "Planner Factor": continue sku = str(r[0]).strip() if not sku or sku == "None": continue factors: dict[tuple[int, int], float] = {} for ci, yw in col_yw.items(): if ci < len(r) and r[ci] is not None: try: f = float(r[ci]) except (TypeError, ValueError): continue factors[(yw // 100, yw % 100)] = f if factors: updates.append({"sku": sku, "factors": factors}) if not fwd_yw: warnings.append("No _meta week map — factors matched by CW label to the run weeks") return {"run_id": run_id, "updates": updates, "warnings": warnings} def apply_plan_factors(run_id: Optional[int], updates: list[dict]) -> dict: """Apply per-(SKU, week) planner factors to `run_id` (latest if None), recomputing total = baseline*factor + on_top_wholesale + on_top_retail.""" n_rows = 0 skus_changed: set[str] = set() factors_changed: list[dict] = [] with engine.begin() as c: if run_id is None: run_id = int(c.execute(text("SELECT MAX(run_id) FROM forecasts")).scalar()) for u in updates: sku = u["sku"] pid = c.execute(text("SELECT id FROM dim_products WHERE upper(sku)=upper(:s)"), {"s": sku}).scalar() if pid is None: continue for (y, w), f in u["factors"].items(): res = c.execute(text(""" UPDATE forecasts SET planner_factor = :f, total = COALESCE(baseline,0) * :f + COALESCE(on_top_wholesale,0) + COALESCE(on_top_retail,0) WHERE run_id = :rid AND product_id = :pid AND year = :y AND week = :w AND COALESCE(planner_factor,1) <> :f """), {"f": f, "rid": run_id, "pid": int(pid), "y": y, "w": w}) if res.rowcount: n_rows += res.rowcount skus_changed.add(sku) for s in sorted(skus_changed)[:200]: factors_changed.append({"sku": s}) return {"run_id": run_id, "n_skus_updated": len(skus_changed), "n_rows_updated": n_rows, "factors_changed": factors_changed}