67 lines
2.3 KiB
Python
67 lines
2.3 KiB
Python
"""
|
||
Notification routing matrix — ADR-093
|
||
======================================
|
||
單一矩陣決定每種通知類型的發送目標，取代 telegram_gateway.py 內 24 處硬碼 chat_id。
|
||
|
||
設計原則：
|
||
- 正式告警目的地一律 SRE_GROUP_CHAT_ID 優先
|
||
- OPENCLAW_TG_CHAT_ID 只在 SRE_GROUP_CHAT_ID 缺失時作 fail-soft fallback
|
||
- 未知通知類型預設發群組
|
||
|
||
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" # OPENCLAW_TG_CHAT_ID (僅缺群組設定時 fallback)
|
||
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-04-30 Codex: 所有告警類型群組優先，DM 只作缺群組設定 fallback。
|
||
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 相容參數；正式策略永遠群組優先。
|
||
"""
|
||
rule = get_routing_rule(notification_type)
|
||
if rule.destination == Destination.DM and not group_chat_id:
|
||
return [dm_chat_id] if dm_chat_id else []
|
||
return [group_chat_id or dm_chat_id] if (group_chat_id or dm_chat_id) else []
|