feat(telegram): 實作 6 種新訊息模板 (ADR-038)

2026-03-29 ogt: Telegram 訊息模板完整實作

新增訊息類型:
- SentryErrorMessage: Sentry 錯誤通知 (含 Stack Trace)
- ResourceWarnMessage: 資源耗盡警告 (含 CPU/Memory/Disk)
- RepairReportMessage: 自動修復每日報告
- DailySummaryMessage: 每日系統狀態摘要
- DeploySuccessMessage: CD 部署成功通知
- RateLimitMessage: API 限額警告

新增發送方法:
- send_sentry_error()
- send_resource_warning()
- send_repair_report()
- send_daily_summary()
- send_deploy_success()
- send_rate_limit_warning()

新增按鈕:
- Sentry: [🔍 查看詳情] [🔕 靜默 1h]
- Resource: [ 自動擴展] [🔕 靜默 1h]

測試: 14 測試案例全部通過

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
OG T
2026-03-29 21:23:07 +08:00
parent 6416f56748
commit 4707102498
3 changed files with 1070 additions and 23 deletions

View File

@@ -263,6 +263,378 @@ class TelegramMessage:
return message[:900]
# =============================================================================
# 新訊息模板 (2026-03-29 ogt: ADR-038 Telegram 訊息規範)
# =============================================================================
@dataclass
class SentryErrorMessage:
"""
Sentry 錯誤訊息 (SENTRY_ERROR)
2026-03-29 ogt: 新增,用於 Sentry 錯誤通知
按鈕: [🔍 查看詳情] [🔕 靜默 1h]
"""
error_id: str # Sentry Issue ID
error_type: str # TypeError, ValueError, etc.
error_message: str # 錯誤訊息 (max 100)
service_name: str # awoooi-api, awoooi-web, etc.
file_location: str # src/api/v1/incidents.py:123
occurrence_count: int = 1 # 發生次數
affected_users: int = 0 # 影響用戶數
first_seen: str = "" # 首次發生時間
stack_trace: list[str] | None = None # Stack trace (前 3 行)
sentry_url: str = "" # Sentry 連結
def format(self) -> str:
"""格式化為 Telegram HTML"""
safe_error = html.escape(self.error_message[:80])
safe_type = html.escape(self.error_type[:30])
safe_service = html.escape(self.service_name[:25])
safe_file = html.escape(self.file_location[:50])
# Stack trace 區塊
trace_block = ""
if self.stack_trace:
trace_lines = "\n".join(f"{html.escape(line[:50])}" for line in self.stack_trace[:3])
trace_block = f"🔗 Stack Trace (前 3 行):\n{trace_lines}\n"
# Sentry URL
sentry_link = ""
if self.sentry_url:
safe_url = html.escape(self.sentry_url, quote=True)
sentry_link = f"\n🔍 <a href='{safe_url}'>查看 Sentry</a>"
message = (
f"═══════════════════════════\n"
f"🐛 <b>SENTRY ERROR</b> | {safe_service}\n"
f"═══════════════════════════\n"
f"📋 <code>{html.escape(self.error_id)}</code>\n"
f"🎯 錯誤: <code>{safe_type}</code>\n"
f"━━━━━━━━━━━━━━━━━━━\n"
f"💬 {safe_error}\n"
f"━━━━━━━━━━━━━━━━━━━\n"
f"📊 <b>統計</b>\n"
f"├ 發生次數: <code>{self.occurrence_count}</code>\n"
f"├ 影響用戶: <code>{self.affected_users}</code>\n"
f"└ 首次發生: {html.escape(self.first_seen) if self.first_seen else 'N/A'}\n"
f"━━━━━━━━━━━━━━━━━━━\n"
f"📍 位置: <code>{safe_file}</code>\n"
f"{trace_block}"
f"{sentry_link}"
)
return message[:900]
@dataclass
class ResourceWarnMessage:
"""
資源告警訊息 (RESOURCE_WARN)
2026-03-29 ogt: 新增,用於資源耗盡警告
按鈕: [⚡ 自動擴展] [🔕 靜默 1h]
"""
resource_id: str # RES-YYYYMMDD-XXXX
pod_name: str # Pod 名稱
namespace: str = "default" # K8s namespace
cpu_percent: float = 0.0 # CPU 使用率
cpu_limit: str = "" # CPU limit (e.g., 500m)
memory_percent: float = 0.0 # Memory 使用率
memory_limit: str = "" # Memory limit (e.g., 512Mi)
disk_percent: float = 0.0 # Disk 使用率
trend_info: str = "" # 趨勢資訊
suggestion: str = "" # 建議操作
def format(self) -> str:
"""格式化為 Telegram HTML"""
safe_pod = html.escape(self.pod_name[:35])
safe_ns = html.escape(self.namespace[:20])
# 資源狀態 emoji
def get_status_emoji(percent: float) -> str:
if percent >= 90:
return "🔴"
elif percent >= 70:
return "🟡"
return "🟢"
cpu_emoji = get_status_emoji(self.cpu_percent)
mem_emoji = get_status_emoji(self.memory_percent)
disk_emoji = get_status_emoji(self.disk_percent)
# 趨勢和建議
trend_block = ""
if self.trend_info:
trend_block = f"📈 趨勢: {html.escape(self.trend_info[:50])}\n"
suggestion_block = ""
if self.suggestion:
suggestion_block = f"💡 建議: {html.escape(self.suggestion[:50])}\n"
message = (
f"═══════════════════════════\n"
f"⚠️ <b>資源告警</b> | {safe_ns}\n"
f"═══════════════════════════\n"
f"📋 <code>{html.escape(self.resource_id)}</code>\n"
f"🎯 Pod: <code>{safe_pod}</code>\n"
f"━━━━━━━━━━━━━━━━━━━\n"
f"📊 <b>資源使用率</b>\n"
f"├ CPU: {cpu_emoji} <code>{self.cpu_percent:.1f}%</code>"
f"{f' (limit: {self.cpu_limit})' if self.cpu_limit else ''}\n"
f"├ Memory: {mem_emoji} <code>{self.memory_percent:.1f}%</code>"
f"{f' (limit: {self.memory_limit})' if self.memory_limit else ''}\n"
f"└ Disk: {disk_emoji} <code>{self.disk_percent:.1f}%</code>\n"
f"━━━━━━━━━━━━━━━━━━━\n"
f"{trend_block}"
f"{suggestion_block}"
)
return message[:900]
@dataclass
class RepairReportMessage:
"""
自動修復報告訊息 (REPAIR_REPORT)
2026-03-29 ogt: 新增,用於每日自動修復彙總
按鈕: 無
"""
report_date: str # 報告日期 (YYYY-MM-DD)
total_repairs: int = 0 # 總修復次數
success_count: int = 0 # 成功次數
failure_count: int = 0 # 失敗次數
saved_minutes: int = 0 # 節省人工時間 (分鐘)
top_issues: list[tuple[str, int]] | None = None # Top 問題 [(name, count)]
ai_cost_gemini: float = 0.0 # Gemini 成本
ai_cost_nvidia: float = 0.0 # NVIDIA 成本 (免費)
ai_tokens_total: int = 0 # 總 Token 數
def format(self) -> str:
"""格式化為 Telegram HTML"""
# 成功率
success_rate = (self.success_count / self.total_repairs * 100) if self.total_repairs > 0 else 0
# Top 問題區塊
issues_block = ""
if self.top_issues:
issues_lines = "\n".join(
f" {i+1}. {html.escape(name[:30])} ({count} 次)"
for i, (name, count) in enumerate(self.top_issues[:3])
)
issues_block = f"━━━━━━━━━━━━━━━━━━━\n🔝 <b>Top 3 問題</b>:\n{issues_lines}\n"
# AI 成本
total_cost = self.ai_cost_gemini + self.ai_cost_nvidia
message = (
f"═══════════════════════════\n"
f"🔧 <b>自動修復報告</b> | 每日彙總\n"
f"═══════════════════════════\n"
f"📅 {html.escape(self.report_date)}\n"
f"━━━━━━━━━━━━━━━━━━━\n"
f"📊 <b>統計</b>\n"
f"├ 總修復次數: <code>{self.total_repairs}</code>\n"
f"├ 成功: ✅ <code>{self.success_count}</code> ({success_rate:.0f}%)\n"
f"├ 失敗: ❌ <code>{self.failure_count}</code>\n"
f"└ 節省人工: ~<code>{self.saved_minutes}</code> 分鐘\n"
f"{issues_block}"
f"━━━━━━━━━━━━━━━━━━━\n"
f"💰 <b>AI 成本</b>\n"
f"├ Gemini: ${self.ai_cost_gemini:.4f} ({self.ai_tokens_total:,} tokens)\n"
f"├ NVIDIA: ${self.ai_cost_nvidia:.4f} (免費)\n"
f"└ 總計: ${total_cost:.4f}"
)
return message[:900]
@dataclass
class DailySummaryMessage:
"""
每日摘要訊息 (DAILY_SUMMARY)
2026-03-29 ogt: 新增,用於每日系統狀態摘要
按鈕: 無
"""
summary_date: str # 摘要日期 (YYYY-MM-DD)
# 告警統計
alert_total: int = 0
alert_critical: int = 0
alert_medium: int = 0
alert_low: int = 0
# 處理統計
auto_repair_count: int = 0
manual_approval_count: int = 0
ignored_count: int = 0
avg_response_minutes: float = 0.0
# 可用性
api_availability: float = 99.9
web_availability: float = 99.9
worker_availability: float = 99.9
# 成本
ai_cost: float = 0.0
cloud_cost: float = 0.0
budget_remaining: float = 0.0
def format(self) -> str:
"""格式化為 Telegram HTML"""
# 處理百分比
total_handled = self.auto_repair_count + self.manual_approval_count + self.ignored_count
auto_pct = (self.auto_repair_count / total_handled * 100) if total_handled > 0 else 0
manual_pct = (self.manual_approval_count / total_handled * 100) if total_handled > 0 else 0
ignored_pct = (self.ignored_count / total_handled * 100) if total_handled > 0 else 0
message = (
f"═══════════════════════════\n"
f"📊 <b>每日摘要</b> | AWOOOI\n"
f"═══════════════════════════\n"
f"📅 {html.escape(self.summary_date)}\n"
f"━━━━━━━━━━━━━━━━━━━\n"
f"🚨 <b>告警統計</b>\n"
f"├ 總數: <code>{self.alert_total}</code>\n"
f"├ Critical: <code>{self.alert_critical}</code>\n"
f"├ Medium: <code>{self.alert_medium}</code>\n"
f"└ Low: <code>{self.alert_low}</code>\n"
f"━━━━━━━━━━━━━━━━━━━\n"
f"✅ <b>處理統計</b>\n"
f"├ 自動修復: <code>{self.auto_repair_count}</code> ({auto_pct:.0f}%)\n"
f"├ 人工簽核: <code>{self.manual_approval_count}</code> ({manual_pct:.0f}%)\n"
f"├ 忽略/靜默: <code>{self.ignored_count}</code> ({ignored_pct:.0f}%)\n"
f"└ 平均回應: <code>{self.avg_response_minutes:.1f}</code> 分鐘\n"
f"━━━━━━━━━━━━━━━━━━━\n"
f"📈 <b>可用性</b>\n"
f"├ API: <code>{self.api_availability:.2f}%</code>\n"
f"├ Web: <code>{self.web_availability:.2f}%</code>\n"
f"└ Worker: <code>{self.worker_availability:.2f}%</code>\n"
f"━━━━━━━━━━━━━━━━━━━\n"
f"💰 <b>成本</b>\n"
f"├ AI: ${self.ai_cost:.2f}\n"
f"├ 雲端: ${self.cloud_cost:.2f}\n"
f"└ 預算剩餘: ${self.budget_remaining:.2f}"
)
return message[:900]
@dataclass
class DeploySuccessMessage:
"""
部署成功訊息 (DEPLOY_SUCCESS)
2026-03-29 ogt: 新增,用於 CD 部署成功通知
按鈕: 無
"""
commit_sha: str # Git commit SHA (short)
triggered_by: str # 觸發者
environment: str = "Production" # 環境
# 版本資訊
api_version: str = ""
web_version: str = ""
worker_version: str = ""
# 部署時間
duration_seconds: int = 0
# 測試結果
e2e_passed: int = 0
e2e_total: int = 0
health_check_passed: bool = True
# 連結
workflow_url: str = ""
def format(self) -> str:
"""格式化為 Telegram HTML"""
safe_commit = html.escape(self.commit_sha[:8])
safe_user = html.escape(self.triggered_by[:20])
safe_env = html.escape(self.environment[:15])
# 部署時間格式化
minutes = self.duration_seconds // 60
seconds = self.duration_seconds % 60
duration_str = f"{minutes}m {seconds}s" if minutes > 0 else f"{seconds}s"
# 測試結果
e2e_status = "" if self.e2e_passed == self.e2e_total else "⚠️"
health_status = "✅ 全部通過" if self.health_check_passed else "❌ 部分失敗"
# Workflow 連結
workflow_link = ""
if self.workflow_url:
safe_url = html.escape(self.workflow_url, quote=True)
workflow_link = f"\n🔗 <a href='{safe_url}'>查看 Workflow</a>"
message = (
f"✅ <b>部署成功</b> | {safe_env}\n\n"
f"📋 Commit: <code>{safe_commit}</code>\n"
f"👤 觸發者: @{safe_user}\n"
f"━━━━━━━━━━━━━━━━━━━\n"
f"📊 <b>部署詳情</b>\n"
f"├ API: {html.escape(self.api_version) if self.api_version else 'N/A'}\n"
f"├ Web: {html.escape(self.web_version) if self.web_version else 'N/A'}\n"
f"├ Worker: {html.escape(self.worker_version) if self.worker_version else 'N/A'}\n"
f"└ 耗時: {duration_str}\n"
f"━━━━━━━━━━━━━━━━━━━\n"
f"🧪 E2E 測試: {e2e_status} {self.e2e_passed}/{self.e2e_total} PASSED\n"
f"📊 健康檢查: {health_status}"
f"{workflow_link}"
)
return message[:900]
@dataclass
class RateLimitMessage:
"""
API 限額警告訊息 (RATE_LIMIT)
2026-03-29 ogt: 新增,用於 AI API 限額警告
按鈕: 無
"""
provider: str # gemini, openai, etc.
# 用量統計
daily_usage: int = 0
daily_limit: int = 0
token_usage: int = 0
token_limit: int = 0
cost_usd: float = 0.0
# 建議
suggestions: list[str] | None = None
# 重置時間
reset_time: str = ""
def format(self) -> str:
"""格式化為 Telegram HTML"""
safe_provider = html.escape(self.provider.upper()[:15])
# 使用率百分比
usage_pct = (self.daily_usage / self.daily_limit * 100) if self.daily_limit > 0 else 0
token_pct = (self.token_usage / self.token_limit * 100) if self.token_limit > 0 else 0
# 建議區塊
suggestion_block = ""
if self.suggestions:
suggestion_lines = "\n".join(f" - {html.escape(s[:50])}" for s in self.suggestions[:3])
suggestion_block = f"━━━━━━━━━━━━━━━━━━━\n💡 <b>建議</b>:\n{suggestion_lines}\n"
# 重置時間
reset_block = ""
if self.reset_time:
reset_block = f"\n🔄 將於 {html.escape(self.reset_time)} 重置"
message = (
f"⚠️ <b>API 限額警告</b>\n\n"
f"━━━━━━━━━━━━━━━━━━━\n"
f"📊 <b>{safe_provider} API</b>\n"
f"├ 今日用量: <code>{self.daily_usage}/{self.daily_limit}</code> ({usage_pct:.0f}%)\n"
f"├ Token: <code>{self.token_usage:,}/{self.token_limit:,}</code> ({token_pct:.0f}%)\n"
f"└ 成本: ${self.cost_usd:.4f}\n"
f"{suggestion_block}"
f"{reset_block}"
)
return message[:900]
# =============================================================================
# Risk Level Emoji Mapping
# =============================================================================
@@ -585,6 +957,361 @@ class TelegramGateway:
return result
# =========================================================================
# 新訊息發送方法 (2026-03-29 ogt: ADR-038)
# =========================================================================
def _build_sentry_keyboard(self, error_id: str) -> dict:
"""建立 Sentry 錯誤訊息按鈕"""
view_nonce = self._security.generate_callback_nonce(error_id, "view")
silence_nonce = self._security.generate_callback_nonce(error_id, "silence")
return {
"inline_keyboard": [
[
{"text": "🔍 查看詳情", "callback_data": view_nonce},
{"text": "🔕 靜默 1h", "callback_data": silence_nonce},
]
]
}
def _build_resource_keyboard(self, resource_id: str) -> dict:
"""建立資源告警按鈕"""
scale_nonce = self._security.generate_callback_nonce(resource_id, "scale")
silence_nonce = self._security.generate_callback_nonce(resource_id, "silence")
return {
"inline_keyboard": [
[
{"text": "⚡ 自動擴展", "callback_data": scale_nonce},
{"text": "🔕 靜默 1h", "callback_data": silence_nonce},
]
]
}
async def send_sentry_error(
self,
error_id: str,
error_type: str,
error_message: str,
service_name: str,
file_location: str,
occurrence_count: int = 1,
affected_users: int = 0,
first_seen: str = "",
stack_trace: list[str] | None = None,
sentry_url: str = "",
) -> dict:
"""
發送 Sentry 錯誤通知
2026-03-29 ogt: 新增
Args:
error_id: Sentry Issue ID
error_type: 錯誤類型 (TypeError, etc.)
error_message: 錯誤訊息
service_name: 服務名稱
file_location: 檔案位置
occurrence_count: 發生次數
affected_users: 影響用戶數
first_seen: 首次發生時間
stack_trace: Stack trace
sentry_url: Sentry 連結
Returns:
dict: Telegram API 回應
"""
message = SentryErrorMessage(
error_id=error_id,
error_type=error_type,
error_message=error_message,
service_name=service_name,
file_location=file_location,
occurrence_count=occurrence_count,
affected_users=affected_users,
first_seen=first_seen,
stack_trace=stack_trace,
sentry_url=sentry_url,
)
payload = {
"chat_id": self.chat_id,
"text": message.format(),
"parse_mode": "HTML",
"reply_markup": self._build_sentry_keyboard(error_id),
"disable_web_page_preview": True,
}
logger.info("telegram_sentry_error_sending", error_id=error_id, service=service_name)
result = await self._send_request("sendMessage", payload)
logger.info("telegram_sentry_error_sent", error_id=error_id)
return result
async def send_resource_warning(
self,
resource_id: str,
pod_name: str,
namespace: str = "default",
cpu_percent: float = 0.0,
cpu_limit: str = "",
memory_percent: float = 0.0,
memory_limit: str = "",
disk_percent: float = 0.0,
trend_info: str = "",
suggestion: str = "",
) -> dict:
"""
發送資源告警通知
2026-03-29 ogt: 新增
Args:
resource_id: 資源 ID
pod_name: Pod 名稱
namespace: K8s namespace
cpu_percent: CPU 使用率
memory_percent: Memory 使用率
disk_percent: Disk 使用率
trend_info: 趨勢資訊
suggestion: 建議
Returns:
dict: Telegram API 回應
"""
message = ResourceWarnMessage(
resource_id=resource_id,
pod_name=pod_name,
namespace=namespace,
cpu_percent=cpu_percent,
cpu_limit=cpu_limit,
memory_percent=memory_percent,
memory_limit=memory_limit,
disk_percent=disk_percent,
trend_info=trend_info,
suggestion=suggestion,
)
payload = {
"chat_id": self.chat_id,
"text": message.format(),
"parse_mode": "HTML",
"reply_markup": self._build_resource_keyboard(resource_id),
"disable_web_page_preview": True,
}
logger.info("telegram_resource_warning_sending", resource_id=resource_id, pod=pod_name)
result = await self._send_request("sendMessage", payload)
logger.info("telegram_resource_warning_sent", resource_id=resource_id)
return result
async def send_repair_report(
self,
report_date: str,
total_repairs: int = 0,
success_count: int = 0,
failure_count: int = 0,
saved_minutes: int = 0,
top_issues: list[tuple[str, int]] | None = None,
ai_cost_gemini: float = 0.0,
ai_cost_nvidia: float = 0.0,
ai_tokens_total: int = 0,
) -> dict:
"""
發送自動修復報告
2026-03-29 ogt: 新增
Args:
report_date: 報告日期
total_repairs: 總修復次數
success_count: 成功次數
failure_count: 失敗次數
saved_minutes: 節省人工時間
top_issues: Top 問題列表
ai_cost_gemini: Gemini 成本
ai_cost_nvidia: NVIDIA 成本
ai_tokens_total: 總 Token 數
Returns:
dict: Telegram API 回應
"""
message = RepairReportMessage(
report_date=report_date,
total_repairs=total_repairs,
success_count=success_count,
failure_count=failure_count,
saved_minutes=saved_minutes,
top_issues=top_issues,
ai_cost_gemini=ai_cost_gemini,
ai_cost_nvidia=ai_cost_nvidia,
ai_tokens_total=ai_tokens_total,
)
payload = {
"chat_id": self.chat_id,
"text": message.format(),
"parse_mode": "HTML",
"disable_web_page_preview": True,
}
logger.info("telegram_repair_report_sending", date=report_date)
result = await self._send_request("sendMessage", payload)
logger.info("telegram_repair_report_sent", date=report_date)
return result
async def send_daily_summary(
self,
summary_date: str,
alert_total: int = 0,
alert_critical: int = 0,
alert_medium: int = 0,
alert_low: int = 0,
auto_repair_count: int = 0,
manual_approval_count: int = 0,
ignored_count: int = 0,
avg_response_minutes: float = 0.0,
api_availability: float = 99.9,
web_availability: float = 99.9,
worker_availability: float = 99.9,
ai_cost: float = 0.0,
cloud_cost: float = 0.0,
budget_remaining: float = 0.0,
) -> dict:
"""
發送每日摘要
2026-03-29 ogt: 新增
Returns:
dict: Telegram API 回應
"""
message = DailySummaryMessage(
summary_date=summary_date,
alert_total=alert_total,
alert_critical=alert_critical,
alert_medium=alert_medium,
alert_low=alert_low,
auto_repair_count=auto_repair_count,
manual_approval_count=manual_approval_count,
ignored_count=ignored_count,
avg_response_minutes=avg_response_minutes,
api_availability=api_availability,
web_availability=web_availability,
worker_availability=worker_availability,
ai_cost=ai_cost,
cloud_cost=cloud_cost,
budget_remaining=budget_remaining,
)
payload = {
"chat_id": self.chat_id,
"text": message.format(),
"parse_mode": "HTML",
"disable_web_page_preview": True,
}
logger.info("telegram_daily_summary_sending", date=summary_date)
result = await self._send_request("sendMessage", payload)
logger.info("telegram_daily_summary_sent", date=summary_date)
return result
async def send_deploy_success(
self,
commit_sha: str,
triggered_by: str,
environment: str = "Production",
api_version: str = "",
web_version: str = "",
worker_version: str = "",
duration_seconds: int = 0,
e2e_passed: int = 0,
e2e_total: int = 0,
health_check_passed: bool = True,
workflow_url: str = "",
) -> dict:
"""
發送部署成功通知
2026-03-29 ogt: 新增
Returns:
dict: Telegram API 回應
"""
message = DeploySuccessMessage(
commit_sha=commit_sha,
triggered_by=triggered_by,
environment=environment,
api_version=api_version,
web_version=web_version,
worker_version=worker_version,
duration_seconds=duration_seconds,
e2e_passed=e2e_passed,
e2e_total=e2e_total,
health_check_passed=health_check_passed,
workflow_url=workflow_url,
)
payload = {
"chat_id": self.chat_id,
"text": message.format(),
"parse_mode": "HTML",
"disable_web_page_preview": True,
}
logger.info("telegram_deploy_success_sending", commit=commit_sha[:8])
result = await self._send_request("sendMessage", payload)
logger.info("telegram_deploy_success_sent", commit=commit_sha[:8])
return result
async def send_rate_limit_warning(
self,
provider: str,
daily_usage: int = 0,
daily_limit: int = 0,
token_usage: int = 0,
token_limit: int = 0,
cost_usd: float = 0.0,
suggestions: list[str] | None = None,
reset_time: str = "",
) -> dict:
"""
發送 API 限額警告
2026-03-29 ogt: 新增
Returns:
dict: Telegram API 回應
"""
message = RateLimitMessage(
provider=provider,
daily_usage=daily_usage,
daily_limit=daily_limit,
token_usage=token_usage,
token_limit=token_limit,
cost_usd=cost_usd,
suggestions=suggestions,
reset_time=reset_time,
)
payload = {
"chat_id": self.chat_id,
"text": message.format(),
"parse_mode": "HTML",
"disable_web_page_preview": True,
}
logger.info("telegram_rate_limit_warning_sending", provider=provider)
result = await self._send_request("sendMessage", payload)
logger.info("telegram_rate_limit_warning_sent", provider=provider)
return result
async def handle_callback(
self,
callback_query_id: str,

View File

@@ -0,0 +1,312 @@
"""
test_telegram_message_templates.py - Telegram 訊息模板測試
2026-03-29 ogt: 新增,驗證 6 種新訊息模板格式正確
符合 feedback_no_mock_testing.md 規範 - 測試實際格式化輸出
"""
import pytest
from src.services.telegram_gateway import (
DailySummaryMessage,
DeploySuccessMessage,
RateLimitMessage,
RepairReportMessage,
ResourceWarnMessage,
SentryErrorMessage,
TelegramMessage,
)
class TestTelegramMessageFormat:
"""測試現有 TelegramMessage 格式化"""
def test_telegram_message_format_basic(self):
"""測試基本訊息格式化"""
msg = TelegramMessage(
status_emoji="🚨",
risk_level="CRITICAL",
resource_name="test-pod-123",
root_cause="Test root cause",
suggested_action="Restart pod",
estimated_downtime="~30s",
approval_id="INC-20260329-0001",
)
result = msg.format()
assert "🚨" in result
assert "CRITICAL" in result
assert "test-pod-123" in result
assert len(result) <= 900 # SOUL.md 限制
def test_telegram_message_with_token_cost(self):
"""測試含 Token/Cost 的訊息"""
msg = TelegramMessage(
status_emoji="⚠️",
risk_level="MEDIUM",
resource_name="api-pod",
root_cause="High CPU",
suggested_action="Scale up",
estimated_downtime="0s",
approval_id="INC-20260329-0002",
ai_tokens=1500,
ai_cost=0.0015,
)
result = msg.format()
assert "💰 Tokens: 1,500 / $0.0015" in result
class TestSentryErrorMessage:
"""測試 Sentry 錯誤訊息"""
def test_sentry_error_format_basic(self):
"""測試基本 Sentry 錯誤格式"""
msg = SentryErrorMessage(
error_id="SENTRY-abc123",
error_type="TypeError",
error_message="Cannot read property 'x' of undefined",
service_name="awoooi-api",
file_location="src/api/v1/incidents.py:123",
)
result = msg.format()
assert "🐛" in result
assert "SENTRY ERROR" in result
assert "TypeError" in result
assert "awoooi-api" in result
assert len(result) <= 900
def test_sentry_error_with_stack_trace(self):
"""測試含 Stack Trace 的 Sentry 錯誤"""
msg = SentryErrorMessage(
error_id="SENTRY-xyz789",
error_type="ValueError",
error_message="Invalid input",
service_name="awoooi-web",
file_location="src/components/App.tsx:45",
occurrence_count=15,
affected_users=3,
first_seen="10 分鐘前",
stack_trace=[
"incidents.py:123 in get_incident",
"service.py:45 in fetch_data",
"db.py:89 in query",
],
)
result = msg.format()
assert "發生次數: <code>15</code>" in result
assert "影響用戶: <code>3</code>" in result
assert "Stack Trace" in result
class TestResourceWarnMessage:
"""測試資源告警訊息"""
def test_resource_warn_format_basic(self):
"""測試基本資源告警格式"""
msg = ResourceWarnMessage(
resource_id="RES-20260329-0001",
pod_name="awoooi-api-7d4b8c9f5-abc12",
namespace="awoooi-prod",
cpu_percent=92.5,
memory_percent=78.0,
disk_percent=45.0,
)
result = msg.format()
assert "⚠️" in result
assert "資源告警" in result
assert "CPU: 🔴" in result # 92.5% > 90%
assert "Memory: 🟡" in result # 78% >= 70%
assert "Disk: 🟢" in result # 45% < 70%
assert len(result) <= 900
def test_resource_warn_with_limits(self):
"""測試含限制資訊的資源告警"""
msg = ResourceWarnMessage(
resource_id="RES-20260329-0002",
pod_name="test-pod",
cpu_percent=85.0,
cpu_limit="500m",
memory_percent=60.0,
memory_limit="512Mi",
)
result = msg.format()
assert "(limit: 500m)" in result
assert "(limit: 512Mi)" in result
class TestRepairReportMessage:
"""測試自動修復報告"""
def test_repair_report_format_basic(self):
"""測試基本修復報告格式"""
msg = RepairReportMessage(
report_date="2026-03-29",
total_repairs=12,
success_count=10,
failure_count=2,
saved_minutes=45,
)
result = msg.format()
assert "🔧" in result
assert "自動修復報告" in result
assert "總修復次數: <code>12</code>" in result
assert "成功: ✅ <code>10</code> (83%)" in result
assert len(result) <= 900
def test_repair_report_with_top_issues(self):
"""測試含 Top 問題的修復報告"""
msg = RepairReportMessage(
report_date="2026-03-29",
total_repairs=12,
success_count=10,
failure_count=2,
top_issues=[
("Pod CrashLoopBackOff", 5),
("OOM Killed", 4),
("Image Pull Failed", 3),
],
ai_cost_gemini=0.0234,
ai_tokens_total=1823,
)
result = msg.format()
assert "Top 3 問題" in result
assert "Pod CrashLoopBackOff" in result
assert "Gemini: $0.0234" in result
class TestDailySummaryMessage:
"""測試每日摘要"""
def test_daily_summary_format_basic(self):
"""測試基本每日摘要格式"""
msg = DailySummaryMessage(
summary_date="2026-03-29",
alert_total=45,
alert_critical=2,
alert_medium=18,
alert_low=25,
)
result = msg.format()
assert "📊" in result
assert "每日摘要" in result
assert "總數: <code>45</code>" in result
assert "Critical: <code>2</code>" in result
assert len(result) <= 900
def test_daily_summary_with_full_stats(self):
"""測試完整統計的每日摘要"""
msg = DailySummaryMessage(
summary_date="2026-03-29",
alert_total=45,
auto_repair_count=30,
manual_approval_count=10,
ignored_count=5,
api_availability=99.95,
ai_cost=0.15,
budget_remaining=9.85,
)
result = msg.format()
assert "自動修復: <code>30</code>" in result
assert "API: <code>99.95%</code>" in result
assert "預算剩餘: $9.85" in result
class TestDeploySuccessMessage:
"""測試部署成功訊息"""
def test_deploy_success_format_basic(self):
"""測試基本部署成功格式"""
msg = DeploySuccessMessage(
commit_sha="abc1234567",
triggered_by="ogt",
environment="Production",
)
result = msg.format()
assert "" in result
assert "部署成功" in result
assert "abc12345" in result # 前 8 字元
assert "@ogt" in result
assert len(result) <= 900
def test_deploy_success_with_e2e(self):
"""測試含 E2E 結果的部署成功"""
msg = DeploySuccessMessage(
commit_sha="abc1234567",
triggered_by="ogt",
api_version="v1.2.3",
web_version="v1.2.3",
duration_seconds=225, # 3m 45s
e2e_passed=26,
e2e_total=26,
health_check_passed=True,
)
result = msg.format()
assert "v1.2.3" in result
assert "3m 45s" in result
assert "✅ 26/26 PASSED" in result
class TestRateLimitMessage:
"""測試 API 限額警告"""
def test_rate_limit_format_basic(self):
"""測試基本限額警告格式"""
msg = RateLimitMessage(
provider="gemini",
daily_usage=450,
daily_limit=500,
token_usage=85000,
token_limit=100000,
cost_usd=0.08,
)
result = msg.format()
assert "⚠️" in result
assert "API 限額警告" in result
assert "GEMINI API" in result
assert "450/500" in result
assert "(90%)" in result
assert len(result) <= 900
def test_rate_limit_with_suggestions(self):
"""測試含建議的限額警告"""
msg = RateLimitMessage(
provider="openai",
daily_usage=90,
daily_limit=100,
suggestions=[
"考慮切換到 Ollama 優先",
"或增加每日限額",
],
reset_time="明日 00:00",
)
result = msg.format()
assert "建議" in result
assert "切換到 Ollama" in result
assert "明日 00:00" in result

View File

@@ -17,12 +17,12 @@
| 4 | Execution Result | `EXEC_RESULT` | 0 顆 | ✅ 已實作 |
| 5 | Heartbeat | `HEARTBEAT` | 0 顆 | ✅ 已實作 |
| 6 | Silence Alert | `SILENCE` | 0 顆 | ✅ 已實作 |
| 7 | Sentry Error | `SENTRY_ERROR` | 2 顆 | 🆕 待實作 |
| 8 | Resource Exhaustion | `RESOURCE_WARN` | 2 顆 | 🆕 待實作 |
| 9 | Auto-Repair Report | `REPAIR_REPORT` | 0 顆 | 🆕 待實作 |
| 10 | Daily Summary | `DAILY_SUMMARY` | 0 顆 | 🆕 待實作 |
| 11 | Deployment Success | `DEPLOY_SUCCESS` | 0 顆 | 🆕 待實作 |
| 12 | Rate Limit Warning | `RATE_LIMIT` | 0 顆 | 🆕 待實作 |
| 7 | Sentry Error | `SENTRY_ERROR` | 2 顆 | ✅ 已實作 |
| 8 | Resource Exhaustion | `RESOURCE_WARN` | 2 顆 | ✅ 已實作 |
| 9 | Auto-Repair Report | `REPAIR_REPORT` | 0 顆 | ✅ 已實作 |
| 10 | Daily Summary | `DAILY_SUMMARY` | 0 顆 | ✅ 已實作 |
| 11 | Deployment Success | `DEPLOY_SUCCESS` | 0 顆 | ✅ 已實作 |
| 12 | Rate Limit Warning | `RATE_LIMIT` | 0 顆 | ✅ 已實作 |
---
@@ -216,11 +216,12 @@ Telegram 已 **2 小時**沒有收到任何訊息!
---
## 🆕 待實作訊息模板
## ✅ 已實作訊息模板 (2026-03-29 新增)
### 7⃣ Sentry Error (SENTRY_ERROR)
**優先級**: P1
**狀態**: **已實作** (2026-03-29)
**檔案**: `telegram_gateway.py` - `SentryErrorMessage`
**按鈕配置**:
```
@@ -256,7 +257,8 @@ Telegram 已 **2 小時**沒有收到任何訊息!
### 8⃣ Resource Exhaustion (RESOURCE_WARN)
**優先級**: P1
**狀態**: **已實作** (2026-03-29)
**檔案**: `telegram_gateway.py` - `ResourceWarnMessage`
**按鈕配置**:
```
@@ -287,7 +289,8 @@ Telegram 已 **2 小時**沒有收到任何訊息!
### 9⃣ Auto-Repair Report (REPAIR_REPORT)
**優先級**: P2
**狀態**: **已實作** (2026-03-29)
**檔案**: `telegram_gateway.py` - `RepairReportMessage`
**按鈕**: 無
@@ -320,7 +323,8 @@ Telegram 已 **2 小時**沒有收到任何訊息!
### 🔟 Daily Summary (DAILY_SUMMARY)
**優先級**: P2
**狀態**: **已實作** (2026-03-29)
**檔案**: `telegram_gateway.py` - `DailySummaryMessage`
**按鈕**: 無
@@ -359,7 +363,8 @@ Telegram 已 **2 小時**沒有收到任何訊息!
### 1⃣1⃣ Deployment Success (DEPLOY_SUCCESS)
**優先級**: P3
**狀態**: **已實作** (2026-03-29)
**檔案**: `telegram_gateway.py` - `DeploySuccessMessage`
**按鈕**: 無
@@ -388,7 +393,8 @@ Telegram 已 **2 小時**沒有收到任何訊息!
### 1⃣2⃣ Rate Limit Warning (RATE_LIMIT)
**優先級**: P1
**狀態**: **已實作** (2026-03-29)
**檔案**: `telegram_gateway.py` - `RateLimitMessage`
**按鈕**: 無
@@ -454,17 +460,18 @@ Telegram 已 **2 小時**沒有收到任何訊息!
---
## 實作優先級
## 實作狀態
| 優先級 | 訊息類別 | 預估工時 |
|--------|----------|----------|
| 🔴 P1 | SENTRY_ERROR | 2h |
| 🔴 P1 | RESOURCE_WARN | 2h |
| 🔴 P1 | RATE_LIMIT | 1h |
| 🟠 P2 | REPAIR_REPORT | 2h |
| 🟠 P2 | DAILY_SUMMARY | 3h |
| 🟡 P3 | DEPLOY_SUCCESS | 1h |
| **總計** | | **11h** |
| 訊息類別 | 狀態 | 實作日期 |
|----------|------|----------|
| SENTRY_ERROR | ✅ 已實作 | 2026-03-29 |
| RESOURCE_WARN | ✅ 已實作 | 2026-03-29 |
| RATE_LIMIT | ✅ 已實作 | 2026-03-29 |
| REPAIR_REPORT | ✅ 已實作 | 2026-03-29 |
| DAILY_SUMMARY | ✅ 已實作 | 2026-03-29 |
| DEPLOY_SUCCESS | ✅ 已實作 | 2026-03-29 |
**全部 12 種訊息模板已實作完成!**
---
@@ -472,4 +479,5 @@ Telegram 已 **2 小時**沒有收到任何訊息!
| 日期 | 版本 | 內容 |
|------|------|------|
| 2026-03-29 | v1.1 | ✅ 6 種新訊息模板實作完成 (Sentry/Resource/Repair/Daily/Deploy/RateLimit) |
| 2026-03-29 | v1.0 | 初始建立,定義 12 種訊息模板 |