""" Polleo Demand Planning — Slack Agent ===================================== Automates the KAM/CM input cycle via Slack: 1. Generate templates for all configured people 2. Post templates to a Slack channel with @mentions 3. Monitor channel for returned files 4. Validate and combine responses 5. Send reminders for missing submissions Usage from Streamlit: from slack_agent import SlackAgent agent = SlackAgent(data_dir="data") agent.distribute_all() # Friday: send all templates agent.collect_responses() # Monday: pull files from Slack agent.send_nudges() # Monday AM: remind stragglers agent.status() # Check who's submitted Usage standalone: python slack_agent.py distribute python slack_agent.py collect python slack_agent.py nudge python slack_agent.py status """ import json import logging import os import sys from datetime import datetime, timedelta from pathlib import Path from typing import Optional import pandas as pd from openpyxl import Workbook, load_workbook from openpyxl.styles import Font, PatternFill, Alignment from openpyxl.utils import get_column_letter logger = logging.getLogger("slack_agent") logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s") # ═══════════════════════════════════════════════════════════════════ # CONFIG # ═══════════════════════════════════════════════════════════════════ DEFAULT_CONFIG = { "slack_config": { "bot_token": "", # xoxb-... (set via env POLLEO_SLACK_TOKEN or here) "channel_id": "", # C0XXXXXXX — the #demand-inputs channel "nudge_day": "Monday", "nudge_hour": 9, "deadline_day": "Monday", "deadline_hour": 17, "status_thread_ts": None, # auto-set: timestamp of the status message "current_cycle": None # auto-set: "2026-CW16" etc. } } class SlackAgent: """Orchestrates KAM/CM input collection via Slack.""" def __init__(self, data_dir: str = "data"): self.data_dir = Path(data_dir) self.data_dir.mkdir(exist_ok=True) self.config_path = self.data_dir / "kam_cm_config.json" self.cycle_path = self.data_dir / "slack_cycle.json" self.config = self._load_config() self.slack = self._init_slack() # ── Config ────────────────────────────────────────────────── def _load_config(self) -> dict: if self.config_path.exists(): with open(self.config_path) as f: return json.load(f) return {} def _save_config(self): with open(self.config_path, "w") as f: json.dump(self.config, f, indent=2, ensure_ascii=False) def _slack_config(self) -> dict: return self.config.get("slack_config", {}) def _people(self) -> dict: return self.config.get("kam_cm_config", {}) def _get_token(self) -> str: """Token from config or env var.""" return ( os.environ.get("POLLEO_SLACK_TOKEN", "") or self._slack_config().get("bot_token", "") ) # ── Cycle tracking ────────────────────────────────────────── def _current_cw(self): iso = datetime.now().isocalendar() return iso[0], iso[1] def _cycle_id(self) -> str: y, w = self._current_cw() return f"{y}-CW{w}" def _load_cycle(self) -> dict: """Load the current distribution cycle state.""" if self.cycle_path.exists(): with open(self.cycle_path) as f: return json.load(f) return {} def _save_cycle(self, cycle: dict): with open(self.cycle_path, "w") as f: json.dump(cycle, f, indent=2, ensure_ascii=False) def _init_cycle(self) -> dict: """Create a fresh cycle tracking object.""" cycle_id = self._cycle_id() y, cw = self._current_cw() cw_start = cw + 1 cw_labels = [f"CW{cw_start + j}" for j in range(13)] people = self._people() submissions = {} for key, info in people.items(): display = info.get("display_name", key) submissions[key] = { "display_name": display, "role": info.get("role", ""), "status": "pending", # pending | submitted | validated | error "distributed_at": None, "file_posted_at": None, # Slack ts of the template message "submitted_at": None, "submitted_file": None, "validation_errors": [], "sku_count": 0, "total_units": 0, } return { "cycle_id": cycle_id, "cw_labels": cw_labels, "started_at": datetime.now().isoformat(), "status_thread_ts": None, "submissions": submissions, } # ── Slack client ──────────────────────────────────────────── def _init_slack(self): """Lazy-init Slack WebClient. Returns None if no token.""" token = self._get_token() if not token: logger.warning("No Slack token configured — running in dry-run mode") return None try: from slack_sdk import WebClient client = WebClient(token=token) # Quick auth check client.auth_test() logger.info("Slack connected") return client except ImportError: logger.error("slack_sdk not installed. Run: pip install slack_sdk") return None except Exception as e: logger.error(f"Slack auth failed: {e}") return None def _channel_id(self) -> str: return self._slack_config().get("channel_id", "") # ═══════════════════════════════════════════════════════════════ # TEMPLATE GENERATION # ═══════════════════════════════════════════════════════════════ def generate_template(self, person_key: str, strict: bool = True) -> Optional[Path]: """ Generate an Excel template for one person. Returns filepath or None. strict=True (default): refuse to generate if the display_name-fallback bug is detected — i.e. config is missing `display_name`, so kam_name falls back to person_key, AND the detail CSV has history for other people but zero rows match this kam_name. Returns None and logs an actionable error. strict=False: still generates the template (blank) but logs a warning. """ people = self._people() if person_key not in people: logger.error(f"Unknown person key: {person_key}") return None info = people[person_key] has_display_name = "display_name" in info kam_name = info.get("display_name", person_key) type_prefix = info.get("role", "VP") person_cats = info.get("categories", "ALL") excluded_mp_cats = self.config.get("excluded_categories_mp", []) buyers = info.get("buyers") or [] # multi-sheet split; empty = single sheet # Load SKU plan list plan_path = self.data_dir / "sku_plan_list.csv" if not plan_path.exists(): logger.error("sku_plan_list.csv not found") return None plan = pd.read_csv(plan_path) # Calendar weeks _, cw = self._current_cw() cw_start = cw + 1 n_fc = 13 cw_labels = [f"CW{cw_start + j}" for j in range(n_fc)] # Filter SKUs template_skus = plan.copy() if person_cats != "ALL": template_skus = template_skus[template_skus["cat"].isin(person_cats)] if type_prefix == "MP" and excluded_mp_cats: template_skus = template_skus[~template_skus["cat"].isin(excluded_mp_cats)] if len(template_skus) == 0: logger.warning(f"No SKUs for {kam_name} ({type_prefix})") return None # Subcategory map (VP only) subcat_map = {} if type_prefix == "VP": sc_path = self.data_dir / "sku_subcat_map.csv" if sc_path.exists(): scm = pd.read_csv(sc_path) subcat_map = dict(zip(scm["sku"], scm["sub_cat"])) # Previous entries for pre-fill. When buyers are defined, index by # (sku, type, buyer) so each sheet pre-fills only that buyer's rows. # Otherwise index by (sku, type) as before. detail_path = self.data_dir / f"{type_prefix.lower()}_input_detail.csv" prev_by_buyer = {} # buyer -> {(sku,type): {cw: value}}; "" = no-buyer single-sheet if detail_path.exists(): detail = pd.read_csv(detail_path) kam_rows = detail[detail["kam"].str.lower() == kam_name.lower()] # ── Safety check: detect display_name fallback bug + name typos ── if len(kam_rows) == 0 and len(detail) > 0: other_kams = sorted(detail["kam"].dropna().unique().tolist()) if other_kams: kn_low = kam_name.lower() candidates = [ k for k in other_kams if k.lower() in kn_low or kn_low in k.lower() ] fallback_bug = (not has_display_name) and (kam_name == person_key) msg_lines = [ f"No previous entries found for kam_name='{kam_name}' in " f"{detail_path.name} " f"(file has {len(detail)} rows from {len(other_kams)} people: {other_kams}).", ] if fallback_bug: msg_lines.append( f" ⚠️ display_name fallback bug: kam_name equals " f"person_key ('{person_key}') because 'display_name' " f"is missing in kam_cm_config.json for this person." ) if candidates: msg_lines.append(f" Possible matches in history: {candidates}") if fallback_bug and len(candidates) == 1: msg_lines.append( f" Fix: add \"display_name\": \"{candidates[0]}\" " f"to kam_cm_config.json → kam_cm_config → {person_key}" ) msg = "\n".join(msg_lines) if strict and fallback_bug: logger.error(f"Template generation REFUSED.\n{msg}") return None else: logger.warning(msg) has_buyer_col = "buyer" in detail.columns for _, row in kam_rows.iterrows(): b = str(row.get("buyer", "") or "") if has_buyer_col else "" key = (row["sku"], row["type"]) cw_vals = {c: int(row[c]) for c in row.index if c.startswith("CW") and row[c] > 0} if cw_vals: prev_by_buyer.setdefault(b, {})[key] = cw_vals prefill_count = sum(len(v) for v in prev_by_buyer.values()) # ── Build workbook ── wb = Workbook() # Remove the default sheet so we can create sheets by buyer name # without a stray "Sheet" left over. default_ws = wb.active wb.remove(default_ws) input_label = "VP (WHOLESALE)" if type_prefix == "VP" else "MP (MARKETING/RETAIL)" hdr_fill = PatternFill("solid", fgColor="2F5496") cw_fill = PatternFill("solid", fgColor="1F3864") input_fill = PatternFill("solid", fgColor="FFFDE7") prefill_fill = PatternFill("solid", fgColor="E2EFDA") if type_prefix == "VP": headers = ["SKU", "Artikl", "Grupacija", "Subkategorija", "OZNAKA", "Type"] + cw_labels data_start_col = 7 else: headers = ["SKU", "Artikl", "Grupacija", "OZNAKA", "Type"] + cw_labels data_start_col = 6 def _build_sheet(ws, sheet_buyer, sheet_prev): ws.cell(1, 1, f"DEMAND INPUT - {input_label}").font = Font(bold=True, size=14, color="1B2A4A") header_line = f"KAM/CM: {kam_name}" + (f" · Buyer: {sheet_buyer}" if sheet_buyer else "") ws.cell(2, 1, header_line).font = Font(size=11, color="2F5496") ws.cell(2, 5, f"Generated: {datetime.now().strftime('%Y-%m-%d')}").font = Font(size=9, color="999999") if sheet_prev: ws.cell(3, 1, f"Pre-filled with {len(sheet_prev)} previous entries " "(green = your last input, yellow = enter new)" ).font = Font(size=9, italic=True, color="548235") hr = 4 for c, h in enumerate(headers, 1): cell = ws.cell(hr, c, h) cell.font = Font(bold=True, size=10, color="FFFFFF") cell.fill = hdr_fill if c < data_start_col else cw_fill cell.alignment = Alignment(horizontal="center") ws.column_dimensions["A"].width = 14 ws.column_dimensions["B"].width = 42 ws.column_dimensions["C"].width = 22 if type_prefix == "VP": ws.column_dimensions["D"].width = 28 ws.column_dimensions["E"].width = 12 ws.column_dimensions["F"].width = 18 else: ws.column_dimensions["D"].width = 12 ws.column_dimensions["E"].width = 18 for j in range(n_fc): ws.column_dimensions[get_column_letter(data_start_col + j)].width = 9 row = hr + 1 for _, sku_row in template_skus.iterrows(): for rtype in ["on-top demand", "regular increase"]: col = 1 ws.cell(row, col, sku_row["sku"]).font = Font(size=10); col += 1 ws.cell(row, col, sku_row["name"]).font = Font(size=10, color="404040"); col += 1 ws.cell(row, col, sku_row["cat"]).font = Font(size=10); col += 1 if type_prefix == "VP": subcat = subcat_map.get(sku_row["sku"], "") ws.cell(row, col, subcat).font = Font(size=9, color="808080"); col += 1 ws.cell(row, col, sku_row["oznaka"]).font = Font(size=10); col += 1 ws.cell(row, col, rtype).font = Font(bold=True, size=9); col += 1 prev_vals = sheet_prev.get((sku_row["sku"], rtype), {}) for j in range(n_fc): cell = ws.cell(row, data_start_col + j) cw_label = cw_labels[j] if cw_label in prev_vals: cell.value = prev_vals[cw_label] cell.fill = prefill_fill cell.font = Font(size=10, color="548235") else: cell.fill = input_fill cell.alignment = Alignment(horizontal="right") cell.number_format = "#,##0" row += 1 ws.auto_filter.ref = f"A{hr}:{get_column_letter(data_start_col - 1 + n_fc)}{row - 1}" ws.freeze_panes = f"{get_column_letter(data_start_col)}5" # Sheet-name sanitiser — Excel forbids : \ / ? * [ ] and 31-char cap. def _safe_sheet_name(name): clean = "".join(c for c in str(name) if c not in r":\\/?*[]").strip() or "Sheet" return clean[:31] if buyers: # One sheet per buyer, in config order. for b in buyers: ws = wb.create_sheet(_safe_sheet_name(b)) _build_sheet(ws, b, prev_by_buyer.get(b, {})) else: ws = wb.create_sheet(f"{type_prefix} Input") _build_sheet(ws, "", prev_by_buyer.get("", {})) # Hidden meta sheet ws_meta = wb.create_sheet("_meta") ws_meta.cell(1, 1, "kam_name"); ws_meta.cell(1, 2, kam_name) ws_meta.cell(2, 1, "type"); ws_meta.cell(2, 2, type_prefix) ws_meta.cell(3, 1, "generated"); ws_meta.cell(3, 2, datetime.now().strftime("%Y-%m-%d %H:%M")) ws_meta.cell(4, 1, "config_key"); ws_meta.cell(4, 2, person_key) ws_meta.cell(5, 1, "buyers"); ws_meta.cell(5, 2, ", ".join(buyers) if buyers else "") ws_meta.sheet_state = "hidden" safe_name = kam_name.replace(" ", "_").replace("/", "_") filename = f"{type_prefix}_Input_{safe_name}.xlsx" filepath = self.data_dir / filename wb.save(filepath) logger.info(f"Generated: {filename} — {len(template_skus)} SKUs, {prefill_count} pre-filled") return filepath # ═══════════════════════════════════════════════════════════════ # DISTRIBUTE — post templates to Slack # ═══════════════════════════════════════════════════════════════ def distribute_all(self, dry_run: bool = False) -> dict: """ Generate templates for every configured person and post to Slack. Returns cycle state dict. """ cycle = self._init_cycle() channel = self._channel_id() _, cw = self._current_cw() cw_range = f"CW{cw+1}–CW{cw+13}" results = {"generated": [], "posted": [], "errors": []} # ── 1. Post cycle header ── header_text = ( f":clipboard: *Demand Input Cycle — {self._cycle_id()}*\n" f"Forecast window: *{cw_range}* (13 weeks)\n" f"Deadline: *{self._slack_config().get('deadline_day', 'Monday')} " f"{self._slack_config().get('deadline_hour', 17)}:00*\n\n" f"Templates are coming in the thread below :point_down:" ) thread_ts = None if self.slack and not dry_run: try: resp = self.slack.chat_postMessage( channel=channel, text=header_text, mrkdwn=True, ) thread_ts = resp["ts"] # Pin it for visibility try: self.slack.pins_add(channel=channel, timestamp=thread_ts) except Exception: pass # already pinned or no permission except Exception as e: logger.error(f"Failed to post header: {e}") results["errors"].append(f"Header post failed: {e}") else: logger.info(f"[DRY RUN] Header: {header_text}") cycle["status_thread_ts"] = thread_ts # ── 2. Generate + post each template ── people = self._people() for person_key, info in people.items(): display = info.get("display_name", person_key) role = info.get("role", "VP") slack_id = info.get("slack_user_id", "") channel_desc = info.get("channel", "") cats = info.get("categories", "ALL") # Generate template filepath = self.generate_template(person_key) if filepath is None: results["errors"].append(f"{display}: template generation failed") cycle["submissions"][person_key]["status"] = "error" continue results["generated"].append(str(filepath)) # Count pre-filled detail_path = self.data_dir / f"{role.lower()}_input_detail.csv" prefill_count = 0 if detail_path.exists(): detail = pd.read_csv(detail_path) prefill_count = len(detail[detail["kam"].str.lower() == display.lower()]) # Build message mention = f"<@{slack_id}>" if slack_id else f"*{display}*" cat_desc = "all categories" if cats == "ALL" else ", ".join(cats) prefill_msg = f"\n:recycle: {prefill_count} entries pre-filled from last cycle" if prefill_count else "" message = ( f":page_facing_up: {mention} — *{role} Input*\n" f"Channel: {channel_desc} | Categories: {cat_desc}{prefill_msg}\n" f"Please fill in and upload your completed file in this thread." ) # Post to Slack if self.slack and not dry_run: try: resp = self.slack.files_upload_v2( channel=channel, file=str(filepath), filename=filepath.name, initial_comment=message, thread_ts=thread_ts, ) file_ts = resp.get("file", {}).get("timestamp", "") cycle["submissions"][person_key]["distributed_at"] = datetime.now().isoformat() cycle["submissions"][person_key]["file_posted_at"] = file_ts results["posted"].append(display) logger.info(f"Posted template for {display} ({role})") except Exception as e: logger.error(f"Failed to post for {display}: {e}") results["errors"].append(f"{display}: Slack post failed — {e}") else: logger.info(f"[DRY RUN] Would post {filepath.name} for {mention}") cycle["submissions"][person_key]["distributed_at"] = datetime.now().isoformat() results["posted"].append(display) # ── 3. Post status tracker ── status_text = self._format_status(cycle) if self.slack and not dry_run and thread_ts: try: resp = self.slack.chat_postMessage( channel=channel, text=status_text, thread_ts=thread_ts, mrkdwn=True, ) cycle["status_message_ts"] = resp["ts"] except Exception as e: logger.error(f"Status post failed: {e}") # Save cycle state self._save_cycle(cycle) logger.info(f"Distribution complete: {len(results['posted'])} posted, {len(results['errors'])} errors") return results # ═══════════════════════════════════════════════════════════════ # CLEANUP — delete bot messages from channel # ═══════════════════════════════════════════════════════════════ def cleanup_channel(self, max_messages: int = 200) -> dict: """ Delete all messages posted by this bot in the configured channel. This removes cycle headers, template posts, nudges, and status updates. Also resets the local slack_cycle.json. Returns {"deleted": int, "files_deleted": int, "errors": [str]} """ if not self.slack: return {"deleted": 0, "files_deleted": 0, "errors": ["No Slack client"]} channel = self._channel_id() if not channel: return {"deleted": 0, "files_deleted": 0, "errors": ["No channel_id configured"]} results = {"deleted": 0, "files_deleted": 0, "errors": []} try: # Get bot's own user ID auth = self.slack.auth_test() bot_user_id = auth["user_id"] except Exception as e: return {"deleted": 0, "files_deleted": 0, "errors": [f"Auth failed: {e}"]} try: # Fetch recent channel messages resp = self.slack.conversations_history( channel=channel, limit=max_messages ) messages = resp.get("messages", []) except Exception as e: return {"deleted": 0, "files_deleted": 0, "errors": [f"Cannot read channel: {e}"]} for msg in messages: # Only delete messages from our bot if msg.get("user") != bot_user_id and msg.get("bot_id") is None: continue ts = msg["ts"] # Delete files attached to this message for f in msg.get("files", []): try: self.slack.files_delete(file=f["id"]) results["files_deleted"] += 1 except Exception: pass # file may already be deleted or belong to another user # Also check thread replies for bot messages with files if msg.get("reply_count", 0) > 0: try: thread_resp = self.slack.conversations_replies( channel=channel, ts=ts, limit=200 ) for reply in thread_resp.get("messages", []): if reply["ts"] == ts: continue # skip parent, handled above if reply.get("user") != bot_user_id and reply.get("bot_id") is None: continue for f in reply.get("files", []): try: self.slack.files_delete(file=f["id"]) results["files_deleted"] += 1 except Exception: pass try: self.slack.chat_delete(channel=channel, ts=reply["ts"]) results["deleted"] += 1 except Exception: pass except Exception as e: results["errors"].append(f"Thread scan failed for {ts}: {e}") # Delete the parent message itself try: self.slack.chat_delete(channel=channel, ts=ts) results["deleted"] += 1 except Exception as e: results["errors"].append(f"Delete failed for {ts}: {e}") # Reset local cycle state if self.cycle_path.exists(): self.cycle_path.unlink() logger.info("Deleted slack_cycle.json") logger.info( f"Cleanup done: {results['deleted']} messages, " f"{results['files_deleted']} files deleted" ) return results # ═══════════════════════════════════════════════════════════════ # COLLECT — pull response files from Slack # ═══════════════════════════════════════════════════════════════ def collect_responses(self, dry_run: bool = False) -> dict: """ Scan the Slack thread for uploaded response files. Download, validate, and mark as submitted in cycle state. """ cycle = self._load_cycle() if not cycle: logger.error("No active cycle found — run distribute_all first") return {"error": "no active cycle"} thread_ts = cycle.get("status_thread_ts") channel = self._channel_id() results = {"collected": [], "errors": [], "skipped": []} if not self.slack: logger.error("No Slack connection") return {"error": "no slack connection"} # Get all replies in the thread try: replies = self.slack.conversations_replies( channel=channel, ts=thread_ts, limit=200, ) except Exception as e: logger.error(f"Failed to read thread: {e}") return {"error": str(e)} messages = replies.get("messages", []) # Look for file uploads in replies for msg in messages: files = msg.get("files", []) user_id = msg.get("user", "") for file_info in files: fname = file_info.get("name", "") if not fname.endswith(".xlsx"): continue # Skip files posted by the bot itself (our templates) if file_info.get("user") == self._get_bot_user_id(): continue # Download the file file_url = file_info.get("url_private_download", "") if not file_url: continue # Match to a person by slack_user_id person_key = self._match_person_by_slack_id(user_id) if not person_key: # Try matching by _meta sheet in the file person_key = "unknown" local_path = self.data_dir / f"_slack_{fname}" try: resp = self.slack.api_call( "files.info", params={"file": file_info["id"]} ) # Download via URL with auth header import urllib.request req = urllib.request.Request( file_url, headers={"Authorization": f"Bearer {self._get_token()}"} ) with urllib.request.urlopen(req) as response: local_path.write_bytes(response.read()) logger.info(f"Downloaded: {fname} from user {user_id}") except Exception as e: logger.error(f"Download failed for {fname}: {e}") results["errors"].append(f"{fname}: download failed — {e}") continue # Read _meta to identify try: wb = load_workbook(local_path, data_only=True) if "_meta" in wb.sheetnames: ws_m = wb["_meta"] meta_kam = ws_m.cell(1, 2).value or "" meta_type = ws_m.cell(2, 2).value or "" meta_key = ws_m.cell(4, 2).value or "" if meta_key in cycle["submissions"]: person_key = meta_key wb.close() except Exception: pass # Validate the file validation = self._validate_response(local_path) if person_key in cycle.get("submissions", {}): sub = cycle["submissions"][person_key] sub["status"] = "validated" if not validation["errors"] else "error" sub["submitted_at"] = datetime.now().isoformat() sub["submitted_file"] = fname sub["validation_errors"] = validation["errors"] sub["sku_count"] = validation["sku_count"] sub["total_units"] = validation["total_units"] # Move to a permanent location final_path = self.data_dir / fname local_path.rename(final_path) results["collected"].append({ "person": person_key, "file": fname, "skus": validation["sku_count"], "units": validation["total_units"], "errors": validation["errors"], }) # Update status in Slack if results["collected"]: self._save_cycle(cycle) self._update_status_message(cycle) logger.info(f"Collection: {len(results['collected'])} files, {len(results['errors'])} errors") return results def _validate_response(self, filepath: Path) -> dict: """Basic validation of a returned template file.""" result = {"sku_count": 0, "total_units": 0, "errors": [], "warnings": []} try: wb = load_workbook(filepath, data_only=True) ws = wb.worksheets[0] # Find CW columns cw_cols = {} for c in range(6, 30): v = ws.cell(4, c).value if v and str(v).startswith("CW"): cw_cols[str(v)] = c elif v is None: break if not cw_cols: result["errors"].append("No CW columns found in header row 4") wb.close() return result skus_seen = set() total = 0 negatives = 0 for r in range(5, ws.max_row + 1): sku = ws.cell(r, 1).value if not sku: continue skus_seen.add(sku) for cw, col in cw_cols.items(): v = ws.cell(r, col).value if v is not None and v != "": try: val = float(v) total += val if val < 0: negatives += 1 except (ValueError, TypeError): pass result["sku_count"] = len(skus_seen) result["total_units"] = int(total) if negatives: result["errors"].append(f"{negatives} negative values found") if result["sku_count"] == 0: result["errors"].append("No SKU data found") if result["total_units"] == 0: result["warnings"].append("All values are zero — might be an empty template") wb.close() except Exception as e: result["errors"].append(f"File read error: {e}") return result # ═══════════════════════════════════════════════════════════════ # COMBINE — merge collected files into vp_input / mp_input # ═══════════════════════════════════════════════════════════════ def combine_collected(self) -> dict: """ Combine all collected response files into vp_input.csv / mp_input.csv. Same logic as the Streamlit combine tab. """ cycle = self._load_cycle() if not cycle: return {"error": "no active cycle"} vp_files = [] mp_files = [] for key, sub in cycle.get("submissions", {}).items(): if sub["status"] not in ("validated", "submitted"): continue fname = sub.get("submitted_file") if not fname: continue filepath = self.data_dir / fname if not filepath.exists(): continue role = sub.get("role", "") if role == "VP": vp_files.append(filepath) elif role == "MP": mp_files.append(filepath) results = {} for type_prefix, files in [("VP", vp_files), ("MP", mp_files)]: if not files: continue all_rows = [] for fp in files: try: wb = load_workbook(fp, data_only=True) kam_name = "Unknown" if "_meta" in wb.sheetnames: kam_name = wb["_meta"].cell(1, 2).value or fp.name # Iterate every visible data sheet. If the workbook has # multiple sheets (one per buyer), the sheet name is the # buyer. Single-sheet workbooks get buyer="". data_sheets = [s for s in wb.sheetnames if s != "_meta"] for sheet_name in data_sheets: ws = wb[sheet_name] buyer = sheet_name if len(data_sheets) > 1 else "" # Auto-detect Type column from header row. # VP layout has Subkategorija so Type is at col 6; MP has Type at col 5. type_col = 5 for c in range(1, 10): hv = ws.cell(4, c).value if hv and str(hv).strip().lower() == "type": type_col = c break cw_cols = {} for c in range(type_col + 1, 30): v = ws.cell(4, c).value if v and str(v).startswith("CW"): cw_cols[str(v)] = c elif v is None: break for r in range(5, ws.max_row + 1): sku = ws.cell(r, 1).value if not sku: continue rtype = ws.cell(r, type_col).value or "" row_data = {"sku": sku, "type": rtype, "kam": kam_name, "buyer": buyer} has_data = False for cw_label, col in cw_cols.items(): v = ws.cell(r, col).value row_data[cw_label] = float(v) if v and v != "" else 0 if v and float(v) > 0: has_data = True if has_data: all_rows.append(row_data) wb.close() except Exception as e: logger.error(f"Error reading {fp.name}: {e}") if not all_rows: continue combined = pd.DataFrame(all_rows) sum_cols = [c for c in combined.columns if c.startswith("CW")] # Detail file (per-KAM audit trail) detail_csv = self.data_dir / f"{type_prefix.lower()}_input_detail.csv" combined.to_csv(detail_csv, index=False) # Summed file (grouped by SKU) — MERGE with existing so past CW # columns (already-submitted weeks) are preserved. Without this, # every combine overwrites with only the current-horizon CWs and # history is lost the moment the forecast rolls forward. summed = combined.groupby("sku")[sum_cols].sum().reset_index() output_csv = self.data_dir / f"{type_prefix.lower()}_input.csv" if output_csv.exists(): try: prev = pd.read_csv(output_csv) prev_cw_cols = [c for c in prev.columns if c.startswith("CW")] new_cw_cols = sum_cols # Columns present in prev but not in new submission — keep them. preserve_cols = [c for c in prev_cw_cols if c not in new_cw_cols] if preserve_cols: # All SKUs union (prev + new) all_skus = sorted(set(prev["sku"]).union(set(summed["sku"]))) prev_idx = prev.set_index("sku") new_idx = summed.set_index("sku") merged_rows = [] for s in all_skus: row = {"sku": s} # preserved past columns from prev for c in preserve_cols: row[c] = prev_idx.at[s, c] if s in prev_idx.index else 0 # new-horizon columns from fresh submission for c in new_cw_cols: row[c] = new_idx.at[s, c] if s in new_idx.index else 0 merged_rows.append(row) summed = pd.DataFrame(merged_rows) # Keep columns ordered by CW number ascending. cw_sorted = sorted(preserve_cols + new_cw_cols, key=lambda c: int(c[2:])) summed = summed[["sku"] + cw_sorted] except Exception as e: logger.warning(f"Could not merge with existing {output_csv.name}: {e}") summed.to_csv(output_csv, index=False) results[type_prefix] = { "skus": len(summed), "kams": combined["kam"].nunique(), "total_units": int(combined[sum_cols].sum().sum()), } logger.info(f"Combined {type_prefix}: {len(summed)} SKUs from {combined['kam'].nunique()} source(s)") return results # ═══════════════════════════════════════════════════════════════ # NUDGE — remind people who haven't submitted # ═══════════════════════════════════════════════════════════════ def send_nudges(self, dry_run: bool = False) -> list: """Send reminder messages to people who haven't submitted yet.""" cycle = self._load_cycle() if not cycle: logger.error("No active cycle") return [] channel = self._channel_id() thread_ts = cycle.get("status_thread_ts") deadline_day = self._slack_config().get("deadline_day", "Monday") deadline_hour = self._slack_config().get("deadline_hour", 17) nudged = [] for key, sub in cycle.get("submissions", {}).items(): if sub["status"] != "pending": continue display = sub["display_name"] role = sub["role"] people = self._people() slack_id = people.get(key, {}).get("slack_user_id", "") mention = f"<@{slack_id}>" if slack_id else f"*{display}*" message = ( f":bell: {mention} — reminder: your *{role} Input* is still pending.\n" f"Deadline: *{deadline_day} {deadline_hour}:00*.\n" f"Previous entries will carry forward if not submitted." ) if self.slack and not dry_run: try: self.slack.chat_postMessage( channel=channel, text=message, thread_ts=thread_ts, mrkdwn=True, ) nudged.append(display) except Exception as e: logger.error(f"Nudge failed for {display}: {e}") else: logger.info(f"[DRY RUN] Nudge: {message}") nudged.append(display) return nudged # ═══════════════════════════════════════════════════════════════ # STATUS # ═══════════════════════════════════════════════════════════════ def status(self) -> dict: """Get current cycle status.""" cycle = self._load_cycle() if not cycle: return {"active": False, "message": "No active cycle"} subs = cycle.get("submissions", {}) pending = [s["display_name"] for s in subs.values() if s["status"] == "pending"] submitted = [s["display_name"] for s in subs.values() if s["status"] in ("validated", "submitted")] errored = [s["display_name"] for s in subs.values() if s["status"] == "error"] return { "active": True, "cycle_id": cycle.get("cycle_id"), "total": len(subs), "submitted": submitted, "pending": pending, "errored": errored, "all_in": len(pending) == 0 and len(errored) == 0, } def _format_status(self, cycle: dict) -> str: """Format a status block for Slack.""" lines = [":bar_chart: *Submission Status*\n"] for key, sub in cycle.get("submissions", {}).items(): display = sub["display_name"] role = sub["role"] st = sub["status"] if st == "pending": icon = ":hourglass_flowing_sand:" elif st in ("validated", "submitted"): icon = ":white_check_mark:" extra = f" — {sub['sku_count']} SKUs, {sub['total_units']:,} units" lines.append(f"{icon} {display} ({role}){extra}") continue elif st == "error": icon = ":warning:" else: icon = ":grey_question:" lines.append(f"{icon} {display} ({role})") return "\n".join(lines) def _update_status_message(self, cycle: dict): """Edit the status message in Slack to reflect current state.""" if not self.slack: return channel = self._channel_id() ts = cycle.get("status_message_ts") if not ts: return try: self.slack.chat_update( channel=channel, ts=ts, text=self._format_status(cycle), mrkdwn=True, ) except Exception as e: logger.error(f"Status update failed: {e}") def _match_person_by_slack_id(self, slack_user_id: str) -> Optional[str]: """Find config key by Slack user ID.""" for key, info in self._people().items(): if info.get("slack_user_id") == slack_user_id: return key return None def _get_bot_user_id(self) -> Optional[str]: """Get the bot's own user ID to skip its uploads.""" if not self.slack: return None try: resp = self.slack.auth_test() return resp.get("user_id") except Exception: return None # ═══════════════════════════════════════════════════════════════════ # CLI # ═══════════════════════════════════════════════════════════════════ def main(): if len(sys.argv) < 2: print("Usage: python slack_agent.py [distribute|collect|combine|nudge|status]") print(" distribute — Generate all templates and post to Slack") print(" collect — Pull response files from Slack thread") print(" combine — Combine collected files into vp/mp_input.csv") print(" nudge — Send reminders to pending people") print(" status — Show current cycle status") sys.exit(1) cmd = sys.argv[1].lower() dry_run = "--dry-run" in sys.argv agent = SlackAgent(data_dir="data") if cmd == "distribute": result = agent.distribute_all(dry_run=dry_run) print(json.dumps(result, indent=2, default=str)) elif cmd == "collect": result = agent.collect_responses(dry_run=dry_run) print(json.dumps(result, indent=2, default=str)) elif cmd == "combine": result = agent.combine_collected() print(json.dumps(result, indent=2, default=str)) elif cmd == "nudge": nudged = agent.send_nudges(dry_run=dry_run) print(f"Nudged: {', '.join(nudged) if nudged else 'nobody — all submitted'}") elif cmd == "status": st = agent.status() print(json.dumps(st, indent=2)) if st.get("all_in"): print("\n✅ All inputs received — ready to combine and run forecast") elif st.get("pending"): print(f"\n⏳ Waiting on: {', '.join(st['pending'])}") else: print(f"Unknown command: {cmd}") sys.exit(1) if __name__ == "__main__": main()