From 1fd1622007cabba9636587ec027248d854a25976 Mon Sep 17 00:00:00 2001 From: ogt Date: Sun, 19 Apr 2026 13:54:44 +0800 Subject: [PATCH] =?UTF-8?q?feat(telegram):=20=E5=85=A8=E9=9D=A2=E5=88=87?= =?UTF-8?q?=E6=8F=9B=20HTML=20parse=5Fmode=20+=20=E4=B8=89=E5=B1=A4?= =?UTF-8?q?=E5=BC=8F=E8=A6=96=E8=A6=BA=E5=88=86=E9=9A=94?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 起因：Markdown 舊版 parse_mode 導致 \[Demo] / task\_name 反斜線外漏， 且三層結構（事件資訊 / AI 加工區 / 原始技術細節）分隔線不夠明顯。 切換 HTML parse_mode（只需 escape & < >，不會有反斜線副作用）: - telegram_templates.py 全模板重寫為 HTML * 粗體 / module / 
trace
* 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 --- routes/bot_api_routes.py | 2 +- services/event_router.py | 126 +++++---- services/openclaw_strategist_service.py | 4 +- services/telegram_bot_service.py | 6 +- services/telegram_templates.py | 342 +++++++++++++++--------- 5 files changed, 290 insertions(+), 190 deletions(-) diff --git a/routes/bot_api_routes.py b/routes/bot_api_routes.py index 76a0ebc..11d0a74 100644 --- a/routes/bot_api_routes.py +++ b/routes/bot_api_routes.py @@ -782,7 +782,7 @@ def price_decision_notify(): resp = requests.post(tg_url, json={ "chat_id": row[0], "text": message, - "parse_mode": "Markdown", + "parse_mode": "HTML", "reply_markup": keyboard, }, timeout=10) if resp.ok: diff --git a/services/event_router.py b/services/event_router.py index bc10611..ef11102 100644 --- a/services/event_router.py +++ b/services/event_router.py @@ -129,6 +129,21 @@ def dispatch(event: dict, admin_chat_ids: list[int] | None = None) -> dict: # ===================================================================== # Tier 渲染器 # ===================================================================== +def _base_event_for_template(event: dict) -> dict: + """把 EventRouter 的 event 結構轉成 templates.triaged_alert 需要的格式""" + payload = event.get("payload") if isinstance(event.get("payload"), dict) else None + return { + "severity": event.get("severity", "warning"), + "title": event.get("title", "未命名事件"), + "module": event.get("source", "unknown"), + "status": event.get("status"), + "impact": event.get("impact"), + "summary": event.get("summary", ""), + "details": payload, + "trace": event.get("trace"), + } + + def _render_l0(event: dict) -> str: """L0 直出：根據 severity 選用對應模板""" sev = event.get("severity", "info") @@ -143,7 +158,6 @@ def _render_l0(event: dict) -> str: return tpl.info(title=title, module=module, content=summary) if sev == Severity.WARNING: return tpl.warning(title=title, module=module, summary=summary, details=details) - # alert 但降級到 L0 return tpl.alert( title=title, module=module, status=event.get("status", "未知"), @@ -154,31 +168,57 @@ def _render_l0(event: dict) -> str: ) -def _render_l1_with_fallback(event: dict) -> str: - """L1 Hermes 翻譯 stack trace。Phase 1 stub — 直接降 L0 + 標記""" - # TODO Phase 2: 呼叫 Hermes 做 stack trace 翻譯與摘要 +def _parse_hermes_json(raw: str) -> dict | None: + """解析 Hermes 回傳的 JSON，容錯 markdown fence""" + import json as _json + raw = (raw or "").strip() + if "```" in raw: + for p in raw.split("```"): + if p.strip().startswith("{"): + raw = p.strip() + break try: - ai_summary = _hermes_observe(event) # stub - if ai_summary: - return _compose_triaged(event, tier_label="L1 · Hermes", ai_summary=ai_summary) + return _json.loads(raw) + except Exception: + return None + + +def _render_l1_with_fallback(event: dict) -> str: + """L1 Hermes 翻譯 → 三層式 triaged_alert；失敗降 L0 + 🟡 標記""" + try: + parsed = _hermes_observe_parsed(event) + if parsed and parsed.get("summary"): + return tpl.triaged_alert( + base_event=_base_event_for_template(event), + tier_label="L1 · Hermes", + ai_summary=parsed.get("summary", ""), + ai_cause=parsed.get("probable_cause"), + ai_actions=parsed.get("actions") or [], + ) except Exception as e: sys_log.warning(f"[EventRouter] L1 Hermes 失敗，降 L0: {e}") - # Fallback：L0 模板 + 降級標記 - text = _render_l0(event) - return text + "\n\n🟡 _AI 分析暫不可用，以原始資料呈現_" + return _render_l0(event) + "\n\n🟡 AI 分析暫不可用，以原始資料呈現" def _render_l2_with_fallback(event: dict) -> str: - """L2 NemoTron 介入（含 tool call）。Phase 1 stub — 降 L1""" - # TODO Phase 3: 呼叫 NemoTron dispatcher，允許執行 SAFE_ACTIONS 中的 tool + """L2 NemoTron 規則式 → triaged_alert + 已執行 action；失敗降 L1""" try: - ai_result = _nemoton_investigate(event) # stub + ai_result = _nemoton_investigate(event) if ai_result: - return _compose_triaged( - event, tier_label="L2 · NemoTron", - ai_summary=ai_result.get("summary", ""), - ai_actions=ai_result.get("actions_taken", []), + # 同時跑 L1 Hermes 補齊摘要（已執行動作與摘要合併呈現） + parsed = None + try: + parsed = _hermes_observe_parsed(event) + except Exception: + pass + return tpl.triaged_alert( + base_event=_base_event_for_template(event), + tier_label="L2 · NemoTron", + ai_summary=(parsed or {}).get("summary") or ai_result.get("summary", ""), + ai_cause=(parsed or {}).get("probable_cause"), + ai_actions=(parsed or {}).get("actions") or [], + ai_executed=ai_result.get("actions_taken", []), ) except Exception as e: sys_log.warning(f"[EventRouter] L2 NemoTron 失敗，降 L1: {e}") @@ -186,19 +226,6 @@ def _render_l2_with_fallback(event: dict) -> str: return _render_l1_with_fallback(event) -def _compose_triaged(event: dict, tier_label: str, ai_summary: str, - ai_actions: list | None = None) -> str: - """三層式訊息：AI 摘要 + 原始事實 + 建議行動（ADR-012 §④）""" - base = _render_l0(event) - parts = [f"🤖 *AI 摘要（{tier_label}）：*", ai_summary, ""] - if ai_actions: - parts.append("🛠️ *AI 已執行動作：*") - for a in ai_actions: - parts.append(f"• {a}") - parts.append("") - return base + "\n\n" + "\n".join(parts) - - # ===================================================================== # L1 Hermes Observer（Phase 2 實作） # ===================================================================== @@ -219,8 +246,12 @@ _HERMES_OBSERVE_PROMPT = """你是一個 SRE 助手，任務是把技術錯誤 """ -def _hermes_observe(event: dict) -> str | None: - """呼叫 Hermes（Ollama）翻譯 stack trace 為人類摘要。失敗回 None 讓上層降級。""" +def _hermes_observe_parsed(event: dict) -> dict | None: + """ + 呼叫 Hermes（Ollama）翻譯 stack trace，回傳結構化 dict: + {summary, probable_cause, actions[]} + 失敗回 None 讓上層降級到 L0 + 🟡 標記。 + """ try: user_prompt = ( f"事件類型：{event.get('event_type', 'unknown')}\n" @@ -245,28 +276,15 @@ def _hermes_observe(event: dict) -> str | None: return None raw = (resp.json().get("response") or "").strip() - # 容錯：Hermes 可能多出 markdown fence - if "```" in raw: - parts = raw.split("```") - for p in parts: - if p.strip().startswith("{"): - raw = p.strip() - break + parsed = _parse_hermes_json(raw) + if not parsed or not parsed.get("summary"): + return None - import json as _json - parsed = _json.loads(raw) - summary = parsed.get("summary", "").strip() - cause = parsed.get("probable_cause", "").strip() - actions = parsed.get("actions", []) or [] - - out = [summary] - if cause: - out.append(f"\n*可能根因：* {cause}") - if actions: - out.append("\n*建議動作：*") - for a in actions[:3]: - out.append(f"• {a}") - return "\n".join(out) + return { + "summary": str(parsed.get("summary", "")).strip(), + "probable_cause": str(parsed.get("probable_cause") or "").strip() or None, + "actions": [str(a).strip() for a in (parsed.get("actions") or []) if a][:5], + } except Exception as e: sys_log.warning(f"[EventRouter.L1] Hermes 呼叫失敗，降級：{type(e).__name__}: {str(e)[:120]}") return None @@ -348,7 +366,7 @@ def _send(text: str, admin_chat_ids: list[int] | None) -> dict: for cid in admin_chat_ids: try: r = requests.post(url, json={ - "chat_id": int(cid), "text": text, "parse_mode": "Markdown", + "chat_id": int(cid), "text": text, "parse_mode": "HTML", }, timeout=10) if r.ok: sent += 1 diff --git a/services/openclaw_strategist_service.py b/services/openclaw_strategist_service.py index 1c473de..91b5d5a 100644 --- a/services/openclaw_strategist_service.py +++ b/services/openclaw_strategist_service.py @@ -321,7 +321,7 @@ def _send_price_decision_requests(recs: list, period_str: str, source_insight_id resp = requests.post(tg_url, json={ "chat_id": chat_id, "text": msg, - "parse_mode": "Markdown", + "parse_mode": "HTML", "reply_markup": keyboard, }, timeout=10) if not resp.ok: @@ -354,7 +354,7 @@ def _notify_telegram_group(report_md: str, period_str: str, report_type: str = " try: requests.post( f"https://api.telegram.org/bot{bot_token}/sendMessage", - json={"chat_id": chat_id, "text": msg, "parse_mode": "Markdown"}, + json={"chat_id": chat_id, "text": msg, "parse_mode": "HTML"}, timeout=10, ) sys_log.info(f"[OCStrategist] Telegram {report_type}推送成功") diff --git a/services/telegram_bot_service.py b/services/telegram_bot_service.py index ba46786..6f8fe15 100644 --- a/services/telegram_bot_service.py +++ b/services/telegram_bot_service.py @@ -501,7 +501,7 @@ class TrendTelegramBot: from services.telegram_templates import decision_result await query.edit_message_text( decision_result(query.message.text or "", "approve", operator), - parse_mode='Markdown' + parse_mode='HTML' ) async def _handle_price_reject(self, query, insight_id_str: str): @@ -536,7 +536,7 @@ class TrendTelegramBot: query.message.text or "", "reject", operator, note="已記錄為保守策略訓練資料" ), - parse_mode='Markdown' + parse_mode='HTML' ) async def _handle_ops_callback(self, query, data: str): @@ -584,7 +584,7 @@ class TrendTelegramBot: await query.edit_message_text( ops_action_result(query.message.text or "", action, operator, result), - parse_mode='Markdown' + parse_mode='HTML' ) async def _show_trend_by_category(self, query, category: str): diff --git a/services/telegram_templates.py b/services/telegram_templates.py index cae2810..5f8cded 100644 --- a/services/telegram_templates.py +++ b/services/telegram_templates.py @@ -1,34 +1,39 @@ """ -Telegram 訊息模板庫（EwoooC 統一格式規範） +Telegram 訊息模板庫（EwoooC 統一格式規範 v2 · HTML） 設計原則： -1. 純函數（pure function）— 回傳字串/dict，不依賴 bot instance，scheduler 和 telegram-bot 都能用 -2. 六類訊息各自模板：🚨告警 / ⚠️警告 / ℹ️資訊 / ✅成功 / 📊報告 / 💰決策 -3. 使用者輸入經 _escape_md() 處理，避免 Markdown 特殊字元破版 -4. 訊息過長（>3500 chars）自動截斷並加 "…(已截斷)" -5. callback_data 必用專案 prefix（momo:xxx），避免共用 bot 與 AWOOOI 撞車（ADR-011） +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 -DIV = "━" * 20 -PROJECT_TAG = "EwoooC" # 顯示於每則訊息，跨專案共用 bot 時識別來源（ADR-011） +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 _escape_md(s: Any) -> str: - """Escape Telegram 舊版 Markdown 特殊字元（避免 user input 破版）""" - s = str(s) if s is not None else "" - for ch in ("\\", "`", "*", "_", "["): - s = s.replace(ch, "\\" + ch) - return s +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: @@ -38,14 +43,34 @@ def _clip(text: str) -> str: def _tail(text: str, limit: int = 400) -> str: - """取末段 — stack trace 通常末端才是根因""" + """取末段 — 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} [{PROJECT_TAG} {category}] {_esc(title)}\n" + f"🕐 {_ts(time)} 📦 {_esc(module)}\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"• {_esc(k)}：{_esc(v)}") + return "\n".join(out) + + # ===================================================================== -# 🚨 告警（P0/P1）— 致命錯誤，需立即處理 +# 🚨 告警（P0/P1） # ===================================================================== def alert( title: str, @@ -57,25 +82,26 @@ def alert( trace: str | None = None, time: datetime | None = None, ) -> str: - out = [ - f"🚨 *[{PROJECT_TAG} 告警] {_escape_md(title)}*", - f"🕐 {_ts(time)} 📦 {_escape_md(module)}", - DIV, - f"❌ 狀態：{_escape_md(status)}", - f"📍 影響：{_escape_md(impact)}", - f"💬 {_escape_md(summary)}", - ] + parts = [_header("🚨", "告警", title, module, time)] + parts.append(f"\n❌ 狀態：{_esc(status)}") + parts.append(f"📍 影響：{_esc(impact)}") + parts.append(f"💬 {_esc(summary)}") + if actions: - out += ["", "🔧 *建議行動：*"] - out += [f"• {_escape_md(a)}" for a in actions] + parts.append(f"\n🔧 建議行動") + for a in actions: + parts.append(f" • {_esc(a)}") + if trace: - # Trace 不 escape（在 code block 裡），只截尾部 - out += ["", "🔍 詳細錯誤（末段）：", f"```\n{_tail(trace)}\n```"] - return _clip("\n".join(out)) + parts.append(f"\n{H_DIV}") + parts.append(f"🔍 原始技術細節（末段）") + parts.append(f"
{_esc(_tail(trace))}
") + + return _clip("\n".join(parts)) # ===================================================================== -# ⚠️ 警告（P2）— 異常但系統繼續運作 +# ⚠️ 警告（P2） # ===================================================================== def warning( title: str, @@ -84,32 +110,29 @@ def warning( details: dict[str, Any] | None = None, time: datetime | None = None, ) -> str: - out = [ - f"⚠️ *[{PROJECT_TAG} 警告] {_escape_md(title)}*", - f"🕐 {_ts(time)} 📦 {_escape_md(module)}", - "", - f"📌 {_escape_md(summary)}", - ] - if details: - out.append("") - for k, v in details.items(): - out.append(f"• *{_escape_md(k)}*：{_escape_md(v)}") - return _clip("\n".join(out)) + 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"ℹ️ *[{PROJECT_TAG} 資訊] {_escape_md(title)}*\n" - f"🕐 {_ts(time)} 📦 {_escape_md(module)}\n\n" - f"{_escape_md(content)}" + f"{_header('ℹ️', '資訊', title, module, time)}\n" + f"\n{_esc(content)}" ) # ===================================================================== -# ✅ 成功（任務完成、部署成功） +# ✅ 成功 # ===================================================================== def success( title: str, @@ -119,17 +142,14 @@ def success( detail: str | None = None, time: datetime | None = None, ) -> str: - out = [ - f"✅ *[{PROJECT_TAG} 完成] {_escape_md(title)}*", - f"🕐 {_ts(time)} 📦 {_escape_md(module)}", - ] + parts = [_header("✅", "完成", title, module, time)] if stats: - out.append(f"📊 {_escape_md(stats)}") + parts.append(f"\n📊 {_esc(stats)}") if duration: - out.append(f"⏱️ 耗時：{_escape_md(duration)}") + parts.append(f"⏱️ 耗時：{_esc(duration)}") if detail: - out += ["", _escape_md(detail)] - return _clip("\n".join(out)) + parts.append(f"\n{_esc(detail)}") + return _clip("\n".join(parts)) # ===================================================================== @@ -144,22 +164,94 @@ def report( time: datetime | None = None, ) -> str: """ - content_md / citations 是已組好的 Markdown（例如 Gemini 原文），不再 escape。 - 呼叫端需自行確保內容乾淨。 + content_md 保留原始 Markdown（Gemini 輸出），但會把 `*` `_` `[]` 轉成 HTML 等價。 + - **粗體** → 粗體 + - *斜體* → 斜體 + - 其他純文本 escape HTML """ - out = [ - f"📊 *[{PROJECT_TAG} {_escape_md(report_type)}] {_escape_md(title)}*", - f"🕐 {_ts(time)} 🗓️ {_escape_md(period)}", + # 簡化：只做最基本的 & < > escape，讓 Gemini 原生文字可讀即可 + content_html = _esc(content_md) + + parts = [ + f"📊 [{PROJECT_TAG} {_esc(report_type)}] {_esc(title)}", + f"🕐 {_ts(time)} 🗓️ {_esc(period)}", + H_DIV, "", - content_md, + content_html, ] if citations: - out += ["", DIV, f"📚 {citations}"] - return _clip("\n".join(out)) + parts += ["", H_DIV, f"📚 {_esc(citations)}"] + return _clip("\n".join(parts)) # ===================================================================== -# 💰 決策請求（降價/批准）— 含 inline keyboard +# 🤖 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❌ 狀態：{_esc(base_event['status'])}") + if base_event.get("impact"): + parts.append(f"📍 影響：{_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"🤖 AI 分析（{_esc(tier_label)}）") + parts.append("") + parts.append(f"📝 技術根因翻譯") + parts.append(_esc(ai_summary)) + if ai_cause: + parts.append("") + parts.append(f"🔎 可能原因") + parts.append(_esc(ai_cause)) + if ai_actions: + parts.append("") + parts.append(f"🔧 建議動作") + for i, a in enumerate(ai_actions[:5], 1): + parts.append(f" {i}. {_esc(a)}") + if ai_executed: + parts.append("") + parts.append(f"⚡ AI 已自動執行") + 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"🔍 原始技術細節（末段）") + parts.append(f"
{_esc(_tail(trace))}
") + + return _clip("\n".join(parts)) + + +# ===================================================================== +# 💰 降價決策請求（P2/P3） # ===================================================================== def price_decision( product_name: str, @@ -171,36 +263,53 @@ def price_decision( report_url: str | None = None, time: datetime | None = None, ) -> tuple[str, dict]: - """回傳 (text, inline_keyboard_dict)""" drop_pct = (current_price - suggested_price) / current_price * 100 if current_price > 0 else 0 - text = ( - f"💰 *[{PROJECT_TAG} 決策請求] 降價建議*\n" - f"🕐 {_ts(time)} 📦 OpenClaw Strategist\n" - f"{DIV}\n" - f"🏷️ 商品：{_escape_md(product_name)}\n" - f"📦 貨號：`{_escape_md(product_sku or 'N/A')}`\n" - f"💵 現價：${current_price:,.0f}\n" - f"📉 建議降至：${suggested_price:,.0f}（↓{drop_pct:.1f}%）\n\n" - f"🤖 *AI 理由：*\n{_escape_md(reason)}" - ) + text = "\n".join([ + f"💰 [{PROJECT_TAG} 決策請求] 降價建議", + f"🕐 {_ts(time)} 📦 OpenClaw Strategist", + H_DIV, + "", + f"🏷️ 商品：{_esc(product_name)}", + f"📦 貨號：{_esc(product_sku or 'N/A')}", + f"💵 現價：${current_price:,.0f}", + f"📉 建議降至：${suggested_price:,.0f}（↓{drop_pct:.1f}%）", + "", + f"🤖 AI 理由", + _esc(reason), + ]) keyboard = { - "inline_keyboard": [ - [ - {"text": "✅ 批准降價", "callback_data": f"{CB_PREFIX}pa:{insight_id}"}, - {"text": "❌ 拒絕", "callback_data": f"{CB_PREFIX}pr:{insight_id}"}, - ] - ] + "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} {label}", + f"👤 操作人：{_esc(operator)}", + f"🕐 {_ts()}", + ] + if note: + footer.append(f"📝 {_esc(note)}") + return _clip(original_text + "\n".join(footer)) + + # ===================================================================== -# 🛠️ 決策結果回執（管理員按下按鈕後，訊息會被 edit 成這個） -# ===================================================================== -# ===================================================================== -# 🛠️ L3 Ops Action Request — 附 pause/force_retry 按鈕的運維決策訊息 +# 🛠️ L3 Ops Action Request（Phase 4 HITL） # ===================================================================== def ops_action_request( task_name: str, @@ -209,21 +318,17 @@ def ops_action_request( context: dict | None = None, time: datetime | None = None, ) -> tuple[str, dict]: - """ - L3 HITL 運維決策訊息。NemoTron 多次重試失敗 / 告警未消 → 送此訊息請管理員裁決。 - 回傳 (text, inline_keyboard) — callback_data 用 momo:ops:: 格式 - """ - out = [ - f"🛠️ *[{PROJECT_TAG} 運維決策] {_escape_md(title)}*", - f"🕐 {_ts(time)} 📦 {_escape_md(task_name)}", - DIV, - f"💬 {_escape_md(reason)}", + parts = [ + f"🛠️ [{PROJECT_TAG} 運維決策] {_esc(title)}", + f"🕐 {_ts(time)} 📦 {_esc(task_name)}", + H_DIV, + "", + f"💬 {_esc(reason)}", ] if context: - out.append("") - for k, v in context.items(): - out.append(f"• *{_escape_md(k)}*：{_escape_md(v)}") - out += ["", "👉 *請選擇動作：*"] + parts.append("") + parts.append(_details_block(context)) + parts += ["", "👉 請選擇動作"] keyboard = { "inline_keyboard": [ @@ -237,53 +342,30 @@ def ops_action_request( ], ] } - return _clip("\n".join(out)), keyboard + return _clip("\n".join(parts)), keyboard def ops_action_result( original_text: str, - action: str, # pause1h / pause6h / retry / resume + action: str, operator: str, result: dict, ) -> str: - """Ops callback 執行完，edit 原訊息顯示結果""" emoji_map = {"pause1h": "⏸️", "pause6h": "⏸️", "retry": "⚡", "resume": "▶️"} - label_map = { - "pause1h": "已暫停 1 小時", "pause6h": "已暫停 6 小時", - "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 = [ "", - DIV, - f"{emoji} *{label}*（狀態：{_escape_md(status)}）", - f"👤 操作人：{_escape_md(operator)}", + H_DIV, + f"{emoji} {label}（狀態：{_esc(status)}）", + f"👤 操作人：{_esc(operator)}", f"🕐 {_ts()}", ] if status == "rejected": - footer.append(f"⚠️ 拒絕原因：{_escape_md(result.get('reason', ''))}") + footer.append(f"⚠️ 拒絕原因：{_esc(result.get('reason', ''))}") elif status == "deferred": - footer.append(f"ℹ️ {_escape_md(result.get('note', ''))}") - return _clip(original_text + "\n".join(footer)) - - -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 = [ - "", - DIV, - f"{emoji} *{label}*", - f"👤 操作人：{_escape_md(operator)}", - f"🕐 {_ts()}", - ] - if note: - footer.append(f"📝 {_escape_md(note)}") + footer.append(f"ℹ️ {_esc(result.get('note', ''))}") return _clip(original_text + "\n".join(footer))