Nexus Agent
fe8b24ca0b
关闭分项告警后不再推送对应恢复消息；仪表盘移除服务器列表并更新设置页说明。 Co-authored-by: Cursor <cursoragent@cursor.com>
318 lines
11 KiB
Python
318 lines
11 KiB
Python
"""Nexus — Telegram Bot Notification Module
|
||
Sends alerts/recovery/system messages via Telegram Bot API.
|
||
All outbound messages use detailed Simplified Chinese.
|
||
"""
|
||
|
||
import html
|
||
import logging
|
||
import re
|
||
import httpx
|
||
from datetime import datetime, timezone
|
||
|
||
from server.config import settings
|
||
from server.utils.notify_toggles import notify_attr_enabled, resource_alert_notify_enabled, resource_recovery_notify_enabled
|
||
|
||
logger = logging.getLogger("nexus.telegram")
|
||
|
||
TELEGRAM_API_BASE = "https://api.telegram.org"
|
||
|
||
_http_client: httpx.AsyncClient | None = None
|
||
|
||
_REDACT_PATTERNS: list[tuple[re.Pattern[str], str]] = [
|
||
(re.compile(r"\b(?:mysql|postgresql|redis)(?:\+\w+)?://\S+", re.I), "<连接地址已隐藏>"),
|
||
(re.compile(r"\bpassword[=:]\S+", re.I), "password=<已隐藏>"),
|
||
(re.compile(r"\bBearer\s+\S+", re.I), "Bearer <已隐藏>"),
|
||
(re.compile(r"\b(?:SECRET|API|ENCRYPTION)_KEY[=:]\S+", re.I), "<密钥已隐藏>"),
|
||
]
|
||
|
||
_METRIC_CN: dict[str, str] = {
|
||
"cpu": "CPU",
|
||
"mem": "内存",
|
||
"disk": "磁盘",
|
||
"time_drift": "时钟偏差",
|
||
}
|
||
|
||
|
||
def _now_cn() -> str:
|
||
return datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC")
|
||
|
||
|
||
def _metric_cn(metric: str) -> str:
|
||
return _METRIC_CN.get(metric.lower(), metric)
|
||
|
||
|
||
def _threshold_for(alert_type: str) -> int:
|
||
mapping = {
|
||
"cpu": settings.CPU_ALERT_THRESHOLD,
|
||
"mem": settings.MEM_ALERT_THRESHOLD,
|
||
"disk": settings.DISK_ALERT_THRESHOLD,
|
||
}
|
||
return mapping.get(alert_type.lower(), 80)
|
||
|
||
|
||
def _system_name() -> str:
|
||
name = (settings.SYSTEM_NAME or "Nexus").strip()
|
||
return html.escape(name)
|
||
|
||
|
||
def sanitize_external_message(text: str, max_len: int = 240) -> str:
|
||
"""Strip connection strings and secrets before external notification channels."""
|
||
if not text:
|
||
return ""
|
||
out = text
|
||
for pattern, replacement in _REDACT_PATTERNS:
|
||
out = pattern.sub(replacement, out)
|
||
if len(out) > max_len:
|
||
out = out[: max_len - 3] + "..."
|
||
return out
|
||
|
||
|
||
async def _get_client() -> httpx.AsyncClient:
|
||
global _http_client
|
||
if _http_client is None or _http_client.is_closed:
|
||
_http_client = httpx.AsyncClient(timeout=10.0)
|
||
return _http_client
|
||
|
||
|
||
async def close_client():
|
||
"""Close the shared httpx client — call during app shutdown"""
|
||
global _http_client
|
||
if _http_client and not _http_client.is_closed:
|
||
await _http_client.aclose()
|
||
_http_client = None
|
||
|
||
|
||
async def send_telegram(message: str) -> bool:
|
||
"""Send a Telegram message via Bot API."""
|
||
if not settings.TELEGRAM_BOT_TOKEN or not settings.TELEGRAM_CHAT_ID:
|
||
return False
|
||
|
||
try:
|
||
client = await _get_client()
|
||
resp = await client.post(
|
||
f"{TELEGRAM_API_BASE}/bot{settings.TELEGRAM_BOT_TOKEN}/sendMessage",
|
||
json={
|
||
"chat_id": settings.TELEGRAM_CHAT_ID,
|
||
"text": message,
|
||
"parse_mode": "HTML",
|
||
},
|
||
)
|
||
if resp.status_code == 200:
|
||
logger.info("Telegram 已发送: %s...", sanitize_external_message(message)[:50])
|
||
return True
|
||
logger.error("Telegram API 错误: %s %s", resp.status_code, resp.text[:100])
|
||
return False
|
||
except Exception as e:
|
||
logger.error("Telegram 发送失败: %s", e)
|
||
return False
|
||
|
||
|
||
async def send_telegram_alert(
|
||
server_name: str,
|
||
alert_type: str,
|
||
value: float,
|
||
server_id: int | None = None,
|
||
) -> bool:
|
||
"""Send server resource alert — respects per-type notify toggle."""
|
||
if not resource_alert_notify_enabled(alert_type):
|
||
logger.debug("告警通知已关闭: %s", alert_type)
|
||
return False
|
||
|
||
metric = _metric_cn(alert_type)
|
||
threshold = _threshold_for(alert_type)
|
||
safe_name = html.escape(server_name)
|
||
sid_line = f"\n子机 ID：<code>{server_id}</code>" if server_id is not None else ""
|
||
|
||
return await send_telegram(
|
||
f"🔴 <b>【资源告警】{metric} 使用率超过阈值</b>\n"
|
||
f"平台：{_system_name()}\n"
|
||
f"子机名称：<b>{safe_name}</b>{sid_line}\n"
|
||
f"当前 {metric} 使用率：<b>{value:.1f}%</b>\n"
|
||
f"告警阈值：{threshold}%\n"
|
||
f"检测来源：Agent 心跳（Layer 2 应用内告警）\n"
|
||
f"具体情况：子机 {metric} 持续高于阈值，可能存在负载过高、进程异常或磁盘/内存不足\n"
|
||
f"建议操作：登录子机排查高占用进程，或在 Nexus 服务器页查看详情\n"
|
||
f"时间：{_now_cn()}"
|
||
)
|
||
|
||
|
||
async def send_telegram_recovery(
|
||
server_name: str,
|
||
metric: str,
|
||
value: float,
|
||
server_id: int | None = None,
|
||
) -> bool:
|
||
"""Send server recovery notification — respects NOTIFY_RECOVERY + per-metric toggle."""
|
||
if not resource_recovery_notify_enabled(metric):
|
||
logger.debug("恢复通知已关闭: %s / NOTIFY_RECOVERY", metric)
|
||
return False
|
||
|
||
metric_cn = _metric_cn(metric)
|
||
threshold = _threshold_for(metric)
|
||
safe_name = html.escape(server_name)
|
||
sid_line = f"\n子机 ID：<code>{server_id}</code>" if server_id is not None else ""
|
||
|
||
return await send_telegram(
|
||
f"🟢 <b>【资源恢复】{metric_cn} 已回落至正常范围</b>\n"
|
||
f"平台：{_system_name()}\n"
|
||
f"子机名称：<b>{safe_name}</b>{sid_line}\n"
|
||
f"当前 {metric_cn} 使用率：{value:.1f}%（阈值 {threshold}%）\n"
|
||
f"检测来源：Agent 心跳（Layer 2 应用内告警）\n"
|
||
f"具体情况：指标已恢复正常，本次告警状态已清除；若再次超阈值将重新推送\n"
|
||
f"时间：{_now_cn()}"
|
||
)
|
||
|
||
|
||
async def send_telegram_system_alert(
|
||
summary: str,
|
||
notify_key: str = "",
|
||
*,
|
||
title: str = "系统告警",
|
||
details: list[str] | None = None,
|
||
action: str = "",
|
||
icon: str = "⚠️",
|
||
) -> bool:
|
||
"""Send system-level alert with structured Chinese detail lines."""
|
||
if notify_key and not notify_attr_enabled(notify_key):
|
||
logger.debug("系统通知已关闭: %s", notify_key)
|
||
return False
|
||
|
||
lines = [
|
||
f"{icon} <b>【{html.escape(title)}】</b>",
|
||
f"平台：{_system_name()}",
|
||
html.escape(sanitize_external_message(summary)),
|
||
]
|
||
if details:
|
||
for item in details:
|
||
lines.append(f"• {html.escape(sanitize_external_message(item))}")
|
||
if action:
|
||
lines.append(f"建议操作：{html.escape(sanitize_external_message(action))}")
|
||
lines.append(f"时间：{_now_cn()}")
|
||
return await send_telegram("\n".join(lines))
|
||
|
||
|
||
async def send_telegram_restart_success() -> bool:
|
||
"""Notify that Python backend has successfully restarted."""
|
||
if not notify_attr_enabled("NOTIFY_RESTART"):
|
||
return False
|
||
return await send_telegram(
|
||
f"🟢 <b>【服务恢复】Nexus 后端进程已自动重启成功</b>\n"
|
||
f"平台：{_system_name()}\n"
|
||
f"检测来源：Layer 2 self_monitor（30 秒周期）\n"
|
||
f"具体情况：Python/FastAPI 进程此前异常退出或僵死，监控已触发自动拉起\n"
|
||
f"当前状态：服务已恢复，API 与 WebSocket 应可正常访问\n"
|
||
f"时间：{_now_cn()}"
|
||
)
|
||
|
||
|
||
async def send_telegram_restart_failed() -> bool:
|
||
"""Notify that Python backend restart failed."""
|
||
if not notify_attr_enabled("NOTIFY_RESTART"):
|
||
return False
|
||
return await send_telegram(
|
||
f"🔴 <b>【服务故障】Nexus 后端自动重启失败</b>\n"
|
||
f"平台：{_system_name()}\n"
|
||
f"检测来源：Layer 2 self_monitor（30 秒周期）\n"
|
||
f"具体情况：监控尝试重启 Python 进程但未成功，平台可能无法提供 API\n"
|
||
f"建议操作：SSH 登录宿主机，检查 docker logs / supervisor 日志并人工介入\n"
|
||
f"时间：{_now_cn()}"
|
||
)
|
||
|
||
|
||
async def send_telegram_sync_complete(
|
||
completed: int,
|
||
failed: int,
|
||
total: int,
|
||
source_path: str,
|
||
operator: str,
|
||
failed_servers: list[str] | None = None,
|
||
duration_seconds: int | None = None,
|
||
) -> bool:
|
||
"""Send push/sync completion notification via Telegram."""
|
||
if not notify_attr_enabled("NOTIFY_SYNC_COMPLETE"):
|
||
return False
|
||
|
||
safe_source = html.escape(sanitize_external_message(source_path, max_len=80))
|
||
safe_operator = html.escape(operator)
|
||
|
||
if failed == 0:
|
||
icon = "🟢"
|
||
title = "文件推送完成"
|
||
situation = "全部目标子机推送成功，无失败项"
|
||
elif failed < total:
|
||
icon = "🟡"
|
||
title = "文件推送部分失败"
|
||
situation = f"共 {total} 台子机，{failed} 台推送失败，请查看失败列表"
|
||
else:
|
||
icon = "🔴"
|
||
title = "文件推送全部失败"
|
||
situation = "所有目标子机均未推送成功，请检查源路径、权限与子机在线状态"
|
||
|
||
lines = [
|
||
f"{icon} <b>【{title}】</b>",
|
||
f"平台：{_system_name()}",
|
||
f"具体情况：{situation}",
|
||
f"源路径：<code>{safe_source}</code>",
|
||
f"操作人：{safe_operator}",
|
||
f"统计：成功 {completed} 台 / 失败 {failed} 台 / 总计 {total} 台",
|
||
]
|
||
if duration_seconds is not None:
|
||
mins, secs = divmod(duration_seconds, 60)
|
||
duration = f"{mins} 分 {secs} 秒" if mins > 0 else f"{secs} 秒"
|
||
lines.append(f"耗时：{duration}")
|
||
if failed_servers:
|
||
names = [html.escape(n) for n in failed_servers[:10]]
|
||
lines.append(f"失败子机：{', '.join(names)}")
|
||
if len(failed_servers) > 10:
|
||
lines.append(f"…及其他 {len(failed_servers) - 10} 台")
|
||
lines.append(f"时间：{_now_cn()}")
|
||
|
||
return await send_telegram("\n".join(lines))
|
||
|
||
|
||
async def send_telegram_script_summary(
|
||
execution_id: int,
|
||
script_label: str,
|
||
operator: str,
|
||
status: str,
|
||
success_count: int,
|
||
failed_count: int,
|
||
total: int,
|
||
board_url: str,
|
||
) -> bool:
|
||
"""Send script execution summary (no per-server failure list) via Telegram."""
|
||
if not notify_attr_enabled("NOTIFY_SCRIPT_COMPLETE"):
|
||
return False
|
||
|
||
safe_label = html.escape(sanitize_external_message(script_label, max_len=80))
|
||
safe_operator = html.escape(operator)
|
||
safe_url = html.escape(board_url)
|
||
|
||
if status == "completed":
|
||
icon = "🟢"
|
||
title = "脚本执行完成"
|
||
elif status == "failed":
|
||
icon = "🔴"
|
||
title = "脚本执行失败"
|
||
else:
|
||
icon = "🟡"
|
||
title = "脚本执行部分失败"
|
||
|
||
if failed_count > 0 and success_count == 0:
|
||
stat_line = f"失败 {failed_count} 台"
|
||
elif failed_count > 0:
|
||
stat_line = f"失败 {failed_count} 台（成功 {success_count} 台）"
|
||
else:
|
||
stat_line = f"成功 {success_count} 台"
|
||
|
||
lines = [
|
||
f"{icon} <b>【{title}】</b>",
|
||
f"平台：{_system_name()}",
|
||
f"脚本：{safe_label} · 执行 #{execution_id}",
|
||
f"统计：{stat_line}",
|
||
f"操作人：{safe_operator}",
|
||
f"详情：<a href=\"{safe_url}\">打开执行看板</a>",
|
||
f"时间：{_now_cn()}",
|
||
]
|
||
return await send_telegram("\n".join(lines))
|