""" Build a standalone CM Action โ€” Past Promotions dashboard. Reads our ERP data, finds past monthly promo campaigns, computes uplift / revenue / RUC per SKU per campaign, and writes a single self-contained HTML file matching the dark Polleo CM Action preview design. The output (`cm_action_history.html`) opens in any browser โ€” no server, no Streamlit. Just double-click. Re-run this script weekly to refresh. Run: python cm_action_history_build.py """ from __future__ import annotations import json from datetime import datetime from pathlib import Path import pandas as pd ROOT = Path(__file__).parent DATA = ROOT / "data" OUT = ROOT / "cm_action_history.html" # ---------- Load data ---------- def _load(name: str) -> pd.DataFrame: p = DATA / name if not p.exists(): return pd.DataFrame() df = pd.read_csv(p) df.columns = [str(c).strip().lower() for c in df.columns] return df erp = _load("erp_promo_calendar.csv") sales = _load("sales_clean.csv") plan = _load("sku_plan_list.csv") prices = _load("sku_prices.csv") costs = _load("sku_costs.csv") if erp.empty or sales.empty: raise SystemExit("Missing erp_promo_calendar.csv or sales_clean.csv in data/. Cannot build.") # Past only iso = datetime.now().isocalendar() cur_y, cur_w = int(iso[0]), int(iso[1]) erp_past = erp[(erp["year"].astype(int) * 100 + erp["week"].astype(int)) < (cur_y * 100 + cur_w)].copy() # Promo type โ€” explode comma-separated; primary = first label erp_past["primary_type"] = erp_past["promo_types"].astype(str).str.split(";").str[0].str.strip() # ---------- Identify campaigns ---------- # A campaign = (primary_type, year). For each campaign, collect SKUs and # the contiguous week range where promo activity was observed. campaigns = [] for (ptype, yr), g in erp_past.groupby(["primary_type", "year"]): weeks = sorted(g["week"].astype(int).unique().tolist()) if not weeks: continue # If gap > 6 weeks, split into separate campaigns blocks = [[weeks[0]]] for w in weeks[1:]: if w - blocks[-1][-1] <= 6: blocks[-1].append(w) else: blocks.append([w]) for blk in blocks: first_w, last_w = blk[0], blk[-1] duration = last_w - first_w + 1 # MONTHLY cadence filter: duration >= 4 weeks OR campaigns whose week # spread is at least 4 weeks (i.e. crosses a month boundary). if duration < 4: continue skus = sorted(g[g["week"].astype(int).isin(blk)]["sku"].unique().tolist()) campaigns.append({ "type": ptype, "year": int(yr), "start_week": first_w, "end_week": last_w, "duration": duration, "skus": skus, }) # Sort: most recent first campaigns.sort(key=lambda c: (-c["year"], -c["end_week"])) # ---------- Compute metrics per campaign ---------- sales["yw"] = sales["year"].astype(int) * 100 + sales["week"].astype(int) prices_map = dict(zip(prices["sku"], prices.get("avg_sell_price", 0))) if "sku" in prices.columns else {} ruc_per_unit_map = {} if not costs.empty and "sku" in costs.columns and "ruc" in costs.columns: ruc_per_unit_map = dict(zip(costs["sku"], costs["ruc"])) name_map = dict(zip(plan.get("sku", []), plan.get("name", []))) if not plan.empty else {} cat_map = dict(zip(plan.get("sku", []), plan.get("cat", ""))) if not plan.empty else {} ozn_map = dict(zip(plan.get("sku", []), plan.get("oznaka", ""))) if not plan.empty else {} CAMPAIGN_DATA = [] for c in campaigns: yr = c["year"]; sw = c["start_week"]; ew = c["end_week"] promo_yws = {yr * 100 + w for w in range(sw, ew + 1)} # Base = the 4 weeks preceding the campaign in the same year base_yws = {yr * 100 + w for w in range(max(1, sw - 4), sw)} sub_during = sales[sales["sku"].isin(c["skus"]) & sales["yw"].isin(promo_yws)] sub_base = sales[sales["sku"].isin(c["skus"]) & sales["yw"].isin(base_yws)] # Per SKU stats sku_rows = [] for sku in c["skus"]: d_rows = sub_during[sub_during["sku"] == sku] b_rows = sub_base[sub_base["sku"] == sku] if len(d_rows) == 0: continue d_qty = float(d_rows["qty_total"].sum()) d_n = max(1, d_rows["yw"].nunique()) b_qty = float(b_rows["qty_total"].sum()) b_n = max(1, b_rows["yw"].nunique()) if len(b_rows) else 1 avg_d = d_qty / d_n avg_b = b_qty / b_n if b_qty > 0 else 0 uplift = (avg_d / avg_b) if avg_b > 0 else 0 price = float(prices_map.get(sku, 0) or 0) ruc_pu = float(ruc_per_unit_map.get(sku, 0) or 0) rev = d_qty * price ruc = d_qty * ruc_pu sku_rows.append({ "sku": sku, "name": (name_map.get(sku, "") or "")[:60], "cat": cat_map.get(sku, "") or "", "tier": ozn_map.get(sku, "") or "", "weeks": int(d_rows["yw"].nunique()), "qty": int(d_qty), "avg_base": round(avg_b, 1), "avg_promo": round(avg_d, 1), "uplift": round(uplift, 2), "rev": round(rev), "ruc": round(ruc), }) if not sku_rows: continue sku_rows.sort(key=lambda r: r["qty"], reverse=True) total_qty = sum(r["qty"] for r in sku_rows) total_rev = sum(r["rev"] for r in sku_rows) total_ruc = sum(r["ruc"] for r in sku_rows) avg_uplift = round(sum(r["uplift"] for r in sku_rows if r["uplift"] > 0) / max(1, sum(1 for r in sku_rows if r["uplift"] > 0)), 2) CAMPAIGN_DATA.append({ "id": f"{yr}-{sw:02d}-{c['type'][:30]}", "type": c["type"], "year": yr, "start_week": sw, "end_week": ew, "duration": c["duration"], "n_skus": len(sku_rows), "total_qty": total_qty, "total_rev": total_rev, "total_ruc": total_ruc, "avg_uplift": avg_uplift, "skus": sku_rows, }) # ---------- Build HTML ---------- print(f"Built {len(CAMPAIGN_DATA)} campaigns. Writing HTML...") generated = datetime.now().strftime("%Y-%m-%d %H:%M") HTML = """ CM Action โ€” Past Promotions

๐Ÿ“œ CM Action โ€” Past

Generated __GENERATED__ ยท CW__CURW__/__CURY__
Search campaign
Year
Min duration (weeks)
Campaigns

๐Ÿ“œ CM Action โ€” Past Promotions

Past monthly+ campaigns from the ERP promo calendar ยท uplift, units, revenue, RUC per SKU

๐Ÿ‘ˆ
Pick a campaign on the left to see SKU-level performance.
""" html = (HTML .replace("__GENERATED__", generated) .replace("__CURW__", str(cur_w)) .replace("__CURY__", str(cur_y)) .replace("__DATA_JSON__", json.dumps(CAMPAIGN_DATA, ensure_ascii=False))) OUT.write_text(html, encoding="utf-8") print(f"Wrote: {OUT}") print(f" Campaigns: {len(CAMPAIGN_DATA)}") print(f" File size: {OUT.stat().st_size // 1024} KB") print() print("Open the file directly in your browser:") print(f" file:///{OUT.as_posix()}")