feat(telegram): ADR-050 P2 - detail/history info actions 實作
All checks were successful
CD Pipeline (Dev) / build-and-deploy-dev (push) Successful in 2m28s

- _send_incident_detail: 取得事件詳情 + AI 信心條形圖,傳送新訊息保留原始簽核卡片
- _send_incident_history: 頻率統計 (1h/24h/7d/30d + 自動修復次數)
- reanalyze: 保留為開發中 placeholder

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
OG T
2026-04-01 18:48:04 +08:00
parent 0bf0a1cea2
commit a9d8fd9c3c

View File

@@ -1808,10 +1808,23 @@ class TelegramGateway:
if parsed.get("is_info_action"):
if not self._security.is_whitelisted(user_id):
raise UserNotWhitelistedError(f"User {user_id} not in whitelist")
await self._answer_callback(
callback_query_id, action,
text={"detail": "📋 功能開發中", "reanalyze": "🔄 功能開發中", "history": "📊 功能開發中"}.get(action, "⏳ 功能開發中"),
)
incident_id = parsed.get("incident_id", approval_id)
if action == "detail":
# ADR-050 P2: 取得事件詳情,傳送新訊息 (保留原始簽核卡片+按鈕)
# 2026-04-01 Claude Code (ADR-050 P2)
await self._answer_callback(callback_query_id, action, text="📋 詳情傳送中...")
await self._send_incident_detail(incident_id)
elif action == "history":
# ADR-050 P2: 取得頻率統計
# 2026-04-01 Claude Code (ADR-050 P2)
await self._answer_callback(callback_query_id, action, text="📊 歷史統計傳送中...")
await self._send_incident_history(incident_id)
else:
# reanalyze: 開發中
await self._answer_callback(callback_query_id, action, text="🔄 功能開發中")
return {
"action": action,
"approval_id": approval_id,
@@ -2225,6 +2238,113 @@ class TelegramGateway:
"disable_web_page_preview": True,
})
async def _send_incident_detail(self, incident_id: str) -> None:
"""
ADR-050 P2: 傳送事件詳情訊息 (不修改原始簽核卡片)
2026-04-01 Claude Code (ADR-050 P2): detail button handler
"""
# 延遲 import 避免循環依賴 (與 approval_service 同一模式)
from src.repositories.incident_repository import get_incident_repository
try:
repo = get_incident_repository()
incident = await repo.get_by_id(incident_id)
if not incident:
await self.send_notification(f"⚠️ 找不到事件 <code>{html.escape(incident_id)}</code>")
return
dc = incident.decision_chain
confidence_bar = "" * int((dc.confidence if dc else 0) * 10) + "" * (10 - int((dc.confidence if dc else 0) * 10))
lines = [
f"📋 <b>事件詳情</b>",
f"",
f"🔖 <b>ID:</b> <code>{html.escape(incident.incident_id)}</code>",
f"📊 <b>狀態:</b> {incident.status.value}",
f"⚡ <b>嚴重度:</b> {incident.severity.value}",
]
if incident.affected_services:
lines.append(f"🎯 <b>受影響服務:</b> {', '.join(html.escape(s) for s in incident.affected_services[:3])}")
if dc:
lines += [
f"",
f"🤖 <b>AI 分析</b> ({html.escape(dc.model_used)})",
f"💡 {html.escape(dc.hypothesis[:200])}{'...' if len(dc.hypothesis) > 200 else ''}",
f"📈 信心: [{confidence_bar}] {dc.confidence:.0%}",
]
if dc.probable_root_causes:
lines.append(f"🔍 根因: {html.escape(dc.probable_root_causes[0][:100])}")
lines += [
f"",
f"🕐 <b>建立:</b> {incident.created_at.strftime('%m/%d %H:%M')}",
]
if incident.frequency_stats:
fs = incident.frequency_stats
lines.append(f"📉 <b>頻率:</b> 1h={fs.count_1h} 24h={fs.count_24h} 7d={fs.count_7d}")
await self.send_notification("\n".join(lines))
except Exception as e:
logger.warning("send_incident_detail_failed", incident_id=incident_id, error=str(e))
await self.send_notification(f"⚠️ 無法取得事件詳情: {html.escape(str(e)[:100])}")
async def _send_incident_history(self, incident_id: str) -> None:
"""
ADR-050 P2: 傳送事件頻率統計訊息
2026-04-01 Claude Code (ADR-050 P2): history button handler
"""
from src.repositories.incident_repository import get_incident_repository
try:
repo = get_incident_repository()
incident = await repo.get_by_id(incident_id)
if not incident:
await self.send_notification(f"⚠️ 找不到事件 <code>{html.escape(incident_id)}</code>")
return
fs = incident.frequency_stats
if not fs:
await self.send_notification(
f"📊 <b>事件歷史</b>\n\n🔖 <code>{html.escape(incident.incident_id)}</code>\n\n無頻率統計資料"
)
return
lines = [
f"📊 <b>事件歷史統計</b>",
f"",
f"🔖 <code>{html.escape(incident.incident_id)}</code>",
f"🔑 告警鍵: <code>{html.escape(fs.anomaly_key[:60])}</code>",
f"",
f"⏱ <b>發生頻率</b>",
f" 1小時: {fs.count_1h}",
f" 24小時: {fs.count_24h}",
f" 7天: {fs.count_7d}",
f" 30天: {fs.count_30d}",
]
if fs.auto_repair_count > 0:
lines += [
f"",
f"🔧 <b>自動修復</b>",
f" 執行次數: {fs.auto_repair_count}",
]
if fs.last_repair_action:
lines.append(f" 最後動作: {html.escape(fs.last_repair_action[:80])}")
await self.send_notification("\n".join(lines))
except Exception as e:
logger.warning("send_incident_history_failed", incident_id=incident_id, error=str(e))
await self.send_notification(f"⚠️ 無法取得歷史統計: {html.escape(str(e)[:100])}")
async def send_notification(
self,
text: str,