feat: 告警狀態變更在原訊息延續 (方案 B)
All checks were successful
CD Pipeline / build-and-deploy (push) Successful in 12m28s

**telegram_gateway.py**
- 新增 append_incident_update(incident_id, status_line)
  - 從 Redis tg_msg:{id} 取 message_id
  - editMessageReplyMarkup: 移除 Row1(批准/拒絕/靜默),保留 Row2(詳情/重診/歷史)
  - sendMessage reply_to_message_id: 在原訊息下方追加狀態行
  - 找不到 message_id 回傳 False(呼叫方自行 fallback)

**decision_manager.py**
- _push_decision_to_telegram: send_approval_card 後存 tg_msg:{id}=message_id (TTL 24h)
- _push_auto_repair_result: 改用 append_incident_update,找不到 message_id 才 fallback 新訊息

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
OG T
2026-04-09 14:21:33 +08:00
parent 20a2ec1455
commit 4f80ba38c0
2 changed files with 120 additions and 16 deletions

View File

@@ -156,7 +156,7 @@ async def _push_decision_to_telegram(
# 2026-03-27 ogt: 修復 INC-INC-INC- 重複前綴 bug
approval_id = incident.incident_id # 已經是 INC-xxx 格式
await gateway.send_approval_card(
tg_result = await gateway.send_approval_card(
approval_id=approval_id,
risk_level=risk_level,
resource_name=target[:50],
@@ -177,6 +177,11 @@ async def _push_decision_to_telegram(
incident_id=incident.incident_id,
)
# 2026-04-09 Claude Sonnet 4.6: 存 message_id → 後續狀態更新在原訊息延續
tg_message_id = tg_result.get("result", {}).get("message_id") if isinstance(tg_result, dict) else None
if tg_message_id:
await redis.setex(f"tg_msg:{incident.incident_id}", 86400, str(tg_message_id))
# 🔴 發送成功後設置去重 key (TTL 10 分鐘)
await redis.setex(dedup_key, 600, "1")
@@ -203,8 +208,10 @@ async def _push_auto_repair_result(
error: str = "",
) -> None:
"""
自動修復執行後,發 Telegram 結果通知
統帥要求: 修復結果必須對應到同一告警訊息
自動修復執行後,在原始告警訊息追加狀態行。
統帥要求: 所有狀態變更必須在原告警訊息延續,不發新訊息。
- append_incident_update() 取 Redis tg_msg:{id} → reply 原訊息 + 換按鈕
- 找不到 message_id 時 fallback 到 send_notification降級
2026-04-09 Claude Sonnet 4.6 Asia/Taipei
"""
try:
@@ -215,28 +222,36 @@ async def _push_auto_repair_result(
inc_id = incident.incident_id
if success:
text = (
f"✅ <b>[自動修復完成]</b>\n"
f"├ 事件: <code>{inc_id}</code>\n"
f"├ 對象: <code>{target[:50]}</code>\n"
f"└ 執行: <code>{action[:80] if action else '已執行'}</code>"
status_line = (
f"✅ <b>自動修復完成</b>\n"
f" <code>{action[:100] if action else '已執行'}</code>"
)
else:
text = (
f"❌ <b>[自動修復失敗]</b>\n"
f"├ 事件: <code>{inc_id}</code>\n"
f"├ 對象: <code>{target[:50]}</code>\n"
status_line = (
f"❌ <b>自動修復失敗,請人工介入</b>\n"
f"├ 動作: <code>{action[:80] if action else '未知'}</code>\n"
f"└ 錯誤: {error[:100] if error else '未知錯誤'}"
)
await gateway.send_notification(text)
logger.info(
"auto_repair_result_sent",
# 優先: reply 原告警訊息並換掉按鈕
appended = await gateway.append_incident_update(
incident_id=inc_id,
success=success,
status_line=status_line,
keep_info_buttons=True, # 保留詳情/重診/歷史,移除批准/拒絕
)
# Fallback: 找不到原訊息 ID舊告警或 Redis 過期)→ 發新訊息
if not appended:
fallback_text = (
f"{'' if success else ''} <b>[自動修復{'完成' if success else '失敗'}]</b> "
f"<code>{inc_id}</code>\n"
f"對象: <code>{target[:50]}</code>\n"
f"{status_line}"
)
await gateway.send_notification(fallback_text)
logger.info("auto_repair_result_sent", incident_id=inc_id, success=success, appended=appended)
except Exception as e:
logger.warning("auto_repair_result_push_failed", incident_id=incident.incident_id, error=str(e))

View File

@@ -2486,6 +2486,95 @@ class TelegramGateway:
# 文字更新失敗不影響整體流程,按鈕已移除
logger.warning("telegram_update_text_failed", message_id=message_id, error=str(e))
async def append_incident_update(
self,
incident_id: str,
status_line: str,
keep_info_buttons: bool = True,
) -> bool:
"""
在原始告警訊息追加狀態行,並換掉操作按鈕。
用於自動修復完成/失敗後更新原訊息,讓狀態變更在同一則訊息上延續。
流程:
1. 從 Redis 取 tg_msg:{incident_id} 得到 message_id
2. editMessageText: 原文 + 分隔線 + status_line
3. editMessageReplyMarkup: 移除 Row 1 (批准/拒絕/靜默),保留 Row 2 (詳情/重診/歷史)
Args:
incident_id: Incident ID用於查 Redis 的 message_id
status_line: 追加的狀態文字,如「✅ 已自動修復: kubectl rollout restart…」
keep_info_buttons: 是否保留詳情/重診/歷史按鈕(預設 True
Returns:
bool: True = 成功 edit 原訊息False = 找不到 message_idfallback 需另行處理)
2026-04-09 Claude Sonnet 4.6 Asia/Taipei (統帥要求: 狀態變更在原訊息延續)
"""
from src.core.redis_client import get_redis
redis = get_redis()
redis_key = f"tg_msg:{incident_id}"
stored = await redis.get(redis_key)
if not stored:
logger.warning(
"append_incident_update_no_message_id",
incident_id=incident_id,
reason="message_id not in Redis",
)
return False
try:
message_id = int(stored)
except (ValueError, TypeError):
logger.warning("append_incident_update_invalid_message_id", stored=stored)
return False
# Step 1: 取得原始訊息文字Telegram Bot API 不提供讀取原文,只能在 editMessageText 裡重建)
# 策略: 只追加 status_line不讀取原文Telegram edit 要傳完整新文字)
# 所以先用 editMessageReplyMarkup 換按鈕,再 sendMessage 同 chat 以 reply 方式追加狀態
# → 實際上用 reply_to_message_id 讓 Telegram 顯示連結更直觀
# Step 1: 換掉按鈕 (移除 Row 1 批准/拒絕/靜默,保留 Row 2 資訊按鈕)
if keep_info_buttons:
new_keyboard = {"inline_keyboard": [
[
{"text": "📋 詳情", "callback_data": f"detail:{incident_id}"},
{"text": "🔄 重診", "callback_data": f"reanalyze:{incident_id}"},
{"text": "📊 歷史", "callback_data": f"history:{incident_id}"},
],
]}
else:
new_keyboard = {"inline_keyboard": []}
try:
await self._send_request("editMessageReplyMarkup", {
"chat_id": self.chat_id,
"message_id": message_id,
"reply_markup": new_keyboard,
})
except TelegramGatewayError as e:
logger.warning("append_incident_update_edit_buttons_failed", message_id=message_id, error=str(e))
# Step 2: Reply 原訊息追加狀態(保留原文不動,以 reply 方式延續)
try:
await self._send_request("sendMessage", {
"chat_id": self.chat_id,
"text": status_line,
"parse_mode": "HTML",
"reply_to_message_id": message_id,
"disable_web_page_preview": True,
})
except TelegramGatewayError as e:
logger.warning("append_incident_update_reply_failed", message_id=message_id, error=str(e))
logger.info(
"append_incident_update_done",
incident_id=incident_id,
message_id=message_id,
)
return True
async def _send_incident_detail(self, incident_id: str) -> None:
"""
ADR-050 P2: 傳送事件詳情訊息 (不修改原始簽核卡片)