666 lines
30 KiB
Python
666 lines
30 KiB
Python
"""
|
||
services/telegram_templates.py
|
||
Telegram 訊息模板系統 v2
|
||
|
||
═══ 訊息分類 ═══════════════════════════════════════════════
|
||
🚨 告警類 — price_alert_msg / threat_alert_msg / system_alert_msg
|
||
📊 報告類 — daily_report_msg / weekly_report_msg / monthly_report_msg
|
||
💰 決策類 — price_decision / batch_decision_msg
|
||
🤖 系統類 — deploy_msg / heal_msg / meta_analysis_msg
|
||
💡 洞察類 — triaged_alert / insight_summary_msg
|
||
═══════════════════════════════════════════════════════════
|
||
|
||
規範：
|
||
1. 所有模板使用 HTML parse_mode
|
||
2. 一律繁體中文，Agent 名稱保留英文（Hermes/NemoTron/OpenClaw/EA）
|
||
3. 每則訊息須含：標題行 / 核心數據 / 建議行動（三段式）
|
||
4. 圖文訊息使用 send_photo_with_caption（附說明文字）
|
||
"""
|
||
|
||
import io
|
||
import json
|
||
import logging
|
||
import os
|
||
from html import escape
|
||
from datetime import datetime
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
sys_log = logging.getLogger("TelegramTpl")
|
||
|
||
TELEGRAM_BOT_TOKEN_ENV = "TELEGRAM_BOT_TOKEN"
|
||
TELEGRAM_CHAT_IDS_ENV = "TELEGRAM_CHAT_IDS"
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════════════════
|
||
# 基礎工具
|
||
# ══════════════════════════════════════════════════════════════════════════════
|
||
|
||
def _get_bot_token() -> Optional[str]:
|
||
from dotenv import load_dotenv
|
||
load_dotenv()
|
||
return os.getenv(TELEGRAM_BOT_TOKEN_ENV)
|
||
|
||
|
||
def _get_chat_ids() -> list:
|
||
token = _get_bot_token()
|
||
if not token:
|
||
sys_log.warning("[TelegramTpl] %s 未設定，跳過 Telegram 通知", TELEGRAM_BOT_TOKEN_ENV)
|
||
return []
|
||
raw = os.getenv(TELEGRAM_CHAT_IDS_ENV, "[]")
|
||
try:
|
||
return json.loads(raw)
|
||
except json.JSONDecodeError:
|
||
sys_log.warning("[TelegramTpl] %s 格式錯誤，應為 JSON 陣列", TELEGRAM_CHAT_IDS_ENV)
|
||
return []
|
||
|
||
|
||
def _send_telegram_raw(text: str, chat_ids: Optional[list] = None,
|
||
reply_markup: Optional[Dict[str, Any]] = None,
|
||
parse_mode: str = "HTML") -> bool:
|
||
"""發送純文字訊息"""
|
||
import requests
|
||
token = _get_bot_token()
|
||
if not token:
|
||
return False
|
||
if chat_ids is None:
|
||
chat_ids = _get_chat_ids()
|
||
if not chat_ids:
|
||
chat_ids = [-1003940688311] # fallback
|
||
|
||
url = f"https://api.telegram.org/bot{token}/sendMessage"
|
||
payload = {"chat_id": chat_ids[0], "text": text, "parse_mode": parse_mode}
|
||
if reply_markup:
|
||
payload["reply_markup"] = json.dumps(reply_markup, ensure_ascii=False)
|
||
try:
|
||
r = requests.post(url, json=payload, timeout=10)
|
||
if not r.ok:
|
||
sys_log.warning("[TelegramTpl] sendMessage HTTP %s: %s", r.status_code, r.text[:200])
|
||
return False
|
||
return True
|
||
except Exception as e:
|
||
sys_log.error("[TelegramTpl] send 失敗: %s", e)
|
||
return False
|
||
|
||
|
||
def send_telegram_with_result(text: str, chat_ids: Optional[list] = None,
|
||
reply_markup: Optional[Dict[str, Any]] = None,
|
||
parse_mode: str = "HTML") -> Dict[str, Any]:
|
||
"""發送 Telegram 並回傳結果明細，供 EventRouter / AIOps 使用。"""
|
||
token = _get_bot_token()
|
||
if not token:
|
||
return {"ok": False, "sent": 0, "failed": 0, "chat_ids": [], "errors": ["token_missing"]}
|
||
|
||
if chat_ids is None:
|
||
chat_ids = _get_chat_ids()
|
||
if not chat_ids:
|
||
chat_ids = [-1003940688311]
|
||
|
||
import requests
|
||
|
||
url = f"https://api.telegram.org/bot{token}/sendMessage"
|
||
sent = 0
|
||
failed = 0
|
||
errors: List[str] = []
|
||
|
||
for chat_id in chat_ids:
|
||
payload = {"chat_id": chat_id, "text": text, "parse_mode": parse_mode}
|
||
if reply_markup:
|
||
payload["reply_markup"] = json.dumps(reply_markup, ensure_ascii=False)
|
||
try:
|
||
response = requests.post(url, json=payload, timeout=10)
|
||
if response.ok:
|
||
sent += 1
|
||
else:
|
||
failed += 1
|
||
errors.append(f"{chat_id}:HTTP {response.status_code}")
|
||
sys_log.warning("[TelegramTpl] sendMessage chat=%s HTTP %s: %s",
|
||
chat_id, response.status_code, response.text[:200])
|
||
except Exception as e:
|
||
failed += 1
|
||
errors.append(f"{chat_id}:{type(e).__name__}")
|
||
sys_log.error("[TelegramTpl] send chat=%s 失敗: %s", chat_id, e)
|
||
|
||
return {
|
||
"ok": sent > 0 and failed == 0,
|
||
"sent": sent,
|
||
"failed": failed,
|
||
"chat_ids": list(chat_ids),
|
||
"errors": errors,
|
||
}
|
||
|
||
|
||
def send_photo(photo_bytes: bytes, caption: str = "",
|
||
chat_ids: Optional[list] = None,
|
||
parse_mode: str = "HTML") -> bool:
|
||
"""發送圖片（含說明文字），供圖表報告使用"""
|
||
import requests
|
||
token = _get_bot_token()
|
||
if not token or not photo_bytes:
|
||
return False
|
||
if chat_ids is None:
|
||
chat_ids = _get_chat_ids()
|
||
if not chat_ids:
|
||
chat_ids = [-1003940688311]
|
||
|
||
url = f"https://api.telegram.org/bot{token}/sendPhoto"
|
||
try:
|
||
r = requests.post(
|
||
url,
|
||
data={"chat_id": chat_ids[0], "caption": caption[:1024],
|
||
"parse_mode": parse_mode},
|
||
files={"photo": ("chart.png", photo_bytes, "image/png")},
|
||
timeout=30,
|
||
)
|
||
if not r.ok:
|
||
sys_log.warning("[TelegramTpl] sendPhoto HTTP %s: %s", r.status_code, r.text[:200])
|
||
return False
|
||
return True
|
||
except Exception as e:
|
||
sys_log.error("[TelegramTpl] sendPhoto 失敗: %s", e)
|
||
return False
|
||
|
||
|
||
def send_report_with_charts(text_msg: str, charts: List[Optional[bytes]],
|
||
chat_ids: Optional[list] = None) -> bool:
|
||
"""先發文字報告，再逐張發送圖表"""
|
||
ok = _send_telegram_raw(text_msg, chat_ids=chat_ids)
|
||
for i, chart in enumerate(charts):
|
||
if chart:
|
||
send_photo(chart, caption=f"圖表 {i+1}/{len(charts)}", chat_ids=chat_ids)
|
||
return ok
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════════════════
|
||
# 🚨 告警類模板
|
||
# ══════════════════════════════════════════════════════════════════════════════
|
||
|
||
def price_alert_msg(threats: List[Dict], analysis_period: str = "近48h") -> str:
|
||
"""
|
||
競品削價告警（Hermes + NemoTron 偵測結果）
|
||
┌─────────────────────────────────┐
|
||
│ 🚨 競品削價告警 · N 個 SKU │
|
||
│ ─────────────────────────────── │
|
||
│ ⚠️ SKU名稱 MOMO $X → 競品 $Y │
|
||
│ 價差 +X% 業績跌幅 -Y% │
|
||
└─────────────────────────────────┘
|
||
"""
|
||
n = len(threats)
|
||
high = [t for t in threats if t.get("risk") == "HIGH"]
|
||
lines = [
|
||
f"🚨 <b>競品削價告警</b> · 偵測到 <b>{n}</b> 個 SKU [{analysis_period}]",
|
||
f"🔴 高危：{len(high)} 個 🟡 中危：{n - len(high)} 個",
|
||
"─" * 32,
|
||
]
|
||
for t in threats[:8]:
|
||
risk_icon = "🔴" if t.get("risk") == "HIGH" else "🟡"
|
||
name = str(t.get("name", t.get("sku", "")))[:22]
|
||
gap = float(t.get("gap_pct", 0))
|
||
delta = float(t.get("sales_7d_delta_pct", t.get("sales_delta", 0)))
|
||
momo_p = t.get("momo_price")
|
||
pchome_p = t.get("pchome_price")
|
||
price_str = f" MOMO <b>${momo_p:,.0f}</b> vs 競品 <b>${pchome_p:,.0f}</b>" \
|
||
if momo_p and pchome_p else ""
|
||
lines.append(
|
||
f"{risk_icon} <b>{name}</b>\n"
|
||
f" 價差 <b>{gap:+.1f}%</b> 業績週跌 <b>{delta:+.1f}%</b>{price_str}"
|
||
)
|
||
if n > 8:
|
||
lines.append(f"<i>…另有 {n-8} 個 SKU（查看完整報告）</i>")
|
||
lines += ["─" * 32,
|
||
"💡 <b>建議：</b>優先處理紅色高危品項，確認競品是否促銷或長期低價"]
|
||
return "\n".join(lines)
|
||
|
||
|
||
def threat_alert_msg(sku: str, name: str, momo_price: float,
|
||
comp_price: float, gap_pct: float,
|
||
sales_delta: float, action: str, confidence: float) -> str:
|
||
"""單品威脅告警（NemoTron 派發）"""
|
||
risk = "🔴 高危" if gap_pct > 15 else "🟡 中危" if gap_pct > 5 else "🟢 低危"
|
||
return (
|
||
f"{risk} <b>競品威脅通報</b>\n"
|
||
f"━━━━━━━━━━━━━━━━━━━━\n"
|
||
f"🏷️ <b>{name}</b> <code>{sku}</code>\n\n"
|
||
f"💴 MOMO <b>NT${momo_price:,.0f}</b>\n"
|
||
f"💴 競品（PChome）<b>NT${comp_price:,.0f}</b>\n"
|
||
f"📉 價差 <b>{gap_pct:+.1f}%</b>（我方較貴）\n"
|
||
f"📊 業績週變 <b>{sales_delta:+.1f}%</b>\n\n"
|
||
f"🤖 <b>AI 建議：</b>{action}\n"
|
||
f"📈 信心度：{confidence:.0%}\n"
|
||
f"━━━━━━━━━━━━━━━━━━━━"
|
||
)
|
||
|
||
|
||
def system_alert_msg(level: str, title: str, detail: str, source: str = "") -> str:
|
||
"""系統級告警（AIOps / 部署失敗等）"""
|
||
icons = {"critical": "🚨", "error": "❌", "warning": "⚠️", "info": "ℹ️"}
|
||
icon = icons.get(level, "⚠️")
|
||
src = f" [{source}]" if source else ""
|
||
return (
|
||
f"{icon} <b>{title}</b>{src}\n"
|
||
f"━━━━━━━━━━━━━━━━━━━━\n"
|
||
f"{detail[:600]}\n"
|
||
f"━━━━━━━━━━━━━━━━━━━━\n"
|
||
f"🕐 {datetime.now().strftime('%m/%d %H:%M')}"
|
||
)
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════════════════
|
||
# 📊 報告類模板（三種週期）
|
||
# ══════════════════════════════════════════════════════════════════════════════
|
||
|
||
def daily_report_header(date_str: str, revenue: float, wow: float,
|
||
threat_count: int, opportunity_count: int) -> str:
|
||
"""
|
||
日報標題卡（附圖前發送的文字說明）
|
||
"""
|
||
wow_icon = "📈" if wow >= 0 else "📉"
|
||
wow_color = "+" if wow >= 0 else ""
|
||
return (
|
||
f"📊 <b>EwoooC 電商日報</b> · {date_str}\n"
|
||
f"══════════════════════════\n"
|
||
f"💰 今日業績 <b>NT${revenue/10000:.1f} 萬</b>\n"
|
||
f"{wow_icon} 週同比 <b>{wow_color}{wow:.1f}%</b>\n"
|
||
f"🔴 競品威脅 <b>{threat_count}</b> 個 SKU\n"
|
||
f"🟢 市場機會 <b>{opportunity_count}</b> 個 SKU\n"
|
||
f"══════════════════════════\n"
|
||
f"🤖 <i>Hermes + NemoTron + OpenClaw 聯合分析</i>"
|
||
)
|
||
|
||
|
||
def weekly_report_header(period: str, curr_rev: float, prev_rev: float,
|
||
wow: float, top_category: str) -> str:
|
||
"""週報標題卡"""
|
||
wow_icon = "📈" if wow >= 0 else "📉"
|
||
return (
|
||
f"📊 <b>EwoooC 電商週報</b> · {period}\n"
|
||
f"══════════════════════════\n"
|
||
f"💰 本週業績 <b>NT${curr_rev/10000:.1f} 萬</b>\n"
|
||
f"📦 前週業績 <b>NT${prev_rev/10000:.1f} 萬</b>\n"
|
||
f"{wow_icon} 週成長率 <b>{wow:+.1f}%</b>\n"
|
||
f"🏆 最強品類 <b>{top_category}</b>\n"
|
||
f"══════════════════════════\n"
|
||
f"🤖 <i>OpenClaw × MCP 全景策略分析</i>"
|
||
)
|
||
|
||
|
||
def monthly_report_header(month_str: str, revenue: float, mom: float,
|
||
yoy: float, top3_categories: List[str]) -> str:
|
||
"""月報標題卡"""
|
||
mom_icon = "📈" if mom >= 0 else "📉"
|
||
yoy_icon = "🚀" if yoy >= 10 else "📈" if yoy >= 0 else "📉"
|
||
cats = " / ".join(top3_categories[:3])
|
||
return (
|
||
f"📅 <b>EwoooC 電商月報</b> · {month_str}\n"
|
||
f"══════════════════════════\n"
|
||
f"💰 月度業績 <b>NT${revenue/10000:.0f} 萬</b>\n"
|
||
f"{mom_icon} 月成長率 <b>{mom:+.1f}%</b> {yoy_icon} 年成長 <b>{yoy:+.1f}%</b>\n"
|
||
f"🏆 TOP 3 品類 <b>{cats}</b>\n"
|
||
f"══════════════════════════\n"
|
||
f"🤖 <i>OpenClaw × Hermes × MCP 月度全景洞察</i>"
|
||
)
|
||
|
||
|
||
def report_section(icon: str, title: str, lines: List[str]) -> str:
|
||
"""通用報告節段（供日/週/月報各節使用）"""
|
||
body = "\n".join(f" {l}" for l in lines)
|
||
return f"\n{icon} <b>{title}</b>\n{body}"
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════════════════
|
||
# 💰 決策類模板
|
||
# ══════════════════════════════════════════════════════════════════════════════
|
||
|
||
def price_decision(product_name: str, product_sku: str,
|
||
current_price: float, suggested_price: float,
|
||
reason: str, insight_id: Optional[int] = None) -> tuple:
|
||
"""降 / 提價決策通知（含 Inline Keyboard）"""
|
||
diff = current_price - suggested_price
|
||
action_text = f"降價 NT${diff:,.0f}" if diff > 0 else \
|
||
f"提價 NT${-diff:,.0f}" if diff < 0 else "維持現價"
|
||
direction = "📉" if diff > 0 else "📈" if diff < 0 else "➡️"
|
||
|
||
message = (
|
||
f"💰 <b>AI 定價決策建議</b>\n"
|
||
f"━━━━━━━━━━━━━━━━━━━━\n"
|
||
f"🏷️ <b>{product_name}</b> <code>{product_sku}</code>\n\n"
|
||
f"現價：<b>NT${current_price:,.0f}</b>\n"
|
||
f"建議：<b>NT${suggested_price:,.0f}</b> {direction} {action_text}\n\n"
|
||
f"💡 <b>依據：</b>{reason}\n"
|
||
)
|
||
if insight_id:
|
||
message += f"🔗 洞察 ID：<code>{insight_id}</code>\n"
|
||
message += f"━━━━━━━━━━━━━━━━━━━━"
|
||
|
||
# ADR-012: callback_data 採短 prefix（momo:pa/pr:{insight_id}）— 64-byte 安全、與 L2 handler 一致
|
||
# 降級：若無 insight_id（不應發生，但保底），以 sku 做後綴 — sku 仍可能超長，呼叫者有責任提供 insight_id
|
||
_cb_id = str(insight_id) if insight_id else f"sku_{product_sku}"[:40]
|
||
keyboard = {"inline_keyboard": [[
|
||
{"text": "✅ 確認執行", "callback_data": f"momo:pa:{_cb_id}"},
|
||
{"text": "❌ 拒絕", "callback_data": f"momo:pr:{_cb_id}"},
|
||
]]}
|
||
return message, keyboard
|
||
|
||
|
||
def batch_decision_msg(items: List[Dict], batch_id: str) -> tuple:
|
||
"""批次定價決策（多 SKU 一次確認）"""
|
||
lines = [f"💰 <b>批次定價決策</b> · 共 {len(items)} 個 SKU\n━━━━━━━━━━━━━━━━━━━━"]
|
||
for i, item in enumerate(items[:6], 1):
|
||
diff = item.get("current_price", 0) - item.get("suggested_price", 0)
|
||
direction = "📉" if diff > 0 else "📈"
|
||
lines.append(
|
||
f"{i}. {direction} <b>{item.get('name','')[:20]}</b>\n"
|
||
f" ${item.get('current_price',0):,.0f} → <b>${item.get('suggested_price',0):,.0f}</b>"
|
||
)
|
||
if len(items) > 6:
|
||
lines.append(f"<i>…另有 {len(items)-6} 個 SKU</i>")
|
||
lines.append("━━━━━━━━━━━━━━━━━━━━")
|
||
# ADR-012: 短 prefix bpa=batch_price_approve / bpr=batch_price_reject，預留 64-byte buffer
|
||
_bid = str(batch_id)[:48]
|
||
keyboard = {"inline_keyboard": [[
|
||
{"text": f"✅ 全部確認（{len(items)}項）",
|
||
"callback_data": f"momo:bpa:{_bid}"},
|
||
{"text": "❌ 取消",
|
||
"callback_data": f"momo:bpr:{_bid}"},
|
||
]]}
|
||
return "\n".join(lines), keyboard
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════════════════
|
||
# 🤖 系統類模板
|
||
# ══════════════════════════════════════════════════════════════════════════════
|
||
|
||
def deploy_msg(status: str, branch: str, commit: str, duration: str = "") -> str:
|
||
"""CI/CD 部署通知"""
|
||
icons = {"success": "✅", "failed": "❌", "started": "🚀"}
|
||
icon = icons.get(status, "ℹ️")
|
||
dur = f" 耗時 {duration}" if duration else ""
|
||
return (
|
||
f"{icon} <b>部署{status.upper()}</b>{dur}\n"
|
||
f"━━━━━━━━━━━━━━━━━━━━\n"
|
||
f"🌿 分支：<code>{branch}</code>\n"
|
||
f"🔖 Commit：<code>{commit[:12]}</code>\n"
|
||
f"🕐 {datetime.now().strftime('%m/%d %H:%M')}"
|
||
)
|
||
|
||
|
||
def heal_msg(status: str, error_type: str, target_file: str,
|
||
commit_sha: str = "", diff_lines: int = 0) -> str:
|
||
"""AiderHeal 自動修復通知"""
|
||
if status == "started":
|
||
icon, title = "🔧", "AiderHeal 啟動"
|
||
elif status == "success":
|
||
icon, title = "✅", "AiderHeal 修復完成"
|
||
elif status == "reverted":
|
||
icon, title = "🔄", "AiderHeal 自動回滾"
|
||
else:
|
||
icon, title = "❌", "AiderHeal 失敗"
|
||
|
||
lines = [f"{icon} <b>{title}</b>", "━━━━━━━━━━━━━━━━━━━━",
|
||
f"🐛 錯誤類型：<code>{error_type}</code>",
|
||
f"📄 目標檔案：<code>{target_file}</code>"]
|
||
if commit_sha:
|
||
lines.append(f"🔖 Commit：<code>{commit_sha[:12]}</code>")
|
||
if diff_lines:
|
||
lines.append(f"📝 修改行數：{diff_lines} 行")
|
||
lines += ["━━━━━━━━━━━━━━━━━━━━",
|
||
f"🕐 {datetime.now().strftime('%m/%d %H:%M')}"]
|
||
return "\n".join(lines)
|
||
|
||
|
||
def meta_analysis_msg(period: str, content: str) -> str:
|
||
"""AI 系統自我審視報告"""
|
||
return (
|
||
f"🤖 <b>AI 系統效能自我審視</b> · {period}\n"
|
||
f"══════════════════════════\n"
|
||
f"{content[:1200]}\n"
|
||
f"══════════════════════════\n"
|
||
f"<i>由 OpenClaw 定期分析 · {datetime.now().strftime('%m/%d %H:%M')}</i>"
|
||
)
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════════════════
|
||
# 💡 洞察類模板
|
||
# ══════════════════════════════════════════════════════════════════════════════
|
||
|
||
def triaged_alert(base_event: Dict[str, Any], tier_label: str,
|
||
ai_summary: str, ai_cause: Optional[str] = None,
|
||
ai_actions: Optional[list] = None,
|
||
ai_executed: Optional[list] = None) -> tuple:
|
||
"""EA L1/L2 自主執行通知（保留原有介面，升級排版）"""
|
||
event_type = base_event.get("event_type", "alert")
|
||
title = escape(str(base_event.get("title", "")))
|
||
summary = escape(str(base_event.get("summary", "")))
|
||
event_id = base_event.get("id", "unknown")
|
||
safe_ai_summary = escape(str(ai_summary or ""))
|
||
safe_ai_cause = escape(str(ai_cause or "")) if ai_cause else None
|
||
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] + [""]
|
||
|
||
trace = base_event.get("trace")
|
||
if trace:
|
||
lines.append(f"<pre>{trace[-400:]}</pre>")
|
||
|
||
# ADR-012: eig=event_ignore，event_id 截斷確保 ≤ 60 bytes（留 buffer）
|
||
_eid = str(event_id)[:52]
|
||
keyboard = {"inline_keyboard": [
|
||
[{"text": "🛑 忽略此事件",
|
||
"callback_data": f"momo:eig:{_eid}"}],
|
||
]}
|
||
return "\n".join(lines), keyboard
|
||
|
||
|
||
def insight_summary_msg(insights: List[Dict], period: str = "近24h") -> str:
|
||
"""AI 洞察摘要彙整（供定期推播）"""
|
||
n = len(insights)
|
||
lines = [
|
||
f"💡 <b>AI 洞察摘要</b> · {period} 共 {n} 筆",
|
||
"━━━━━━━━━━━━━━━━━━━━",
|
||
]
|
||
type_icons = {
|
||
"price_alert": "🔴", "recommendation": "💰", "weekly_strategy": "📊",
|
||
"meta_analysis": "🤖", "market_opportunity": "🟢", "mcp_cache": "🌐",
|
||
}
|
||
for ins in insights[:6]:
|
||
icon = type_icons.get(ins.get("insight_type", ""), "💡")
|
||
content_preview = str(ins.get("content", ""))[:80].replace("\n", " ")
|
||
conf = float(ins.get("confidence", 0))
|
||
lines.append(f"{icon} {content_preview}… <i>({conf:.0%})</i>")
|
||
if n > 6:
|
||
lines.append(f"<i>…另有 {n-6} 筆洞察</i>")
|
||
lines.append("━━━━━━━━━━━━━━━━━━━━")
|
||
return "\n".join(lines)
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════════════════
|
||
# 舊版相容介面（保留供現有程式碼調用）
|
||
# ══════════════════════════════════════════════════════════════════════════════
|
||
|
||
def alert(title: str, content: str, actions: Optional[list] = None) -> str:
|
||
msg = system_alert_msg("error", title, content)
|
||
if actions:
|
||
msg += "\n" + "\n".join(f" • {a}" for a in actions)
|
||
return msg
|
||
|
||
|
||
def warning(title: str, summary: str, details: Optional[dict] = None) -> str:
|
||
detail_str = "\n".join(f" {k}: {v}" for k, v in (details or {}).items())
|
||
return system_alert_msg("warning", title, summary + ("\n" + detail_str if detail_str else ""))
|
||
|
||
|
||
def info(title: str, module: str, content: str, time: Optional[Any] = None) -> str:
|
||
t_str = f" · {time}" if time else ""
|
||
return f"ℹ️ <b>{title}</b> [{module}]{t_str}\n\n{content}"
|
||
|
||
|
||
def success(title: str, module: str, stats: str = "") -> str:
|
||
return f"✅ <b>{title}</b> [{module}]\n{stats}"
|
||
|
||
|
||
def report(title: str, report_type: str, period: str, content_md: str) -> str:
|
||
icons = {"weekly_strategy": "📊", "daily": "📅", "monthly": "📆", "meta_analysis": "🤖"}
|
||
icon = icons.get(report_type, "📋")
|
||
return (
|
||
f"{icon} <b>{title}</b> ({report_type})\n"
|
||
f"📅 期間：{period}\n"
|
||
f"══════════════════════════\n"
|
||
f"{content_md}"
|
||
)
|
||
|
||
|
||
def _send_telegram(msg: str, chat_ids: Optional[list] = None,
|
||
reply_markup: Optional[Dict[str, Any]] = None) -> bool:
|
||
return _send_telegram_raw(msg, chat_ids=chat_ids, reply_markup=reply_markup)
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════════════════
|
||
# 決策回執模板（L2 / L3 按鈕點擊後編輯原訊息使用）
|
||
# ADR-012 Phase 4：審計留痕 — 操作者、時間、action、結果
|
||
# ══════════════════════════════════════════════════════════════════════════════
|
||
|
||
# Asia/Taipei（UTC+8）— 容器可能跑 UTC，這裡手動偏移避免依賴 tzdata
|
||
_TAIPEI_UTC_OFFSET_HOURS = 8
|
||
|
||
|
||
def _now_taipei_hhmm() -> str:
|
||
"""回傳 Asia/Taipei 的 HH:MM（容器 TZ 不可靠，手動加 8h）。"""
|
||
try:
|
||
from datetime import timedelta, timezone
|
||
tz = timezone(timedelta(hours=_TAIPEI_UTC_OFFSET_HOURS))
|
||
return datetime.now(tz).strftime("%H:%M")
|
||
except Exception:
|
||
return datetime.now().strftime("%H:%M")
|
||
|
||
|
||
def _html_escape(s: Any) -> str:
|
||
"""最小 HTML 轉義（Telegram HTML parse_mode 允許 <b><i><code><pre><a>，
|
||
故只轉 < > & 避免使用者輸入破版）。"""
|
||
text = "" if s is None else str(s)
|
||
return (text.replace("&", "&")
|
||
.replace("<", "<")
|
||
.replace(">", ">"))
|
||
|
||
|
||
def decision_result(original_text: str, action: str, operator: str,
|
||
note: Optional[str] = None, **extras: Any) -> str:
|
||
"""L2 單品定價決策回執（approve / reject 後編輯原訊息）。
|
||
|
||
Args:
|
||
original_text: 原 Telegram 訊息文字（query.message.text）
|
||
action: "approve" / "reject"
|
||
operator: 操作者顯示名稱（user.full_name 或 id_<tg_id>）
|
||
note: 選填補充說明（例：訓練保守策略）
|
||
**extras: 前向相容預留（目前忽略）
|
||
|
||
Returns:
|
||
原訊息 + 分隔線 + 稽核區塊（HTML parse_mode 安全）
|
||
"""
|
||
act = (action or "").lower()
|
||
if act == "approve":
|
||
icon, label = "✅", "已執行"
|
||
elif act == "reject":
|
||
icon, label = "❌", "已拒絕"
|
||
else:
|
||
icon, label = "ℹ️", f"已處理（{_html_escape(action)}）"
|
||
|
||
ts = _now_taipei_hhmm()
|
||
op_esc = _html_escape(operator or "unknown")
|
||
base = original_text or ""
|
||
lines = [
|
||
base,
|
||
"",
|
||
"━━━━━━━━━━━━━━━━━━━━",
|
||
f"{icon} <b>{label}</b> by <b>{op_esc}</b> at {ts}",
|
||
]
|
||
if note:
|
||
lines.append(f"📝 {_html_escape(note)}")
|
||
return "\n".join(lines)
|
||
|
||
|
||
def ops_action_result(original_text: str, action: str, operator: str,
|
||
result: Optional[Dict[str, Any]] = None,
|
||
**extras: Any) -> str:
|
||
"""L3 運維決策回執（pause1h / pause6h / retry / resume 執行後編輯原訊息）。
|
||
|
||
Args:
|
||
original_text: 原 Telegram 訊息文字
|
||
action: "pause1h" / "pause6h" / "retry" / "resume" / 其他
|
||
operator: 操作者顯示名稱
|
||
result: OPS_ACTIONS 函數回傳 dict，常見欄位：
|
||
status: "ok" / "success" / "error" / "skipped"
|
||
error: 錯誤訊息（若 status=error）
|
||
message / detail: 成功訊息
|
||
task_name, duration_min 等
|
||
**extras: 前向相容預留
|
||
|
||
Returns:
|
||
原訊息 + 分隔線 + 運維執行結果區塊
|
||
"""
|
||
action_labels = {
|
||
"pause1h": "暫停 1 小時",
|
||
"pause6h": "暫停 6 小時",
|
||
"retry": "立即重試",
|
||
"resume": "恢復執行",
|
||
"execute": "執行",
|
||
"skip": "略過",
|
||
"rollback": "回滾",
|
||
}
|
||
act_label = action_labels.get((action or "").lower(), _html_escape(action or "未知動作"))
|
||
|
||
result = result or {}
|
||
status = str(result.get("status", "")).lower()
|
||
if status in ("ok", "success", "done"):
|
||
status_icon, status_text = "✅", "成功"
|
||
elif status in ("error", "failed", "fail"):
|
||
status_icon, status_text = "❌", "失敗"
|
||
elif status == "skipped":
|
||
status_icon, status_text = "⏭️", "已略過"
|
||
else:
|
||
status_icon, status_text = "ℹ️", status or "已提交"
|
||
|
||
ts = _now_taipei_hhmm()
|
||
op_esc = _html_escape(operator or "unknown")
|
||
|
||
# 擷取結果訊息（err > message > detail）
|
||
detail_msg: Optional[str] = None
|
||
for key in ("error", "message", "detail"):
|
||
if result.get(key):
|
||
detail_msg = str(result[key])[:300]
|
||
break
|
||
|
||
lines = [
|
||
original_text or "",
|
||
"",
|
||
"━━━━━━━━━━━━━━━━━━━━",
|
||
f"🛠️ <b>運維動作：</b>{act_label}",
|
||
f"👤 操作員：<b>{op_esc}</b> 🕐 {ts}",
|
||
f"{status_icon} <b>執行結果：</b>{_html_escape(status_text)}",
|
||
]
|
||
if detail_msg:
|
||
lines.append(f"📋 {_html_escape(detail_msg)}")
|
||
|
||
# 額外關鍵欄位（task_name / duration_min 等）
|
||
task_name = result.get("task_name")
|
||
if task_name:
|
||
lines.append(f"📌 任務：<code>{_html_escape(task_name)}</code>")
|
||
duration_min = result.get("duration_min")
|
||
if duration_min:
|
||
lines.append(f"⏱️ 時長：{_html_escape(duration_min)} 分鐘")
|
||
|
||
return "\n".join(lines)
|