Files
awoooi/apps/api/src/services/notification_matrix.py
Your Name ee2cc2bfc3
Some checks failed
CD Pipeline / tests (push) Failing after 1m23s
CD Pipeline / build-and-deploy (push) Has been skipped
CD Pipeline / post-deploy-checks (push) Has been skipped
Code Review / ai-code-review (push) Successful in 15s
fix(alerts): 收斂 Telegram 告警到 SRE 戰情室
2026-06-12 11:06:16 +08:00

65 lines
2.1 KiB
Python

"""
Notification routing matrix — ADR-093
======================================
單一矩陣決定每種通知類型的發送目標,取代 telegram_gateway.py 內 24 處硬碼 chat_id。
設計原則:
- 正式告警目的地一律為 SRE_GROUP_CHAT_ID
- SRE_GROUP_CHAT_ID 缺失時必須顯示配置缺口,不得旁路到個人或其他群組
- 未知通知類型預設發群組
2026-04-25 ogt + Claude Sonnet 4.6
"""
from __future__ import annotations
from dataclasses import dataclass
from enum import Enum
class Destination(str, Enum):
DM = "dm" # legacy alias: 2026-06-12 起不再旁路至 DM
GROUP = "group" # SRE_GROUP_CHAT_ID
BOTH = "both" # legacy alias: 2026-04-30 起視為 group-first
@dataclass(frozen=True)
class RoutingRule:
destination: Destination
strip_buttons_for_group: bool = False # BOTH 時群組版是否去除 Callback Button
# ADR-093 D1-D4 路由矩陣
# 2026-06-12 Codex: 所有正式告警只送 AwoooI SRE 戰情室;缺群組設定時回空清單。
NOTIFICATION_ROUTING: dict[str, RoutingRule] = {
"TYPE-1": RoutingRule(Destination.GROUP),
"TYPE-2": RoutingRule(Destination.GROUP),
"TYPE-3": RoutingRule(Destination.GROUP),
"TYPE-4": RoutingRule(Destination.GROUP),
"TYPE-4D": RoutingRule(Destination.GROUP),
"TYPE-5S": RoutingRule(Destination.GROUP),
"TYPE-6B": RoutingRule(Destination.GROUP),
"TYPE-7E": RoutingRule(Destination.GROUP),
"TYPE-8M": RoutingRule(Destination.GROUP),
}
_DEFAULT_RULE = RoutingRule(Destination.GROUP)
def get_routing_rule(notification_type: str) -> RoutingRule:
"""根據通知類型回傳路由規則。未知類型預設發群組。"""
return NOTIFICATION_ROUTING.get(notification_type, _DEFAULT_RULE)
def resolve_chat_ids(
notification_type: str,
dm_chat_id: str,
group_chat_id: str,
*,
tg_group_cutover: bool = False,
) -> list[str]:
"""
回傳此通知應發送的 chat_id 清單。
tg_group_cutover 僅保留為舊 caller 相容參數;正式策略永遠群組優先。
"""
_ = get_routing_rule(notification_type)
return [group_chat_id] if group_chat_id else []