118 lines
3.7 KiB
Python
118 lines
3.7 KiB
Python
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),
|
||
}
|