1fd1622007
All checks were successful
CD Pipeline / deploy (push) Successful in 1m12s
起因：Markdown 舊版 parse_mode 導致 \[Demo] / task\_name 反斜線外漏，
且三層結構（事件資訊 / AI 加工區 / 原始技術細節）分隔線不夠明顯。
切換 HTML parse_mode（只需 escape & < >，不會有反斜線副作用）:
- telegram_templates.py 全模板重寫為 HTML
* <b>粗體</b> / <code>module</code> / <pre>trace</pre>
* H_DIV (━×20) 節間強分隔 / L_DIV (─×18) 節內弱分隔
* 新增 triaged_alert() 實作 ADR-012 §④ 三層式結構
[事件資訊] → ━━━ → [🤖 AI 分析] → ━━━ → [🔍 原始技術細節]
event_router.py:
- _hermes_observe_parsed() 回結構化 dict {summary, cause, actions}
取代舊的字串版本
- _render_l1/l2_with_fallback 改用 tpl.triaged_alert() 統一格式
- _send() parse_mode 改 HTML
Call sites 同步改 HTML:
- routes/bot_api_routes.py price_decision_notify
- services/openclaw_strategist_service.py 兩個發送處
- services/telegram_bot_service.py 三個 edit_message_text
(_handle_price_approve / _handle_price_reject / _handle_ops_callback)
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
372 lines
12 KiB
Python
372 lines
12 KiB
Python
"""
|
||
Telegram 訊息模板庫（EwoooC 統一格式規範 v2 · HTML）
|
||
|
||
設計原則：
|
||
1. 純函數 — scheduler / telegram-bot / event_router 都能呼叫
|
||
2. 六類訊息 + 三個 HITL 變體：🚨 告警 / ⚠️ 警告 / ℹ️ 資訊 / ✅ 成功 / 📊 報告 / 💰 決策 / 🛠️ Ops
|
||
3. 使用 Telegram HTML parse_mode（相容性最好，只 escape & < >，不會有反斜線 escape 破版）
|
||
4. 三層式結構：事件資訊 / 🤖 AI 加工區 / 🔍 原始技術細節 — 明確分隔線區隔
|
||
5. callback_data 必用 momo: prefix（ADR-011）
|
||
6. 訊息 >3500 chars 自動截斷
|
||
|
||
呼叫端發送時務必使用 `parse_mode='HTML'`
|
||
"""
|
||
|
||
from datetime import datetime
|
||
from typing import Any
|
||
|
||
MAX_LEN = 3500
|
||
H_DIV = "━" * 20 # 強分隔線（節與節之間）
|
||
L_DIV = "─" * 18 # 弱分隔線（AI 區內部）
|
||
PROJECT_TAG = "EwoooC" # 跨專案共用 bot 識別來源（ADR-011）
|
||
CB_PREFIX = "momo:"
|
||
PARSE_MODE = "HTML" # 統一 parse_mode
|
||
|
||
|
||
def _ts(dt: datetime | None = None) -> str:
|
||
return (dt or datetime.now()).strftime("%Y-%m-%d %H:%M")
|
||
|
||
|
||
def _esc(s: Any) -> str:
|
||
"""Escape HTML 特殊字元（Telegram HTML 只認 & < >）"""
|
||
if s is None:
|
||
return ""
|
||
return (str(s).replace("&", "&")
|
||
.replace("<", "<")
|
||
.replace(">", ">"))
|
||
|
||
|
||
def _clip(text: str) -> str:
|
||
if len(text) <= MAX_LEN:
|
||
return text
|
||
return text[: MAX_LEN - 20] + "\n…（已截斷）"
|
||
|
||
|
||
def _tail(text: str, limit: int = 400) -> str:
|
||
"""取末段 — stack trace 根因通常在末端"""
|
||
if len(text) <= limit:
|
||
return text
|
||
return "…\n" + text[-limit:]
|
||
|
||
|
||
def _header(emoji: str, category: str, title: str, module: str,
|
||
time: datetime | None = None) -> str:
|
||
"""統一標題區：emoji + 分類 + 標題 + 時間/模組"""
|
||
return (
|
||
f"{emoji} <b>[{PROJECT_TAG} {category}] {_esc(title)}</b>\n"
|
||
f"🕐 {_ts(time)} 📦 <code>{_esc(module)}</code>\n"
|
||
f"{H_DIV}"
|
||
)
|
||
|
||
|
||
def _details_block(details: dict[str, Any] | None) -> str:
|
||
"""結構化明細區塊"""
|
||
if not details:
|
||
return ""
|
||
out = []
|
||
for k, v in details.items():
|
||
out.append(f"• <b>{_esc(k)}</b>：{_esc(v)}")
|
||
return "\n".join(out)
|
||
|
||
|
||
# =====================================================================
|
||
# 🚨 告警（P0/P1）
|
||
# =====================================================================
|
||
def alert(
|
||
title: str,
|
||
module: str,
|
||
status: str,
|
||
impact: str,
|
||
summary: str,
|
||
actions: list[str] | None = None,
|
||
trace: str | None = None,
|
||
time: datetime | None = None,
|
||
) -> str:
|
||
parts = [_header("🚨", "告警", title, module, time)]
|
||
parts.append(f"\n❌ <b>狀態</b>：{_esc(status)}")
|
||
parts.append(f"📍 <b>影響</b>：{_esc(impact)}")
|
||
parts.append(f"💬 {_esc(summary)}")
|
||
|
||
if actions:
|
||
parts.append(f"\n🔧 <b>建議行動</b>")
|
||
for a in actions:
|
||
parts.append(f" • {_esc(a)}")
|
||
|
||
if trace:
|
||
parts.append(f"\n{H_DIV}")
|
||
parts.append(f"🔍 <b>原始技術細節（末段）</b>")
|
||
parts.append(f"<pre>{_esc(_tail(trace))}</pre>")
|
||
|
||
return _clip("\n".join(parts))
|
||
|
||
|
||
# =====================================================================
|
||
# ⚠️ 警告（P2）
|
||
# =====================================================================
|
||
def warning(
|
||
title: str,
|
||
module: str,
|
||
summary: str,
|
||
details: dict[str, Any] | None = None,
|
||
time: datetime | None = None,
|
||
) -> str:
|
||
parts = [_header("⚠️", "警告", title, module, time)]
|
||
parts.append(f"\n📌 {_esc(summary)}")
|
||
|
||
db = _details_block(details)
|
||
if db:
|
||
parts.append("")
|
||
parts.append(db)
|
||
|
||
return _clip("\n".join(parts))
|
||
|
||
|
||
# =====================================================================
|
||
# ℹ️ 資訊
|
||
# =====================================================================
|
||
def info(title: str, module: str, content: str, time: datetime | None = None) -> str:
|
||
return _clip(
|
||
f"{_header('ℹ️', '資訊', title, module, time)}\n"
|
||
f"\n{_esc(content)}"
|
||
)
|
||
|
||
|
||
# =====================================================================
|
||
# ✅ 成功
|
||
# =====================================================================
|
||
def success(
|
||
title: str,
|
||
module: str,
|
||
stats: str | None = None,
|
||
duration: str | None = None,
|
||
detail: str | None = None,
|
||
time: datetime | None = None,
|
||
) -> str:
|
||
parts = [_header("✅", "完成", title, module, time)]
|
||
if stats:
|
||
parts.append(f"\n📊 {_esc(stats)}")
|
||
if duration:
|
||
parts.append(f"⏱️ <b>耗時</b>：{_esc(duration)}")
|
||
if detail:
|
||
parts.append(f"\n{_esc(detail)}")
|
||
return _clip("\n".join(parts))
|
||
|
||
|
||
# =====================================================================
|
||
# 📊 報告（日報 / 週報 / Meta-Analysis）
|
||
# =====================================================================
|
||
def report(
|
||
title: str,
|
||
report_type: str,
|
||
period: str,
|
||
content_md: str,
|
||
citations: str | None = None,
|
||
time: datetime | None = None,
|
||
) -> str:
|
||
"""
|
||
content_md 保留原始 Markdown（Gemini 輸出），但會把 `*` `_` `[]` 轉成 HTML 等價。
|
||
- **粗體** → <b>粗體</b>
|
||
- *斜體* → <i>斜體</i>
|
||
- 其他純文本 escape HTML
|
||
"""
|
||
# 簡化：只做最基本的 & < > escape，讓 Gemini 原生文字可讀即可
|
||
content_html = _esc(content_md)
|
||
|
||
parts = [
|
||
f"📊 <b>[{PROJECT_TAG} {_esc(report_type)}] {_esc(title)}</b>",
|
||
f"🕐 {_ts(time)} 🗓️ <code>{_esc(period)}</code>",
|
||
H_DIV,
|
||
"",
|
||
content_html,
|
||
]
|
||
if citations:
|
||
parts += ["", H_DIV, f"📚 {_esc(citations)}"]
|
||
return _clip("\n".join(parts))
|
||
|
||
|
||
# =====================================================================
|
||
# 🤖 Triaged Alert — L1/L2 AI 加工訊息（ADR-012 §④ 三層式）
|
||
# =====================================================================
|
||
def triaged_alert(
|
||
base_event: dict,
|
||
tier_label: str, # "L1 · Hermes" / "L2 · NemoTron"
|
||
ai_summary: str, # Hermes 翻譯
|
||
ai_cause: str | None = None, # 可能根因
|
||
ai_actions: list[str] | None = None, # 建議動作
|
||
ai_executed: list[str] | None = None, # L2 已執行的 action（如 retry_task → scheduled）
|
||
) -> str:
|
||
"""
|
||
三層式訊息：
|
||
[事件資訊] → [🤖 AI 加工區] → [🔍 原始技術細節]
|
||
base_event 欄位：title, module, status, impact, summary, details, trace
|
||
"""
|
||
sev = base_event.get("severity", "warning")
|
||
emoji = "🚨" if sev == "alert" else "⚠️"
|
||
category = "告警" if sev == "alert" else "警告"
|
||
|
||
parts = [_header(emoji, category, base_event.get("title", ""),
|
||
base_event.get("module", "unknown"))]
|
||
|
||
# Section 1: 事件資訊
|
||
if base_event.get("status"):
|
||
parts.append(f"\n❌ <b>狀態</b>：{_esc(base_event['status'])}")
|
||
if base_event.get("impact"):
|
||
parts.append(f"📍 <b>影響</b>：{_esc(base_event['impact'])}")
|
||
if base_event.get("summary"):
|
||
parts.append(f"💬 {_esc(base_event['summary'])}")
|
||
db = _details_block(base_event.get("details"))
|
||
if db:
|
||
parts.append("")
|
||
parts.append(db)
|
||
|
||
# Section 2: AI 加工區（明顯分隔）
|
||
parts.append(f"\n{H_DIV}")
|
||
parts.append(f"🤖 <b>AI 分析（{_esc(tier_label)}）</b>")
|
||
parts.append("")
|
||
parts.append(f"📝 <b>技術根因翻譯</b>")
|
||
parts.append(_esc(ai_summary))
|
||
if ai_cause:
|
||
parts.append("")
|
||
parts.append(f"🔎 <b>可能原因</b>")
|
||
parts.append(_esc(ai_cause))
|
||
if ai_actions:
|
||
parts.append("")
|
||
parts.append(f"🔧 <b>建議動作</b>")
|
||
for i, a in enumerate(ai_actions[:5], 1):
|
||
parts.append(f" {i}. {_esc(a)}")
|
||
if ai_executed:
|
||
parts.append("")
|
||
parts.append(f"⚡ <b>AI 已自動執行</b>")
|
||
for a in ai_executed:
|
||
parts.append(f" • {_esc(a)}")
|
||
|
||
# Section 3: 原始技術細節（可選）
|
||
trace = base_event.get("trace")
|
||
if trace:
|
||
parts.append(f"\n{H_DIV}")
|
||
parts.append(f"🔍 <b>原始技術細節（末段）</b>")
|
||
parts.append(f"<pre>{_esc(_tail(trace))}</pre>")
|
||
|
||
return _clip("\n".join(parts))
|
||
|
||
|
||
# =====================================================================
|
||
# 💰 降價決策請求（P2/P3）
|
||
# =====================================================================
|
||
def price_decision(
|
||
product_name: str,
|
||
product_sku: str,
|
||
current_price: float,
|
||
suggested_price: float,
|
||
reason: str,
|
||
insight_id: int,
|
||
report_url: str | None = None,
|
||
time: datetime | None = None,
|
||
) -> tuple[str, dict]:
|
||
drop_pct = (current_price - suggested_price) / current_price * 100 if current_price > 0 else 0
|
||
text = "\n".join([
|
||
f"💰 <b>[{PROJECT_TAG} 決策請求] 降價建議</b>",
|
||
f"🕐 {_ts(time)} 📦 <code>OpenClaw Strategist</code>",
|
||
H_DIV,
|
||
"",
|
||
f"🏷️ <b>商品</b>：{_esc(product_name)}",
|
||
f"📦 <b>貨號</b>：<code>{_esc(product_sku or 'N/A')}</code>",
|
||
f"💵 <b>現價</b>：${current_price:,.0f}",
|
||
f"📉 <b>建議降至</b>：${suggested_price:,.0f}（↓{drop_pct:.1f}%）",
|
||
"",
|
||
f"🤖 <b>AI 理由</b>",
|
||
_esc(reason),
|
||
])
|
||
keyboard = {
|
||
"inline_keyboard": [[
|
||
{"text": "✅ 批准降價", "callback_data": f"{CB_PREFIX}pa:{insight_id}"},
|
||
{"text": "❌ 拒絕", "callback_data": f"{CB_PREFIX}pr:{insight_id}"},
|
||
]]
|
||
}
|
||
if report_url:
|
||
keyboard["inline_keyboard"].append([{"text": "🔗 查看報表", "url": report_url}])
|
||
return _clip(text), keyboard
|
||
|
||
|
||
def decision_result(
|
||
original_text: str,
|
||
decision: str, # "approve" or "reject"
|
||
operator: str,
|
||
note: str | None = None,
|
||
) -> str:
|
||
emoji = "✅" if decision == "approve" else "❌"
|
||
label = "已批准降價" if decision == "approve" else "已拒絕降價"
|
||
footer = [
|
||
"",
|
||
H_DIV,
|
||
f"{emoji} <b>{label}</b>",
|
||
f"👤 <b>操作人</b>：{_esc(operator)}",
|
||
f"🕐 {_ts()}",
|
||
]
|
||
if note:
|
||
footer.append(f"📝 {_esc(note)}")
|
||
return _clip(original_text + "\n".join(footer))
|
||
|
||
|
||
# =====================================================================
|
||
# 🛠️ L3 Ops Action Request（Phase 4 HITL）
|
||
# =====================================================================
|
||
def ops_action_request(
|
||
task_name: str,
|
||
title: str,
|
||
reason: str,
|
||
context: dict | None = None,
|
||
time: datetime | None = None,
|
||
) -> tuple[str, dict]:
|
||
parts = [
|
||
f"🛠️ <b>[{PROJECT_TAG} 運維決策] {_esc(title)}</b>",
|
||
f"🕐 {_ts(time)} 📦 <code>{_esc(task_name)}</code>",
|
||
H_DIV,
|
||
"",
|
||
f"💬 {_esc(reason)}",
|
||
]
|
||
if context:
|
||
parts.append("")
|
||
parts.append(_details_block(context))
|
||
parts += ["", "👉 <b>請選擇動作</b>"]
|
||
|
||
keyboard = {
|
||
"inline_keyboard": [
|
||
[
|
||
{"text": "⏸️ 暫停 1h", "callback_data": f"{CB_PREFIX}ops:pause1h:{task_name}"},
|
||
{"text": "⏸️ 暫停 6h", "callback_data": f"{CB_PREFIX}ops:pause6h:{task_name}"},
|
||
],
|
||
[
|
||
{"text": "⚡ 立即重試", "callback_data": f"{CB_PREFIX}ops:retry:{task_name}"},
|
||
{"text": "▶️ 解除暫停", "callback_data": f"{CB_PREFIX}ops:resume:{task_name}"},
|
||
],
|
||
]
|
||
}
|
||
return _clip("\n".join(parts)), keyboard
|
||
|
||
|
||
def ops_action_result(
|
||
original_text: str,
|
||
action: str,
|
||
operator: str,
|
||
result: dict,
|
||
) -> str:
|
||
emoji_map = {"pause1h": "⏸️", "pause6h": "⏸️", "retry": "⚡", "resume": "▶️"}
|
||
label_map = {"pause1h": "已暫停 1 小時", "pause6h": "已暫停 6 小時",
|
||
"retry": "已立即重試", "resume": "已解除暫停"}
|
||
emoji = emoji_map.get(action, "🛠️")
|
||
label = label_map.get(action, action)
|
||
status = result.get("status", "unknown")
|
||
footer = [
|
||
"",
|
||
H_DIV,
|
||
f"{emoji} <b>{label}</b>（狀態：<code>{_esc(status)}</code>）",
|
||
f"👤 <b>操作人</b>：{_esc(operator)}",
|
||
f"🕐 {_ts()}",
|
||
]
|
||
if status == "rejected":
|
||
footer.append(f"⚠️ <b>拒絕原因</b>：{_esc(result.get('reason', ''))}")
|
||
elif status == "deferred":
|
||
footer.append(f"ℹ️ {_esc(result.get('note', ''))}")
|
||
return _clip(original_text + "\n".join(footer))
|