"""Block Kit builders for the daily summary post and the 3rd-escalation alert. Two push surfaces, no App Home (see CLAUDE.md "Slack surfaces"): - daily summary: header, vs-yesterday deltas, escalation + action/routine fields, top sites, mismatch callout, and a Open dashboard link button to Grafana. Long WO lists go in modals, never the channel (<100-block limit). - 3rd-escalation alert: standalone, batched, one @here — suppressed entirely on zero-3rd days (return None). Owned by the slack-blockkit-designer agent; built in Phase 4 (docs/BUILD.md). Slack constraints enforced here: - Message / modal max 100 blocks. - Actions block max 25 elements (kept to 5 per block for readability). - section text max 3000 chars. - Header text max 150 chars. - No APM deep-links anywhere — Grafana only. """ from __future__ import annotations # --------------------------------------------------------------------------- # Constants # --------------------------------------------------------------------------- ESCALATION_CATEGORIES: tuple[str, ...] = ( "1st Escalation", "2nd Escalation", "3rd Escalation", "SIM Ticket", "Other Escalation", ) # Categories that drive the "drill" buttons — non-routine and worth calling out. # Order determines button priority when we cap at 5 per actions block. DRILL_CATEGORY_PRIORITY: tuple[str, ...] = ( "3rd Escalation", "2nd Escalation", "1st Escalation", "SIM Ticket", "Vendor No-Show", "Awaiting Scheduling", "Report / Docs Needed", "Awaiting Report / Invoice", "Awaiting Vendor / Parts", "Status Inquiry", "Other Escalation", ) # Max WOs to render inline before the "+M more" overflow note in the alert. ALERT_MAX_WOS = 20 # Max blocks reserved for WO rows in the modal (leaves headroom for # title/footer blocks). Each WO gets 1 section block. MODAL_HEADER_BLOCKS = 1 # title is not a block; we add 1 context block as preamble MODAL_FOOTER_BLOCKS = 1 # overflow context block (may not be emitted) MODAL_MAX_WO_BLOCKS = 100 - MODAL_HEADER_BLOCKS - MODAL_FOOTER_BLOCKS # = 98 # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _divider() -> dict: return {"type": "divider"} def _header(text: str) -> dict: # Header block text is plain_text, max 150 chars. return { "type": "header", "text": {"type": "plain_text", "text": text[:150], "emoji": True}, } def _section(text: str) -> dict: # Section text is mrkdwn, max 3000 chars. return { "type": "section", "text": {"type": "mrkdwn", "text": text[:3000]}, } def _context(text: str) -> dict: return { "type": "context", "elements": [{"type": "mrkdwn", "text": text[:3000]}], } def _fields_section(fields: list[str]) -> dict: """Section with up to 10 fields (Slack limit), each <=2000 chars.""" return { "type": "section", "fields": [{"type": "mrkdwn", "text": f[:2000]} for f in fields[:10]], } def _button(text: str, action_id: str, value: str) -> dict: """Interactive button (opens modal via interactivity handler).""" return { "type": "button", "text": {"type": "plain_text", "text": text, "emoji": False}, "action_id": action_id, "value": value, } def _link_button(text: str, url: str) -> dict: """Link button — opens URL directly, no action_id.""" return { "type": "button", "text": {"type": "plain_text", "text": text, "emoji": True}, "url": url, } def _actions(elements: list[dict]) -> dict: """Actions block; Slack limit 25 elements, we keep <=5 per block.""" return {"type": "actions", "elements": elements[:25]} def _delta_str(now: int, prev: int) -> str: """Format a vs-yesterday delta as ▲+N or ▼-N or ~0.""" diff = now - prev if diff > 0: return f"▲+{diff}" if diff < 0: return f"▼{diff}" return "~0" def _fmt_date(dt: str) -> str: """Format 'YYYY-MM-DD' as 'May 28, 2026' for display.""" try: from datetime import date d = date.fromisoformat(dt) # %-d is Linux-specific; strip the leading zero portably instead. return f"{d.strftime('%b')} {d.day}, {d.year}" except Exception: return dt # --------------------------------------------------------------------------- # Public surface builders # --------------------------------------------------------------------------- def build_daily_summary( today: dict, yesterday: dict | None, dashboard_url: str, ) -> list[dict]: """Build the daily summary post blocks. Returns a list of Block Kit block dicts ready to pass to ``blocks=`` in a ``chat.postMessage`` call. Always stays well under 100 blocks. Args: today: The contents of today's ``summary.json``. yesterday: The contents of yesterday's ``summary.json``, or None when there is no previous snapshot (first run, etc.). dashboard_url: The Grafana dashboard URL for the link button. """ blocks: list[dict] = [] dt = today.get("dt", "") date_label = _fmt_date(dt) if dt else "Today" # ------------------------------------------------------------------ # 1. Header # ------------------------------------------------------------------ blocks.append(_header(f"APM Work Orders — {date_label}")) # ------------------------------------------------------------------ # 2. Context line: classified total + vs-yesterday deltas # ------------------------------------------------------------------ classified = today.get("classified_total", 0) blank = today.get("blank_comment_rows", 0) escalation = today.get("escalation_total", 0) action = today.get("action_needed", 0) routine = today.get("routine", 0) if yesterday is not None: delta_classified = _delta_str(classified, yesterday.get("classified_total", 0)) delta_escalation = _delta_str(escalation, yesterday.get("escalation_total", 0)) delta_action = _delta_str(action, yesterday.get("action_needed", 0)) context_text = ( f"*{classified}* WOs classified ({blank} blank-comment rows excluded) " f"| Escalations {delta_escalation} | Action-needed {delta_action} " f"| vs yesterday: total {delta_classified}" ) else: context_text = ( f"*{classified}* WOs classified ({blank} blank-comment rows excluded) " f"| No prior day for delta comparison" ) blocks.append(_section(context_text)) blocks.append(_divider()) # ------------------------------------------------------------------ # 3. Escalation breakdown # ------------------------------------------------------------------ category_counts: dict[str, int] = today.get("category_counts", {}) third_count = today.get( "third_escalation_count", category_counts.get("3rd Escalation", 0) ) esc_lines: list[str] = [] for cat in ESCALATION_CATEGORIES: count = category_counts.get(cat, 0) if count == 0: continue prefix = "*" if cat == "3rd Escalation" else "" suffix = "*" if cat == "3rd Escalation" else "" esc_lines.append(f"{prefix}{cat}: {count}{suffix}") if esc_lines: blocks.append(_section("*Escalations*\n" + "\n".join(esc_lines))) else: blocks.append(_section("*Escalations*\nNone today")) # ------------------------------------------------------------------ # 4. Action-needed vs routine fields # ------------------------------------------------------------------ blocks.append( _fields_section( [ f"*Action-needed*\n{action}", f"*Routine*\n{routine}", f"*3rd Escalations*\n{third_count}", f"*Total classified*\n{classified}", ] ) ) blocks.append(_divider()) # ------------------------------------------------------------------ # 5. Top 5 sites # ------------------------------------------------------------------ top_sites: list[dict] = today.get("top_sites", [])[:5] if top_sites: site_lines = "\n".join( f"{i + 1}. *{s['site']}* — {s['count']} WOs" for i, s in enumerate(top_sites) ) blocks.append(_section(f"*Top sites*\n{site_lines}")) blocks.append(_divider()) # ------------------------------------------------------------------ # 6. Mismatch callout (only when mismatches present) # ------------------------------------------------------------------ mismatches: list[dict] = today.get("mismatches", []) if mismatches: mm_count = len(mismatches) blocks.append( _section( f":warning: *{mm_count} mismatch{'es' if mm_count != 1 else ''} " f"flagged* — comment intent contradicts structured state. " f"See the Grafana mismatch panel for details." ) ) blocks.append(_divider()) # ------------------------------------------------------------------ # 7. Drill buttons (action_id="drill_category") + dashboard link button # ------------------------------------------------------------------ # Pick the most-common non-routine categories up to 4 slots; the 5th slot # is always the dashboard link button, keeping each actions block <=5. drill_cats: list[tuple[str, int]] = [] for cat in DRILL_CATEGORY_PRIORITY: count = category_counts.get(cat, 0) if count > 0: drill_cats.append((cat, count)) if len(drill_cats) >= 4: break button_elements: list[dict] = [] for cat, count in drill_cats: short_label = cat.replace(" Escalation", " Esc.").replace("Awaiting ", "") short_label = short_label[:20] # button text kept concise button_elements.append( _button(f"{short_label} ({count})", "drill_category", cat) ) # Dashboard link button always present (no action_id — url button). button_elements.append(_link_button("Open dashboard", dashboard_url)) blocks.append(_actions(button_elements)) # ------------------------------------------------------------------ # 8. Footer context # ------------------------------------------------------------------ generated_at = today.get("generated_at", "") footer_parts = [f"Generated {generated_at}" if generated_at else ""] footer_parts.append(f"dt={dt}") blocks.append(_context(" | ".join(p for p in footer_parts if p))) # Safety check — should never fire in practice given the bounded sections # above, but surface a warning in the footer rather than silently truncate. if len(blocks) > 95: # Trim from the middle, preserve header and footer. over = len(blocks) - 95 blocks = blocks[:3] + blocks[3 + over :] blocks.append( _context(f"+{over} block(s) trimmed — view full breakdown in Grafana.") ) return blocks def build_escalation_alert(escalations: list[dict]) -> list[dict] | None: """Build the batched 3rd-escalation alert blocks. Zero-3rd suppression: returns None when the list is empty. The caller must check for None before posting — do NOT post a message with no content. Args: escalations: Rows from ``details.json`` pre-filtered to ``category == "3rd Escalation"``. Returns: A list of Block Kit block dicts, or None. """ if not escalations: return None blocks: list[dict] = [] # ------------------------------------------------------------------ # 1. Header # ------------------------------------------------------------------ count = len(escalations) blocks.append( _header(f"3rd Escalation Alert — {count} WO{'s' if count != 1 else ''}") ) # ------------------------------------------------------------------ # 2. ONE @here mention (never one ping per WO) # ------------------------------------------------------------------ blocks.append( _section( f" *{count} work order{'s' if count != 1 else ''} " f"at 3rd escalation* require immediate attention." ) ) blocks.append(_divider()) # ------------------------------------------------------------------ # 3. Compact WO list, capped at ALERT_MAX_WOS # ------------------------------------------------------------------ visible = escalations[:ALERT_MAX_WOS] overflow = len(escalations) - len(visible) for wo in visible: wo_num = wo.get("wo_number", "???") site = wo.get("site", "—") desc = wo.get("wo_description", "") # Truncate description to keep the block tidy. if len(desc) > 80: desc = desc[:77] + "..." blocks.append(_section(f"*{wo_num}* — {site} — {desc}")) if overflow > 0: blocks.append( _context(f"+{overflow} more — see the Grafana dashboard for the full list.") ) return blocks def build_wo_modal(title: str, wos: list[dict], dashboard_url: str) -> dict: """Build a ``views.open`` payload (modal) listing work orders. Caps to Slack's 100-block modal limit by rendering the first chunk of WOs and appending a "+M more" overflow context block when the list is too long. No APM deep-links — Grafana only. Args: title: Modal title (max 24 chars enforced by Slack; truncated here). wos: List of per-WO dicts from ``details.json``. dashboard_url: The Grafana dashboard URL for the overflow note link. Returns: A dict suitable for passing to ``views.open`` as the ``view=`` arg. """ # Slack enforces a 24-char modal title limit. title_text = title[:24] modal_blocks: list[dict] = [] # Each WO is one section block. Reserve space for the optional overflow # context block. max_wo = MODAL_MAX_WO_BLOCKS - 1 # one slot reserved for potential overflow visible = wos[:max_wo] overflow_count = len(wos) - len(visible) for wo in visible: wo_num = wo.get("wo_number", "???") site = wo.get("site", "—") dept = wo.get("department", "—") category = wo.get("category", "—") snippet = wo.get("last_comment", "") if len(snippet) > 150: snippet = snippet[:147] + "..." text_parts = [ f"*{wo_num}*", f"_{site}_ | {dept} | {category}", ] if snippet: text_parts.append(snippet) modal_blocks.append(_section("\n".join(text_parts))) if overflow_count > 0: modal_blocks.append( _context( f"+{overflow_count} more — <{dashboard_url}|view the full list in Grafana>." ) ) elif not modal_blocks: modal_blocks.append(_section("No work orders to display.")) return { "type": "modal", "title": {"type": "plain_text", "text": title_text, "emoji": False}, "close": {"type": "plain_text", "text": "Close", "emoji": False}, "blocks": modal_blocks, }