"""Promo Calendar — unified view of promotions across all sources,
with conflict detection, Gantt timeline, source-coverage heatmap and
quick-add / Excel import."""
from __future__ import annotations
import calendar as _cal
from datetime import datetime, timedelta, date
from html import escape as _html_escape
from pathlib import Path as _PPath
import pandas as pd
import plotly.express as px
import plotly.graph_objects as go
import streamlit as st
from promo_data import (
SOURCES, STATUSES, OUTCOMES, TYPES,
SOURCE_TO_DEPT, DEPTS, ROLES, ROLE_DEFAULT_SOURCE,
load_promos, add_promo, update_promo, delete_promo,
detect_conflicts, parse_week_range, horizon_weeks, yw_to_int,
import_excel, append_log,
acknowledge_conflict, exclude_skus_from_promo, load_category_to_skus,
)
st.set_page_config(
page_title="Promo Calendar",
page_icon="📅",
layout="wide",
initial_sidebar_state="expanded",
)
# ---------- Global style polish ----------
st.markdown("""
""", unsafe_allow_html=True)
# ---------- Helpers for the month calendar view ----------
SRC_COLOR = {
"B2C — MP (retail)": "#7C6FEE",
"B2C — WEB": "#E8734A",
"B2B — FMCG": "#34D399",
"B2B — FITNESS": "#38BDF8",
}
HR_MONTHS = ["Siječanj", "Veljača", "Ožujak", "Travanj", "Svibanj", "Lipanj",
"Srpanj", "Kolovoz", "Rujan", "Listopad", "Studeni", "Prosinac"]
HR_DAYS_SHORT = ["Pon", "Uto", "Sri", "Čet", "Pet", "Sub", "Ned"]
def _promo_date_range(r) -> tuple[date, date] | tuple[None, None]:
"""ISO year-week pair → (Monday of start week, Sunday of end week)."""
try:
sy = int(r.get("start_year") or 0)
sw = int(r.get("start_week") or 0)
ey = int(r.get("end_year") or sy)
ew = int(r.get("end_week") or sw)
if not sy or not sw:
return None, None
s = datetime.strptime(f"{sy}-W{sw:02d}-1", "%G-W%V-%u").date()
e = (datetime.strptime(f"{ey}-W{ew:02d}-1", "%G-W%V-%u") + timedelta(days=6)).date()
return s, e
except Exception:
return None, None
@st.cache_data(ttl=20)
def _load_cm_promotion_detail() -> pd.DataFrame:
"""Load PromoTool's cm_promotions.csv (per-SKU breakdown). Returns
empty df if the file doesn't exist."""
p = _PPath(__file__).parent.parent / "PromoTool" / "data" / "cm_promotions.csv"
if not p.exists():
return pd.DataFrame()
try:
df = pd.read_csv(p)
df.columns = [str(c).strip().lower() for c in df.columns]
return df
except Exception:
return pd.DataFrame()
# Multi-language Grupacija normalization → canonical HR. Mirrors the CAT_MAP
# inside update_sales.py so PromoCalendar agrees with the rest of the
# pipeline on what "RTD & SNACKS" or "BIO I SUPERFOODS" mean across SLO/DE
# exports.
_GRUP_CAT_MAP = {
"OBLAČILA IN OBUTEV": "ODJEĆA I OBUĆA",
"BEKLEIDUNG UND SCHUHE": "ODJEĆA I OBUĆA",
"KAMPFSPORT AUSRÜSTUNG": "BORILAČKA OPREMA",
"BORILNA OPREMA": "BORILAČKA OPREMA",
"SPORTNAHRUNG": "SPORTSKA PREHRANA",
"ŠPORTNA PREHRANA": "SPORTSKA PREHRANA",
"BIO IN SUPERFOODS": "BIO I SUPERFOODS",
"BIO UND SUPERFOODS": "BIO I SUPERFOODS",
"PROTEINE": "PROTEINI",
"DRINKWARE IN DOM": "DRINKWARE I HOME",
"DRINKWARE UND HOME": "DRINKWARE I HOME",
"FITNESS EQUIPMENT": "FITNESS OPREMA",
"GADGETS": "GADGETI",
"USLUGE": "OTHER",
"STORITVE": "OTHER",
"DIENSTLEISTUNGEN": "OTHER",
"PARIS": "OTHER",
}
@st.cache_data(ttl=120)
def _load_sku_grupacija_map() -> dict:
"""Load {sku: grupacija_canonical_HR} from sku_subcat_map.csv.
Returns empty dict if the file is missing — Grupacija filter will then
show no options and be a no-op."""
p = _PPath(__file__).parent.parent / "data" / "sku_subcat_map.csv"
if not p.exists():
return {}
try:
df = pd.read_csv(p)
except Exception:
return {}
if "sku" not in df.columns or "grup" not in df.columns:
return {}
out = {}
for sku, g in zip(df["sku"].astype(str), df["grup"].astype(str)):
g = g.strip()
if not g:
continue
out[sku.strip()] = _GRUP_CAT_MAP.get(g, g)
return out
@st.cache_data(ttl=120)
def _load_sku_name_map() -> dict:
"""Load {sku: name} from sku_subcat_map.csv (fallback: sku_category_map.csv)."""
base = _PPath(__file__).parent.parent / "data"
for fname in ("sku_subcat_map.csv", "sku_category_map.csv"):
p = base / fname
if not p.exists():
continue
try:
df = pd.read_csv(p)
except Exception:
continue
if "sku" not in df.columns or "name" not in df.columns:
continue
return {str(s).strip(): str(n).strip()
for s, n in zip(df["sku"], df["name"]) if pd.notna(s)}
return {}
def _promo_grupacije(skus_str, sku_to_grup: dict) -> set:
"""Derive the set of canonical-HR grupacije referenced by a promo's
SKU list (comma-separated). Returns empty set when no SKUs match
or no SKU list is present (category-level promo)."""
if not isinstance(skus_str, str) or not skus_str.strip():
return set()
out = set()
for s in skus_str.split(","):
s = s.strip()
if s and s in sku_to_grup:
out.add(sku_to_grup[s])
return out
@st.cache_data(ttl=120)
def _load_sku_prices() -> dict:
"""Load {sku: price_eur} from main project's sku_prices.csv if available."""
p = _PPath(__file__).parent.parent / "data" / "sku_prices.csv"
if not p.exists():
return {}
try:
df = pd.read_csv(p)
df.columns = [str(c).strip().lower() for c in df.columns]
if "sku" not in df.columns:
return {}
price_col = next((c for c in
["price", "price_eur", "retail_price", "mp_price", "cijena"]
if c in df.columns), None)
if not price_col:
return {}
return dict(zip(df["sku"].astype(str),
pd.to_numeric(df[price_col], errors="coerce").fillna(0)))
except Exception:
return {}
def _synthesize_sku_breakdown(promo: dict) -> pd.DataFrame:
"""Generate deterministic dummy SKU-level projection for promos that
weren't created via Promo Tool (so demo always shows numbers).
Reproducible via hash(promo_id + sku)."""
import hashlib
skus = [x.strip() for x in str(promo.get("skus", "") or "").split(",") if x.strip()]
if not skus:
return pd.DataFrame()
prices = _load_sku_prices()
names = _load_sku_name_map()
total_units = int(promo.get("units", 0) or 0)
base_per = max(1, total_units // len(skus)) if total_units else 0
rows = []
for sku in skus:
seed = int(hashlib.md5(f"{promo.get('id','')}{sku}".encode()).hexdigest()[:8], 16)
price = float(prices.get(sku, 0))
if price <= 0:
price = 5 + (seed % 25) # 5–30 € fallback
if base_per:
jitter = 0.7 + (seed % 100) / 100 * 0.6 # 0.7..1.3
units = int(base_per * jitter)
else:
units = 50 + (seed % 250)
disc = 15 + (seed % 11) # 15..25 %
promo_price = round(price * (1 - disc / 100), 2)
revenue = round(units * promo_price)
margin = 0.18 + (seed % 12) / 100 # 18..30 %
ruc = round(revenue * margin)
rows.append({
"sku": sku, "sku_name": names.get(sku, sku),
"discount_pct": disc, "mechanic": "Discount %",
"total_units": units,
"revenue_eur": revenue, "ruc_eur": ruc,
})
return pd.DataFrame(rows)
# ---------- Header ----------
st.markdown("""
📅 Promo Calendar
Unified view of all promotions across B2C MP, B2C WEB, B2B FMCG, B2B FITNESS — with conflict detection
""", unsafe_allow_html=True)
df = load_promos()
iso = datetime.now().isocalendar()
cur_y, cur_w = int(iso[0]), int(iso[1])
# ---------- Sidebar ----------
with st.sidebar:
st.markdown("### 📅 Promo Calendar")
st.caption(f"CW{cur_w}, {cur_y} · {len(df)} promos in store")
page = st.radio(
"Page",
["📊 Calendar dashboard", "➕ Quick-add", "📥 Import / Export"],
key="page", label_visibility="collapsed",
)
st.divider()
st.markdown("**Statuses**")
status_filter = st.multiselect(
"Statuses", STATUSES, default=STATUSES, key="f_status",
label_visibility="collapsed",
help="Hide promos in unwanted statuses across the whole app.",
)
# Attach dept to df once for downstream use
df["dept"] = df["source"].map(SOURCE_TO_DEPT).fillna("Nabava")
def _apply_filters(df_in: pd.DataFrame, src_filter, status_filter,
grup_filter, sku_filter, sku_to_grup) -> pd.DataFrame:
"""Apply filters. Empty filter = pass-through.
- src_filter / status_filter: simple .isin() on the column
- grup_filter: promo kept if AT LEAST ONE of its SKUs maps to a
selected grupacija. Category-level promos without explicit SKUs
DROP when narrowed (per user direction: strict)
- sku_filter: promo kept if AT LEAST ONE of its SKUs is in the
selected list. Drops category-level promos when narrowed."""
if df_in.empty:
return df_in
out = df_in.copy()
if src_filter:
out = out[out["source"].isin(src_filter)]
if status_filter:
out = out[out["status"].isin(status_filter)]
if grup_filter:
sel = set(grup_filter)
keep = out["skus"].apply(lambda s: bool(_promo_grupacije(s, sku_to_grup) & sel))
out = out[keep]
if sku_filter:
sel_sku = set(sku_filter)
def _has_sku(s):
if not isinstance(s, str) or not s.strip():
return False
return any(tok.strip() in sel_sku for tok in s.split(","))
out = out[out["skus"].apply(_has_sku)]
return out
# ============================================================
# PAGE 1: CALENDAR DASHBOARD
# ============================================================
def _conflict_badge(kind: str) -> str:
if kind == "category_vs_sku":
return "🟠 Category↔SKU"
if kind == "category_vs_category":
return "🟠 Category↔Category"
return "🟠 SKU overlap"
if page.startswith("📊"):
# ---------- Top-of-page filter bar ----------
# 3 dropdowns: Sources / Grupacije / SKU. (Statuses lives in the sidebar
# because it's a hide/show toggle rather than a scoping filter.)
# Visible to all users — everyone running promos needs to see what's
# active and what's in scope.
sku_to_grup = _load_sku_grupacija_map()
sku_to_name = _load_sku_name_map()
grupacije_in_data = sorted({g for g in sku_to_grup.values() if g})
# Build the SKU dropdown options. Restrict to SKUs that actually appear
# in at least one promo — otherwise the dropdown lists 7k SKUs that have
# nothing to match against. Format: "SKU — Name".
skus_in_promos = set()
for s in df["skus"].astype(str):
for tok in s.split(","):
tok = tok.strip()
if tok:
skus_in_promos.add(tok)
sku_options = sorted(
[(sku, sku_to_name.get(sku, "")) for sku in skus_in_promos],
key=lambda t: t[0],
)
sku_label = {sku: (f"{sku} — {name}" if name else sku) for sku, name in sku_options}
fc1, fc2, fc3 = st.columns(3)
src_filter = fc1.multiselect(
"Sources", list(SOURCES.keys()), default=list(SOURCES.keys()),
key="f_src",
)
grup_filter = fc2.multiselect(
"Grupacije", grupacije_in_data, default=grupacije_in_data,
key="f_grup",
help="Higher-level group (RTD & SNACKS, PROTEINI, FITNESS OPREMA, ...). "
"Narrows promos to those whose SKU list contains at least one "
"SKU in the selected grupacija(s). Promos without explicit SKUs "
"drop when this filter is narrowed.",
) if grupacije_in_data else []
sku_filter = fc3.multiselect(
"SKU", [sku for sku, _ in sku_options],
format_func=lambda s: sku_label.get(s, s),
key="f_sku",
help="Filter to promos that contain at least one of the selected "
"SKU(s). Dropdown shows only SKUs currently in use by some "
"promo.",
) if sku_options else []
df_view = _apply_filters(df, src_filter, status_filter,
grup_filter, sku_filter, sku_to_grup)
n_filtered = len(df_view)
n_total = len(df)
if n_filtered < n_total:
st.caption(f"Showing **{n_filtered}** of {n_total} promos after filters.")
st.divider()
# ---------- KPIs ----------
conflicts = detect_conflicts(df_view)
# Department-specific approval queues (Nabava / Marketing tabs) were
# removed — this module is read/edit-only, not a director workflow.
# render_dept_queue() is also gone.
with st.container():
st.caption("Sve aktivacije iz oba odjela na jednom mjestu, s automatskim "
"flagom potencijalnih preklapanja.")
# --- View toggle + dept filter ---
if "anchor_date" not in st.session_state:
st.session_state["anchor_date"] = datetime.now().date()
if "cal_view" not in st.session_state:
st.session_state["cal_view"] = "Mjesec"
vc1, vc2 = st.columns([1.5, 3])
vc1.radio("Pregled", ["Mjesec", "Tjedan"], horizontal=True,
key="cal_view", label_visibility="collapsed")
dept_sel = vc2.multiselect(
"Odjeli (channel)", DEPTS, default=DEPTS,
key="cal_dept_filter", label_visibility="collapsed",
help="Filtriraj prijedloge po odjelu koji je predložio akciju.",
)
# Apply dept filter locally for the unified tab only
if dept_sel and len(df_view):
df_uv = df_view[df_view["dept"].isin(dept_sel)].copy()
else:
df_uv = df_view.iloc[0:0].copy()
view = st.session_state["cal_view"]
# --- Period navigation (shifts by month or by 7 days) ---
def _shift(delta_units: int):
d = st.session_state["anchor_date"]
if st.session_state["cal_view"] == "Mjesec":
m = d.month + delta_units
y = d.year
while m < 1: m += 12; y -= 1
while m > 12: m -= 12; y += 1
last = _cal.monthrange(y, m)[1]
st.session_state["anchor_date"] = date(y, m, min(d.day, last))
else:
st.session_state["anchor_date"] = d + timedelta(days=delta_units * 7)
def _go_today():
st.session_state["anchor_date"] = datetime.now().date()
nav_prev, nav_picker, nav_next, nav_today = st.columns([1, 4, 1, 1])
nav_prev.button("◀", use_container_width=True, key="cal_prev",
on_click=_shift, args=(-1,))
nav_next.button("▶", use_container_width=True, key="cal_next",
on_click=_shift, args=(1,))
nav_today.button("Danas", use_container_width=True, key="cal_today",
on_click=_go_today)
with nav_picker:
anchor = st.session_state["anchor_date"]
if view == "Mjesec":
pc1, pc2 = st.columns(2)
new_m = pc1.selectbox(
"M", list(range(1, 13)),
index=anchor.month - 1,
format_func=lambda i: HR_MONTHS[i - 1],
label_visibility="collapsed", key="cal_m_pick",
)
new_y = pc2.number_input(
"Y", min_value=2020, max_value=2100,
value=int(anchor.year), step=1,
label_visibility="collapsed", key="cal_y_pick",
)
if int(new_m) != anchor.month or int(new_y) != anchor.year:
last = _cal.monthrange(int(new_y), int(new_m))[1]
st.session_state["anchor_date"] = date(
int(new_y), int(new_m), min(anchor.day, last))
st.rerun()
else:
new_d = st.date_input(
"D", value=anchor, label_visibility="collapsed",
key="cal_d_pick",
)
if new_d != anchor:
st.session_state["anchor_date"] = new_d
st.rerun()
anchor = st.session_state["anchor_date"]
if view == "Mjesec":
cal_y, cal_m = anchor.year, anchor.month
period_start = date(cal_y, cal_m, 1)
period_end = date(cal_y, cal_m, _cal.monthrange(cal_y, cal_m)[1])
period_header = f"📅 {HR_MONTHS[cal_m - 1]} {cal_y}"
else:
# Monday of the ISO week containing anchor
period_start = anchor - timedelta(days=anchor.weekday())
period_end = period_start + timedelta(days=6)
iso_y, iso_w, _ = anchor.isocalendar()
period_header = (f"📆 Tjedan {iso_w} · "
f"{period_start.strftime('%d.%m.')}–"
f"{period_end.strftime('%d.%m.%Y')}")
# --- Promos active this period ---
def _touches_period(r):
s, e = _promo_date_range(r)
if s is None:
return False
return not (e < period_start or s > period_end)
df_month = df_uv[df_uv.apply(_touches_period, axis=1)] if len(df_uv) else df_uv
# Reuse old vars for downstream code paths
month_start, month_end = period_start, period_end
# --- Per-day chips index ---
chips_by_day: dict[date, list[dict]] = {}
for _, r in df_month.iterrows():
s, e = _promo_date_range(r)
if s is None:
continue
d = max(s, month_start)
stop = min(e, month_end)
while d <= stop:
chips_by_day.setdefault(d, []).append(r.to_dict())
d += timedelta(days=1)
# --- KPIs (scoped to this month) ---
all_skus = set()
for s in df_month["skus"].astype(str):
for x in s.split(","):
x = x.strip()
if x:
all_skus.add(x)
total_units = int(df_month["units"].astype(float).sum()) if len(df_month) else 0
pending_n = int(df_month["status"].isin(["💡 idea", "🔧 preparing"]).sum()) if len(df_month) else 0
k1, k2, k3, k4, k5 = st.columns(5)
k1.metric("In this period", f"{len(df_month)}")
k2.metric("Pending review", f"{pending_n}")
k3.metric("SKUs in promos", f"{len(all_skus):,}")
k4.metric("Conflicts", f"{len(conflicts)}",
delta="alert" if conflicts else "ok",
delta_color="inverse")
k5.metric("Total promo units", f"{total_units:,}")
st.markdown(
f"{period_header}
",
unsafe_allow_html=True,
)
# ---------- Calendar render (month grid or week strip) ----------
selected_id = st.session_state.get("selected_promo_id", "")
today = datetime.now().date()
def _render_day_cell(d: date, in_period: bool) -> str:
cls = "pc-day"
if not in_period:
cls += " pc-day-other"
if d == today:
cls += " pc-day-today"
day_lbl = d.strftime("%d.%m.") if not in_period else str(d.day)
cell = [f"", f"
{day_lbl}
"]
chip_limit = 4 if view == "Mjesec" else 10
day_chips = chips_by_day.get(d, []) if in_period else []
for r in day_chips[:chip_limit]:
color = SRC_COLOR.get(r["source"], "#6C63FF")
chip_cls = "pc-chip"
status = str(r.get("status", "") or "")
if "idea" in status or "preparing" in status:
chip_cls += " pc-chip-pending"
if selected_id and str(r.get("id", "")) == selected_id:
chip_cls += " pc-chip-selected"
full_name = _html_escape(str(r.get("name", "")))
trunc = 22 if view == "Mjesec" else 36
name = _html_escape(str(r.get("name", ""))[:trunc])
cell.append(
f"
{name}"
)
if len(day_chips) > chip_limit:
cell.append(f"
+{len(day_chips)-chip_limit} more")
cell.append("
"]
for hd in HR_DAYS_SHORT:
cal_html.append(f"
{hd}
")
weeks = _cal.Calendar(firstweekday=0).monthdatescalendar(cal_y, cal_m)
for wk in weeks:
for d in wk:
cal_html.append(_render_day_cell(d, d.month == cal_m))
cal_html.append("
"]
for hd in HR_DAYS_SHORT:
cal_html.append(f"
{hd}
")
d = period_start
for i in range(7):
cal_html.append(_render_day_cell(d, True))
d += timedelta(days=1)
cal_html.append("
"
f"
🎯 {_html_escape(sel['name'])}
"
f"
{_html_escape(date_lbl)} · "
f"{duration_days} dana · {_html_escape(sel['source'])} · "
f"odjel: {sel.get('dept','?')} · "
f"status: {_html_escape(sel.get('status',''))}"
f"
Akcija direktora
",
unsafe_allow_html=True,
)
note_default = (
f"Clash s '{b['name']}' ({b['dept']}) na "
f"{len(conflict_skus)} SKU(s) — molim revidirajte."
)
note_key = f"clash_note_{other_id}"
note = st.text_input(
"Komentar za predlagača (može se preskočiti)",
value=st.session_state.get(note_key, note_default),
key=note_key,
label_visibility="collapsed",
placeholder="Komentar koji ide u audit log obje strane",
)
rb1, rb2, rb3, rb4 = st.columns(4)
if rb1.button(
f"↩️ Vrati ovu predlagaču",
key=f"sb_this_{other_id}",
use_container_width=True,
help=f"Status OVE akcije → 💡 idea, predlagač ({sel.get('owner') or sel.get('dept','?')}) dobiva flag.",
):
update_promo(selected_id, {"status": "🔧 preparing"})
append_log(selected_id, "Direktor nabave",
f"Sent back due to clash with {b['name']}", note)
st.success("Vraćeno predlagaču ove akcije.")
st.rerun()
if rb2.button(
"↩️ Vrati drugu predlagaču",
key=f"sb_other_{other_id}",
use_container_width=True,
help=f"Status DRUGE akcije → 🔧 preparing, predlagač je revidira u Promo Toolu.",
):
update_promo(other_id, {"status": "🔧 preparing"})
append_log(other_id, "Direktor nabave",
f"Sent back due to clash with {sel['name']}", note)
st.success(f"Druga akcija vraćena predlagaču ({b['dept']}).")
st.rerun()
if rb3.button(
"↩️ Vrati oboje",
key=f"sb_both_{other_id}",
use_container_width=True,
help="Oba prijedloga idu natrag na doradu s istim komentarom.",
):
update_promo(selected_id, {"status": "🔧 preparing"})
update_promo(other_id, {"status": "🔧 preparing"})
append_log(selected_id, "Direktor nabave",
f"Sent back due to clash with {b['name']}", note)
append_log(other_id, "Direktor nabave",
f"Sent back due to clash with {sel['name']}", note)
st.success("Oba prijedloga vraćena predlagačima.")
st.rerun()
if rb4.button(
"✅ Acknowledge (oba OK)",
key=f"ack_{other_id}",
use_container_width=True,
help="Direktor svjesno dopušta oba prijedloga. "
"Conflict warning se gasi za ovaj par; status ostaje nepromijenjen.",
):
acknowledge_conflict(selected_id, other_id)
append_log(selected_id, "Direktor nabave",
"Acknowledged clash (kept both intentionally)",
f"With {b['name']}")
append_log(other_id, "Direktor nabave",
"Acknowledged clash (kept both intentionally)",
f"With {sel['name']}")
st.success("Acknowledged — clash više neće biti označen.")
st.rerun()
# ---- SKU breakdown (real from PromoTool, else synthesized dummy) ----
cm_df = _load_cm_promotion_detail()
sku_detail = pd.DataFrame()
is_estimated = False
if not cm_df.empty and "id" in cm_df.columns:
sku_detail = cm_df[cm_df["id"].astype(str) == selected_id].copy()
if sku_detail.empty:
sku_detail = _synthesize_sku_breakdown(sel)
is_estimated = not sku_detail.empty
if not sku_detail.empty:
if is_estimated:
st.info(
"📊 **Estimated values** — ovaj prijedlog nije submitan kroz Promo Tool, "
"pa su količine / prihod / RUC deterministički generirane na bazi "
"`sku_prices.csv` (demo dummy data, reproducibilno po promo id-u + SKU-u)."
)
cols_keep = [c for c in
["sku", "sku_name", "discount_pct", "mechanic",
"total_units", "revenue_eur", "ruc_eur"]
if c in sku_detail.columns]
show_df = sku_detail[cols_keep].rename(columns={
"sku": "SKU", "sku_name": "Artikl",
"discount_pct": "Disc %", "mechanic": "Mehanika",
"total_units": "Proj. količina",
"revenue_eur": "Proj. prihod (€)",
"ruc_eur": "Proj. RUC (€)",
})
st.markdown("**Artikli i projekcija**" + (" *(dummy)*" if is_estimated else ""))
st.dataframe(show_df, use_container_width=True, hide_index=True)
t_units = int(pd.to_numeric(sku_detail.get("total_units", 0),
errors="coerce").fillna(0).sum())
t_rev = float(pd.to_numeric(sku_detail.get("revenue_eur", 0),
errors="coerce").fillna(0).sum())
t_ruc = float(pd.to_numeric(sku_detail.get("ruc_eur", 0),
errors="coerce").fillna(0).sum())
ruc_pct = (t_ruc / t_rev * 100) if t_rev else 0
m1, m2, m3, m4, m5 = st.columns(5)
m1.metric("SKUs", f"{len(sku_detail)}")
m2.metric("Proj. količina", f"{t_units:,}")
m3.metric("Proj. prihod", f"{t_rev:,.0f} €")
m4.metric("Proj. RUC", f"{t_ruc:,.0f} €")
m5.metric("RUC %", f"{ruc_pct:.1f}%")
else:
st.caption(
"ℹ️ Prijedlog nema niti SKU listu niti kategoriju — dodaj artikle "
"kroz **Edit** ispod ili re-submitaj iz Promo Toola."
)
m1, m2 = st.columns(2)
m1.metric("SKUs", "0")
m2.metric("Total units", f"{int(sel.get('units', 0) or 0):,}")
# ---- Audit log + edit ----
log = str(sel.get("approval_log", "") or "").strip()
if log:
with st.expander("📜 Audit log"):
for line in log.splitlines():
st.caption(line)
with st.expander("📝 Edit / delete"):
cf1, cf2 = st.columns(2)
new_notes = cf1.text_area("Notes",
value=str(sel.get("notes", "") or ""),
key="ed_notes_det")
new_owner = cf2.text_input("Owner",
value=str(sel.get("owner", "") or ""),
key="ed_owner_det")
cf3, cf4, cf5 = st.columns(3)
new_skus = cf3.text_input("SKUs (comma-separated)",
value=str(sel.get("skus", "") or ""),
key="ed_skus_det")
new_sw = cf4.number_input("Start CW", min_value=0, max_value=53,
value=int(sel.get("start_week", 0) or 0),
key="ed_sw_det")
new_ew = cf5.number_input("End CW", min_value=0, max_value=53,
value=int(sel.get("end_week", 0) or 0),
key="ed_ew_det")
cs1, cs2 = st.columns(2)
if cs1.button("💾 Save edits", type="primary",
use_container_width=True, key="ed_save_det"):
update_promo(selected_id, {
"notes": new_notes, "owner": new_owner,
"skus": new_skus,
"start_week": int(new_sw), "end_week": int(new_ew),
})
append_log(selected_id, actor, "Edited metadata", comment)
st.rerun()
if cs2.button("🗑️ Delete promo",
use_container_width=True, key="del_btn_det"):
delete_promo(selected_id)
st.session_state["selected_promo_id"] = ""
st.rerun()
# ============================================================
# PAGE 2: QUICK ADD
# ============================================================
elif page.startswith("➕"):
st.markdown("### ➕ Quick-add promotion")
st.caption("Lightweight entry. For deeper planning (P&L, uplift forecast, "
"cannibalization, override grid) use the Promo Tool.")
src_options = list(SOURCES.keys())
default_src_idx = 0
with st.form("qa_form", clear_on_submit=True):
c1, c2 = st.columns(2)
name = c1.text_input("Promo name *", placeholder="e.g. Summer Protein Sale")
source = c2.selectbox("Source *", src_options, index=default_src_idx)
c3, c4 = st.columns(2)
ptype = c3.selectbox("Type", TYPES)
outcome = c4.selectbox("Outcome", [""] + OUTCOMES)
c5, c6 = st.columns(2)
skus = c5.text_input("SKUs (comma-separated)",
placeholder="POL09754, POL09755")
weekrange = c6.text_input("Calendar weeks (e.g. 20-23)",
placeholder="20-23")
c7, c8, c9 = st.columns(3)
category = c7.text_input("Category")
owner = c8.text_input("Owner")
units = c9.number_input("Total units (optional)", min_value=0,
value=0, step=10)
c10, c11 = st.columns(2)
status = c10.selectbox("Status", STATUSES, index=0)
notes = c11.text_input("Notes")
if st.form_submit_button("➕ Add to calendar", type="primary"):
if not name.strip():
st.error("Promo name is required.")
else:
try:
sw, ew = parse_week_range(weekrange)
except ValueError:
st.error("Calendar weeks must be a number or range like '20-23'.")
sw, ew = 0, 0
if not sw:
st.error("Calendar weeks are required.")
else:
add_promo({
"name": name, "source": source, "type": ptype,
"outcome": outcome, "status": status,
"start_year": cur_y, "start_week": sw,
"end_year": cur_y, "end_week": ew,
"skus": skus, "units": units,
"category": category, "owner": owner, "notes": notes,
})
st.success(f"Added '{name}' to calendar (CW{sw}-CW{ew}).")
st.rerun()
# ============================================================
# PAGE 3: IMPORT / EXPORT
# ============================================================
elif page.startswith("📥"):
st.markdown("### 📥 Import promotions from Excel")
st.caption(
"Drop an .xlsx with promotions. Required column: **name** "
"(or `promo`/`naziv`/`promotion`). Other columns are auto-detected "
"if their headers match common synonyms — `source`, `type`, "
"`outcome`, `status`, `skus` (or `artikli`), `category`, "
"`start_week` (or `od_tjedna`), `end_week`, `units`, `owner`, `notes`. "
"Each Excel row becomes one promotion. Existing promotions are kept."
)
up = st.file_uploader("Excel file", type=["xlsx", "xls"], key="imp_xls")
if up is not None:
if st.button("Import now", type="primary", key="imp_btn"):
with st.spinner("Importing…"):
res = import_excel(up.getvalue())
if res["added"]:
st.success(f"Imported {res['added']} promotion(s).")
if res["skipped"]:
st.warning(f"Skipped {res['skipped']} row(s).")
for e in res["errors"][:10]:
st.caption(f"⚠️ {e}")
if res["added"]:
st.rerun()
st.divider()
st.markdown("### 📤 Export current calendar")
if df.empty:
st.caption("Nothing to export yet.")
else:
csv_bytes = df.to_csv(index=False).encode("utf-8")
st.download_button("⬇️ Download all promos (CSV)", csv_bytes,
file_name=f"promo_calendar_{cur_y}w{cur_w}.csv",
mime="text/csv")
# Excel export
try:
import io as _io
from openpyxl import Workbook as _WB
buf = _io.BytesIO()
wb = _WB()
ws = wb.active
ws.title = "Promo Calendar"
ws.append(list(df.columns))
for _, r in df.iterrows():
ws.append([r[c] for c in df.columns])
wb.save(buf)
st.download_button("⬇️ Download all promos (Excel)",
buf.getvalue(),
file_name=f"promo_calendar_{cur_y}w{cur_w}.xlsx",
mime="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")
except Exception:
pass
st.divider()
st.markdown("### 🧹 Reset")
with st.expander("Wipe all promotions"):
st.caption("Deletes the entire promo_calendar.csv. Cannot be undone.")
confirm = st.checkbox("I understand", key="wipe_confirm")
if st.button("🗑️ Wipe all promotions",
disabled=not confirm, key="wipe_btn"):
from promo_data import save_promos as _save
_save(pd.DataFrame(columns=df.columns))
st.success("Wiped.")
st.rerun()