fix(notifications): 新增後端 /api/v1/notifications/channels 路由
Some checks failed
CD Pipeline / build-and-deploy (push) Failing after 2m4s

前端 /notifications 頁面呼叫此 endpoint 但後端不存在 (404)
新增 notifications.py:回傳 4 個真實頻道狀態
- Telegram OpenClaw Bot (BOT_TOKEN 設定檢查)
- Telegram Nemotron Bot (BOT_TOKEN 設定檢查)
- SSE Web Stream (永遠 active)
- Redis Stream awoooi:signals (ping 檢查)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
OG T
2026-04-10 16:17:37 +08:00
parent 9a3002ed76
commit 527ce9faaf
2 changed files with 89 additions and 0 deletions

View File

@@ -0,0 +1,85 @@
"""
通知頻道狀態 API
================
GET /api/v1/notifications/channels — 回傳所有通知頻道的真實狀態
頻道:
1. Telegram (OpenClaw bot) — 檢查 BOT_TOKEN 是否設定
2. Telegram (AWOOOI bot) — 檢查 AWOOOI_TG_BOT_TOKEN 是否設定
3. SSE (Server-Sent Events) — 永遠 active (HTTP endpoint 存在)
4. Redis Stream — 檢查 Redis 連線狀態
建立時間: 2026-04-10 (台北時區)
建立者: Claude Code (Sprint 5R B5 修復)
"""
from __future__ import annotations
from fastapi import APIRouter
from src.core.config import settings
router = APIRouter()
@router.get("/notifications/channels")
async def list_notification_channels() -> list[dict]:
"""
回傳所有通知頻道的真實設定狀態。
不做網路探測 (避免延遲),只檢查 config 是否完整。
"""
channels: list[dict] = []
# 1. OpenClaw Telegram Bot (告警 + 審核按鈕)
openclaw_ok = bool(
getattr(settings, "OPENCLAW_TG_BOT_TOKEN", None)
or getattr(settings, "OPENCLAW_BOT_TOKEN", None)
)
channels.append({
"name": "Telegram (OpenClaw Bot)",
"type": "telegram",
"status": "active" if openclaw_ok else "error",
"description": "告警通知 + HITL 審核按鈕",
"features": ["alerts", "approvals", "auto_repair"],
})
# 2. Nemotron Telegram Bot (AI 回覆)
nemotron_ok = bool(getattr(settings, "NEMOTRON_BOT_TOKEN", None))
channels.append({
"name": "Telegram (Nemotron Bot)",
"type": "telegram",
"status": "active" if nemotron_ok else "error",
"description": "AI 分析結果回覆",
"features": ["ai_responses", "rag_query"],
})
# 3. SSE (Server-Sent Events) — 前端實時推播
channels.append({
"name": "SSE (Web Stream)",
"type": "sse",
"status": "active",
"description": "前端儀表板實時數據推播",
"features": ["dashboard", "approvals", "incidents"],
"endpoint": "/api/v1/dashboard/stream",
})
# 4. Redis Stream — 感測器信號通道
try:
import redis.asyncio as aioredis
r = aioredis.from_url(settings.REDIS_URL, socket_connect_timeout=1)
await r.ping()
await r.aclose()
redis_status = "active"
except Exception:
redis_status = "error"
channels.append({
"name": "Redis Stream (awoooi:signals)",
"type": "stream",
"status": redis_status,
"description": "Sensor Agent 信號通道",
"features": ["sensor_signals", "dedup"],
"stream_key": "awoooi:signals",
})
return channels

View File

@@ -62,6 +62,7 @@ from src.api.v1 import (
from src.api.v1 import drift as drift_v1 # Phase 25 P2: Config Drift Detection
from src.api.v1 import rag as rag_v1 # Phase 33 ADR-067: RAG 知識庫
from src.api.v1 import monitoring as monitoring_v1 # 2026-04-03: 監控工具狀態
from src.api.v1 import notifications as notifications_v1 # 2026-04-10: 通知頻道狀態
from src.api.v1 import stats as stats_v1 # Phase 6.5: Statistics Analytics
from src.api.v1 import telegram as telegram_v1 # Phase 5.4: Telegram Gateway
from src.api.v1 import terminal as terminal_v1 # Phase 19.1: Omni-Terminal SSE
@@ -506,6 +507,9 @@ app.include_router(
app.include_router(
signoz_webhook_v1.router, prefix="/api/v1", tags=["SignOz Webhook"]
) # Phase 21: SignOz → Telegram (ADR-037)
app.include_router(
notifications_v1.router, prefix="/api/v1", tags=["Notifications"]
) # 2026-04-10: 通知頻道狀態
app.include_router(
terminal_v1.router, prefix="/api/v1", tags=["Omni-Terminal"]
) # Phase 19.1: Omni-Terminal SSE