303 lines
13 KiB
Python
303 lines
13 KiB
Python
import json
|
||
import logging
|
||
import os
|
||
import time
|
||
from typing import Any, Dict, Optional
|
||
|
||
import requests
|
||
|
||
from services.ai_orchestrator import AIOrchestrator
|
||
from services.auto_heal_service import auto_heal_service
|
||
from services.logger_manager import SystemLogger
|
||
from services.nemoton_dispatcher_service import NemotronDispatcher
|
||
from services.openclaw_strategist_service import generate_weekly_strategy_report
|
||
from services.telegram_templates import alert, report, success, warning, info as tpl_info
|
||
|
||
sys_log = SystemLogger("EventRouter").get_logger()
|
||
|
||
# ─── 環境 ────────────────────────────────────────────────────
|
||
HERMES_URL = os.getenv("HERMES_URL", "http://192.168.0.111:11434")
|
||
HERMES_MODEL = os.getenv("HERMES_MODEL", "hermes3:latest")
|
||
HERMES_TIMEOUT = int(os.getenv("HERMES_TIMEOUT", "180")) or None
|
||
HERMES_KEEP_ALIVE = os.getenv("HERMES_KEEP_ALIVE", "24h")
|
||
|
||
NEMOTRON_URL = os.getenv("NEMOTRON_URL", "http://192.168.0.111:1144")
|
||
NEMOTRON_TIMEOUT = int(os.getenv("NEMOTRON_TIMEOUT", "60"))
|
||
|
||
TELEGRAM_BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN", "")
|
||
TELEGRAM_CHAT_IDS_RAW = os.getenv("TELEGRAM_CHAT_IDS", "[]")
|
||
try:
|
||
TELEGRAM_CHAT_IDS = json.loads(TELEGRAM_CHAT_IDS_RAW)
|
||
except json.JSONDecodeError:
|
||
TELEGRAM_CHAT_IDS = []
|
||
|
||
SILENCE_DURATION_MIN = int(os.getenv("SILENCE_DURATION_MIN", "30"))
|
||
|
||
# ─── 分類規則（與 watcher_agent.py 保持一致） ────────────────────
|
||
def _classify(event: Dict[str, Any]) -> str:
|
||
sev = event.get("severity", "info")
|
||
has_trace = bool(event.get("trace"))
|
||
event_type = event.get("event_type", "")
|
||
|
||
if sev in ("info", "success"):
|
||
return "L0"
|
||
if sev == "warning":
|
||
return "L1" if has_trace else "L0"
|
||
if sev == "alert":
|
||
if event_type in {"price_threat", "db_connection_error", "crawler_timeout",
|
||
"nim_quota_exhausted", "embedding_failure"}:
|
||
return "L2"
|
||
return "L1"
|
||
return "L0"
|
||
|
||
# ─── 主入口 ───────────────────────────────────────────────────
|
||
def dispatch(event: Dict[str, Any], admin_chat_ids: Optional[list] = None) -> Dict[str, Any]:
|
||
"""
|
||
輸入 event 格式（與 watcher payload 對齊）：
|
||
{
|
||
"source": "watcher",
|
||
"event_type": "sales_anomaly",
|
||
"severity": "alert",
|
||
"title": "...",
|
||
"summary": "...",
|
||
"payload": {...},
|
||
"trace": "...", # 可選
|
||
"suggested_actions": [...]
|
||
}
|
||
回傳：
|
||
{"tier": "L0|L1|L2|L3", "sent": int, "errors": [...], "latency_ms": float}
|
||
"""
|
||
t0 = time.time()
|
||
tier = _classify(event)
|
||
sys_log.info(f"[EventRouter] route {event.get('event_type')} → {tier}")
|
||
|
||
errors = []
|
||
sent = 0
|
||
|
||
try:
|
||
if tier == "L0":
|
||
text = _render_l0(event)
|
||
elif tier == "L1":
|
||
text = _render_l1(event)
|
||
elif tier == "L2":
|
||
text = _render_l2(event)
|
||
else:
|
||
text = _render_l0(event)
|
||
|
||
sent = _send_telegram(text, admin_chat_ids)
|
||
except Exception as e:
|
||
sys_log.error(f"[EventRouter] 渲染失敗，降級 L0: {e}")
|
||
text = _render_l0(event) + "\n\n🟡 <i>AI 分析暫不可用，以原始資料呈現</i>"
|
||
try:
|
||
sent = _send_telegram(text, admin_chat_ids)
|
||
except Exception:
|
||
sent = 0
|
||
errors.append("L0 fallback send failed")
|
||
|
||
latency = (time.time() - t0) * 1000
|
||
sys_log.info(f"[EventRouter] dispatched tier={tier} sent={sent} errors={len(errors)} latency={latency:.0f}ms")
|
||
return {"tier": tier, "sent": sent, "errors": errors, "latency_ms": latency}
|
||
|
||
# ─── L0 直出 ─────────────────────────────────────────────────
|
||
def _render_l0(event: Dict[str, Any]) -> str:
|
||
sev = event.get("severity", "info")
|
||
title = event.get("title", "未命名事件")
|
||
module = event.get("source", "unknown")
|
||
summary = event.get("summary", "")
|
||
details = event.get("payload") if isinstance(event.get("payload"), dict) else None
|
||
|
||
if sev == "success":
|
||
return success(title=title, module=module, stats=summary)
|
||
if sev == "info":
|
||
return info(title=title, module=module, content=summary)
|
||
if sev == "warning":
|
||
return warning(title=title, module=module, summary=summary, details=details)
|
||
return alert(
|
||
title=title, module=module,
|
||
status=event.get("status", "未知"),
|
||
impact=event.get("impact", "未評估"),
|
||
summary=summary,
|
||
actions=event.get("suggested_actions"),
|
||
trace=event.get("trace"),
|
||
)
|
||
|
||
# ─── L1：Hermes 翻譯 ────────────────────────────────────────
|
||
def _render_l1(event: Dict[str, Any]) -> str:
|
||
try:
|
||
parsed = _hermes_observe_parsed(event)
|
||
if parsed and parsed.get("summary"):
|
||
return report.triaged_alert(
|
||
base_event=_event_base(event),
|
||
tier_label="L1 · Hermes",
|
||
ai_summary=parsed.get("summary", ""),
|
||
ai_cause=parsed.get("probable_cause"),
|
||
ai_actions=parsed.get("actions") or [],
|
||
)
|
||
except Exception as e:
|
||
sys_log.warning(f"[EventRouter] L1 Hermes 失敗，降 L0: {e}")
|
||
return _render_l0(event) + "\n\n🟡 <i>AI 分析暫不可用，以原始資料呈現</i>"
|
||
|
||
# ─── L2：NemoTron 規劃 + 審核閘 ─────────────────────────────
|
||
def _render_l2(event: Dict[str, Any]) -> str:
|
||
try:
|
||
ai_result = _nemoton_investigate(event)
|
||
if ai_result:
|
||
parsed = _hermes_observe_parsed(event) # 補齊摘要
|
||
return report.triaged_alert(
|
||
base_event=_event_base(event),
|
||
tier_label="L2 · NemoTron",
|
||
ai_summary=(parsed or {}).get("summary", "") or ai_result.get("summary", ""),
|
||
ai_cause=(parsed or {}).get("probable_cause"),
|
||
ai_actions=(parsed or {}).get("actions", []),
|
||
ai_executed=ai_result.get("actions_taken", []),
|
||
)
|
||
except Exception as e:
|
||
sys_log.warning(f"[EventRouter] L2 NemoTron 失敗，降 L1: {e}")
|
||
return _render_l1(event)
|
||
|
||
# ─── L3：OpenClaw 策略師（週報/分析） ───────────────────────
|
||
def _render_l3(event: Dict[str, Any]) -> str:
|
||
"""週報或 Meta-Analysis 類型交由 OpenClaw"""
|
||
# 範例：週日週報
|
||
if event.get("event_type") == "weekly_meta":
|
||
return generate_weekly_strategy_report()
|
||
return _render_l2(event)
|
||
|
||
# ─── Hermes Observer（Ollama） ────────────────────────────────
|
||
_HERMES_OBSERVE_PROMPT = """你是一個 SRE 助手，將技術錯誤翻譯成人類可理解的摘要。
|
||
|
||
請根據以下事件產出**繁體中文**分析，嚴格以下列 JSON 格式輸出（不要加 markdown 代碼塊、不要加說明）：
|
||
{"summary": "一句話技術根因（中文，<60 字）", "probable_cause": "最可能的原因（中文，<40 字）", "actions": ["建議動作1", "建議動作2"]}
|
||
|
||
限制：
|
||
- summary 翻譯英文錯誤為中文，去除技術 jargon
|
||
- probable_cause 推測根因（基於 stack trace 和事件類型）
|
||
- actions 最多 3 個，具體可執行
|
||
- 若資訊不足，summary 填 "資訊不足"、actions 填 ["請檢查原始 trace"]
|
||
"""
|
||
|
||
def _hermes_observe_parsed(event: Dict[str, Any]) -> Optional[Dict[str, Any]]:
|
||
"""呼叫 Ollama（hermes3）翻譯 stack trace，回傳結構化 dict"""
|
||
try:
|
||
user_prompt = (
|
||
f"事件類型：{event.get('event_type', 'unknown')}\n"
|
||
f"來源模組：{event.get('source', 'unknown')}\n"
|
||
f"標題：{event.get('title', '')}\n"
|
||
f"簡述：{event.get('summary', '')}\n"
|
||
f"技術 trace：\n{(event.get('trace') or '')[-800:]}"
|
||
)
|
||
resp = requests.post(
|
||
f"{HERMES_URL}/api/generate",
|
||
json={
|
||
"model": HERMES_MODEL,
|
||
"system": _HERMES_OBSERVE_PROMPT,
|
||
"prompt": user_prompt,
|
||
"stream": False,
|
||
"keep_alive": HERMES_KEEP_ALIVE,
|
||
"options": {"temperature": 0.1, "num_predict": 300},
|
||
},
|
||
timeout=HERMES_TIMEOUT,
|
||
)
|
||
if not resp.ok:
|
||
sys_log.warning(f"[EventRouter.L1] Hermes HTTP {resp.status_code}")
|
||
return None
|
||
|
||
raw = (resp.json().get("response") or "").strip()
|
||
parsed = json.loads(raw) if raw.startswith("{") else None
|
||
if not parsed or not parsed.get("summary"):
|
||
return None
|
||
|
||
return {
|
||
"summary": str(parsed.get("summary", "")).strip(),
|
||
"probable_cause": str(parsed.get("probable_cause") or "").strip() or None,
|
||
"actions": [str(a).strip() for a in (parsed.get("actions") or []) if a][:5],
|
||
}
|
||
except Exception as e:
|
||
sys_log.warning(f"[EventRouter.L1] Hermes 呼叫失敗，降級：{type(e).__name__}: {str(e)[:120]}")
|
||
return None
|
||
|
||
# ─── NemoTron Investigator（規則式 L2，不呼叫 NIM） ────────────
|
||
_L2_RULES: dict[str, list] = {
|
||
"db_connection_error": [
|
||
("query_km", {"query": "DNS resolve 失敗 momo-postgres"}),
|
||
("retry_task", {"task_name": "<auto>", "backoff_sec": 60}),
|
||
],
|
||
"crawler_timeout": [
|
||
("silence_alert", {"duration_min": SILENCE_DURATION_MIN}),
|
||
("retry_task", {"task_name": "<auto>", "backoff_sec": 300}),
|
||
],
|
||
"nim_quota_exhausted": [
|
||
("silence_alert", {"duration_min": 720}),
|
||
],
|
||
"embedding_failure": [
|
||
("silence_alert", {"duration_min": 10}),
|
||
],
|
||
}
|
||
|
||
def _nemoton_investigate(event: Dict[str, Any]) -> Optional[Dict[str, Any]]:
|
||
"""規則式 L2：依 event_type 查 _L2_RULES，執行安全 actions"""
|
||
event_type = event.get("event_type", "")
|
||
rules = _L2_RULES.get(event_type)
|
||
if not rules:
|
||
return None
|
||
|
||
actions_taken = []
|
||
for action_name, params in rules:
|
||
action_fn = getattr(agent_actions.SAFE_ACTIONS.get(action_name), None)
|
||
if not action_fn:
|
||
continue
|
||
p = dict(params)
|
||
if p.get("task_name") == "<auto>":
|
||
p["task_name"] = event.get("payload", {}).get("task_name", "") or event.get("source", "").split(".")[-1]
|
||
if action_name == "silence_alert" and "event_key" not in p:
|
||
p["event_key"] = f"{event.get('source', '?')}:{event_type}"
|
||
try:
|
||
result = action_fn(**p)
|
||
status = result.get("status", "unknown")
|
||
actions_taken.append(f"{action_name} → {status}")
|
||
except Exception as e:
|
||
actions_taken.append(f"{action_name} → error: {str(e)[:80]}")
|
||
sys_log.error(f"[EventRouter.L2] action {action_name} 例外: {e}")
|
||
|
||
summary = f"依規則 _L2_RULES[{event_type}] 執行 {len(actions_taken)} 個安全動作"
|
||
return {"summary": summary, "actions_taken": actions_taken}
|
||
|
||
# ─── 工具：構建 event 基礎 ─────────────────────────────────────
|
||
def _event_base(event: Dict[str, Any]) -> Dict[str, Any]:
|
||
return {
|
||
"severity": event.get("severity", "warning"),
|
||
"title": event.get("title", "未命名事件"),
|
||
"module": event.get("source", "unknown"),
|
||
"status": event.get("status"),
|
||
"impact": event.get("impact"),
|
||
"summary": event.get("summary", ""),
|
||
"details": event.get("payload"),
|
||
"trace": event.get("trace"),
|
||
}
|
||
|
||
# ─── 工具：Telegram 發送 ───────────────────────────────────────
|
||
def _send_telegram(text: str, admin_chat_ids: Optional[list] = None) -> int:
|
||
if not TELEGRAM_BOT_TOKEN:
|
||
sys_log.warning("[EventRouter] TELEGRAM_BOT_TOKEN 未設定")
|
||
return 0
|
||
|
||
if admin_chat_ids is None:
|
||
admin_chat_ids = TELEGRAM_CHAT_IDS
|
||
if not admin_chat_ids:
|
||
admin_chat_ids = [-1003940688311] # fallback
|
||
|
||
url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage"
|
||
sent = 0
|
||
for cid in admin_chat_ids:
|
||
try:
|
||
r = requests.post(url, json={
|
||
"chat_id": int(cid),
|
||
"text": text,
|
||
"parse_mode": "HTML",
|
||
}, timeout=10)
|
||
if r.ok:
|
||
sent += 1
|
||
except Exception as e:
|
||
sys_log.error(f"[EventRouter] Telegram 發送失敗: {e}")
|
||
return sent
|