[V10.289] 重排 EA HITL Telegram 告警格式 | telegram_templates.py

services/telegram_templates.py

@@ -21,6 +21,7 @@ import io
 import json
 import logging
 import os
+import re
 from html import escape
 from datetime import datetime
 from typing import Any, Dict, List, Optional
@@ -467,6 +468,183 @@ def meta_analysis_msg(period: str, content: str) -> str:
 # 💡 洞察類模板
 # ══════════════════════════════════════════════════════════════════════════════
 
+def _short_text(value: Any, limit: int = 120) -> str:
+    text = str(value or "").strip()
+    if len(text) <= limit:
+        return text
+    return text[: max(0, limit - 1)].rstrip() + "…"
+
+
+def _split_action_parts(action: Any) -> List[str]:
+    return [
+        part.strip()
+        for part in re.split(r"\s*[｜|]\s*", str(action or ""))
+        if part and part.strip()
+    ]
+
+
+def _parse_ea_action(action: Any) -> Dict[str, Any]:
+    parts = _split_action_parts(action)
+    item: Dict[str, Any] = {
+        "raw": str(action or ""),
+        "title": parts[0] if parts else str(action or ""),
+        "notes": [],
+    }
+    title_match = re.match(r"^\[([^\]]+)\]\s*(.+)$", item["title"])
+    if title_match:
+        item["sku"] = title_match.group(1).strip()
+        item["name"] = title_match.group(2).strip()
+
+    for part in parts[1:]:
+        if "MOMO" in part and "PChome" in part:
+            item["comparison"] = part
+            momo = re.search(r"MOMO\s*\$([0-9,]+)", part)
+            pchome = re.search(r"PChome\s*\$([0-9,]+)", part)
+            pct = re.search(r"\(([+-]?\d+(?:\.\d+)?)%\)", part)
+            if momo:
+                item["momo_price"] = momo.group(1)
+            if pchome:
+                item["pchome_price"] = pchome.group(1)
+            if pct:
+                item["gap_pct"] = float(pct.group(1))
+                item["gap_pct_text"] = f"{float(pct.group(1)):+.1f}%"
+        elif part.startswith("PChome "):
+            item["pchome_id"] = part.replace("PChome", "", 1).strip()
+        elif part.startswith("建議"):
+            item["action"] = part
+        elif "NT$" in part or "流失" in part or "價差" in part or "價格優勢" in part:
+            item["impact"] = part
+            amount = re.search(r"NT\$\s*([0-9,]+)", part)
+            if amount:
+                try:
+                    item["impact_amount"] = int(amount.group(1).replace(",", ""))
+                except ValueError:
+                    pass
+        else:
+            item["notes"].append(part)
+    return item
+
+
+def _format_ea_risk_summary(actions: List[Dict[str, Any]]) -> List[str]:
+    count = len(actions)
+    gap_values = [a["gap_pct"] for a in actions if isinstance(a.get("gap_pct"), (int, float))]
+    amounts = [a["impact_amount"] for a in actions if isinstance(a.get("impact_amount"), int)]
+    lines = [f"• 待審 SKU：<b>{count}</b> 件"]
+    if gap_values:
+        low, high = min(gap_values), max(gap_values)
+        if low == high:
+            lines.append(f"• 價差幅度：<b>{low:+.1f}%</b>")
+        else:
+            lines.append(f"• 價差範圍：<b>{low:+.1f}%～{high:+.1f}%</b>")
+    if amounts:
+        lines.append(f"• 最大單件價差：<b>NT$ {max(amounts):,}</b>")
+    lines.append("• 核心判斷：先確認同款 identity_v2，再決定跟價、促銷或曝光")
+    return lines
+
+
+def _format_ea_action_card(item: Dict[str, Any], index: int) -> List[str]:
+    sku = escape(str(item.get("sku") or ""))
+    name = escape(_short_text(item.get("name") or item.get("title") or "", 58))
+    heading = f"<b>{index}. [{sku}] {name}</b>" if sku else f"<b>{index}. {name}</b>"
+    lines = [heading]
+
+    momo_price = item.get("momo_price")
+    pchome_price = item.get("pchome_price")
+    if momo_price or pchome_price:
+        momo_text = f"${momo_price}" if momo_price else "—"
+        pchome_text = f"${pchome_price}" if pchome_price else "—"
+        lines.append(f"   MOMO：<b>{momo_text}</b>　PChome：<b>{pchome_text}</b>")
+
+    gap_text = item.get("gap_pct_text")
+    impact = item.get("impact")
+    if gap_text or impact:
+        impact_text = escape(str(impact or ""))
+        if gap_text and impact_text:
+            lines.append(f"   價差：<b>{escape(str(gap_text))}</b>　{impact_text}")
+        elif gap_text:
+            lines.append(f"   價差：<b>{escape(str(gap_text))}</b>")
+        else:
+            lines.append(f"   影響：{impact_text}")
+
+    action = str(item.get("action") or "").replace("建議", "", 1).strip(" ：:")
+    if action:
+        lines.append(f"   動作：{escape(_short_text(action, 86))}")
+
+    if item.get("pchome_id"):
+        lines.append(f"   PChome：<code>{escape(_short_text(item['pchome_id'], 40))}</code>")
+    return lines
+
+
+def _format_ea_escalation_alert(
+    *,
+    base_event: Dict[str, Any],
+    tier_label: str,
+    ai_summary: str,
+    ai_cause: Optional[str],
+    ai_actions: Optional[list],
+) -> str:
+    event_type = escape(str(base_event.get("event_type", "ea_escalation")))
+    title = escape(str(base_event.get("title", "EA 升級審核")))
+    summary = escape(str(base_event.get("summary", "")))
+    cause_parts = [
+        escape(part.strip())
+        for part in str(ai_cause or "").split("|")
+        if part and part.strip()
+    ]
+    parsed_actions = [_parse_ea_action(action) for action in (ai_actions or [])]
+    shown_actions = parsed_actions[:5]
+    hidden_count = max(0, len(parsed_actions) - len(shown_actions))
+
+    lines = [
+        f"⚡ <b>{escape(str(tier_label))}</b>",
+        f"📌 <b>{title}</b>",
+        f"<code>{event_type}</code>",
+        "━━━━━━━━━━━━━━━━━━━━",
+        "🧭 <b>決策狀態</b>",
+    ]
+    if summary:
+        lines.append(f"• {summary}")
+    for part in cause_parts[:3]:
+        lines.append(f"• {part}")
+
+    if ai_summary:
+        lines += [
+            "",
+            "🧠 <b>背景摘要</b>",
+            f"• {escape(_short_text(ai_summary, 280))}",
+        ]
+
+    if parsed_actions:
+        lines += [
+            "",
+            "📊 <b>風險摘要</b>",
+            *_format_ea_risk_summary(parsed_actions),
+            "",
+            "📋 <b>TOP 待審 SKU</b>",
+        ]
+        for idx, item in enumerate(shown_actions, start=1):
+            if idx > 1:
+                lines.append("")
+            lines.extend(_format_ea_action_card(item, idx))
+        if hidden_count:
+            lines.append(f"\n<i>另有 {hidden_count} 件，請至觀測台查看完整清單。</i>")
+        lines += [
+            "",
+            "✅ <b>建議處置</b>",
+            "• 先人工確認 PChome identity_v2 與規格一致",
+            "• 同款：評估跟價、組合促銷或加強 MOMO 價格優勢曝光",
+            "• 非同款：標記待審，避免進入自動調價或簡報決策",
+        ]
+    else:
+        lines += [
+            "",
+            "✅ <b>建議處置</b>",
+            "• 先確認資料來源與最近錯誤紀錄",
+            "• 補齊可審核證據後再批准執行",
+        ]
+
+    return "\n".join(lines)
+
 def triaged_alert(base_event: Dict[str, Any], tier_label: str,
                   ai_summary: str, ai_cause: Optional[str] = None,
                   ai_actions: Optional[list] = None,
@@ -481,25 +659,35 @@ def triaged_alert(base_event: Dict[str, Any], tier_label: str,
     safe_actions = [escape(str(a)) for a in (ai_actions or [])]
     safe_executed = [escape(str(a)) for a in (ai_executed or [])]
 
-    lines = [
-        f"⚡ <b>{tier_label} · {event_type}</b>",
-        f"📌 {title}",
-        "",
-    ]
-    if summary:
-        lines += [f"🔍 <b>概要：</b>{summary}", ""]
-    if safe_ai_summary:
-        lines += [f"🧠 <b>AI 摘要：</b>{safe_ai_summary[:400]}", ""]
-    if safe_ai_cause:
-        lines += [f"💡 <b>可能原因：</b>{safe_ai_cause}", ""]
-    if safe_actions:
-        lines += ["<b>📋 建議行動：</b>"] + [f"  • {a}" for a in safe_actions] + [""]
-    if safe_executed:
-        lines += ["<b>✅ 已執行：</b>"] + [f"  • {a}" for a in safe_executed] + [""]
+    if event_type == "ea_escalation":
+        message = _format_ea_escalation_alert(
+            base_event=base_event,
+            tier_label=tier_label,
+            ai_summary=str(ai_summary or ""),
+            ai_cause=ai_cause,
+            ai_actions=ai_actions,
+        )
+    else:
+        lines = [
+            f"⚡ <b>{tier_label} · {event_type}</b>",
+            f"📌 {title}",
+            "",
+        ]
+        if summary:
+            lines += [f"🔍 <b>概要：</b>{summary}", ""]
+        if safe_ai_summary:
+            lines += [f"🧠 <b>AI 摘要：</b>{safe_ai_summary[:400]}", ""]
+        if safe_ai_cause:
+            lines += [f"💡 <b>可能原因：</b>{safe_ai_cause}", ""]
+        if safe_actions:
+            lines += ["<b>📋 建議行動：</b>"] + [f"  • {a}" for a in safe_actions] + [""]
+        if safe_executed:
+            lines += ["<b>✅ 已執行：</b>"] + [f"  • {a}" for a in safe_executed] + [""]
 
-    trace = base_event.get("trace")
-    if trace:
-        lines.append(f"<pre>{trace[-400:]}</pre>")
+        trace = base_event.get("trace")
+        if trace:
+            lines.append(f"<pre>{trace[-400:]}</pre>")
+        message = "\n".join(lines)
 
     # ADR-012: eig=event_ignore，event_id 截斷確保 ≤ 60 bytes（留 buffer）
     _eid = str(event_id)[:52]
@@ -507,7 +695,7 @@ def triaged_alert(base_event: Dict[str, Any], tier_label: str,
         [{"text": "🛑 忽略此事件",
           "callback_data": f"momo:eig:{_eid}"}],
     ]}
-    return "\n".join(lines), keyboard
+    return message, keyboard
 
 
 def insight_summary_msg(insights: List[Dict], period: str = "近24h") -> str: