import re from typing import Any, Dict, List, Optional _MARKDOWN_PREFIX_RE = re.compile(r"^\s{0,3}(?:[#>*-]+|\d+[.)])\s*") _TABLE_LINE_RE = re.compile(r"^\s*\|.*\|\s*$") _SEPARATOR_LINE_RE = re.compile(r"^\s*[-=:_`~]{3,}\s*$") def _clean_topic_line(raw: Any) -> str: text = str(raw or "").strip() if not text: return "" if _SEPARATOR_LINE_RE.fullmatch(text): return "" if _TABLE_LINE_RE.fullmatch(text): return "" text = _MARKDOWN_PREFIX_RE.sub("", text).strip() return text def _clean_topic_lines(content: str) -> List[str]: rows: List[str] = [] for line in str(content or "").splitlines(): cleaned = _clean_topic_line(line) if cleaned: rows.append(cleaned) return rows def _extract_highlights(content: str) -> List[str]: rows: List[str] = [] for line in str(content or "").splitlines(): raw = str(line or "").strip() if not raw: continue cleaned = _clean_topic_line(raw) if not cleaned: continue if raw.lstrip().startswith(("-", "*")) or ":" in cleaned or ":" in cleaned: value = cleaned[:120] if value not in rows: rows.append(value) if len(rows) >= 3: break return rows def _unique_rows(rows: List[str]) -> List[str]: deduped: List[str] = [] seen = set() for row in rows: value = str(row or "").strip() if not value or value in seen: continue seen.add(value) deduped.append(value) return deduped def _build_summary_card_view(title: str, content: str) -> Dict[str, Any]: lines = _clean_topic_lines(content) fallback_title = title or (lines[0] if lines else "") summary_source = [line for line in lines if line != fallback_title] narrative_lines = [ line for line in summary_source if not line.startswith(("-", "*")) and ":" not in line and ":" not in line ] summary = " ".join((narrative_lines or summary_source)[:2]).strip() if not summary and lines: summary = lines[0] summary = summary[:220].strip() highlights = _unique_rows(_extract_highlights(content))[:3] snippet_source = _unique_rows( [line for line in summary_source if line != summary and line not in highlights] ) snippet = " ".join(snippet_source[:2]).strip()[:180].strip() return { "type": "summary_card", "title": fallback_title[:120], "summary": summary, "highlights": highlights, "snippet": snippet, } def build_topic_publish_payload(bot_id: str, packet: Dict[str, Any], message_id: Optional[int]) -> Optional[Dict[str, Any]]: packet_type = str(packet.get("type") or "").strip().upper() is_progress = bool(packet.get("is_progress")) is_tool_hint = bool(packet.get("is_tool_hint")) if packet_type == "BUS_EVENT" and is_progress: return None if packet_type == "BUS_EVENT": content = str(packet.get("content") or packet.get("text") or "").strip() else: content = str(packet.get("text") or "").strip() if not content: return None lines = _clean_topic_lines(content) title = (lines[0] if lines else content[:120]).strip() if len(title) > 120: title = f"{title[:117].rstrip()}..." source_channel = str(packet.get("channel") or "dashboard").strip().lower() or "dashboard" dedupe_key = f"{bot_id}:message:{message_id}" if message_id else "" return { "title": title, "content": content, "level": "info", "source": source_channel, "dedupe_key": dedupe_key, "is_progress": is_progress, "is_tool_hint": is_tool_hint, "view": _build_summary_card_view(title, content), }