180 lines
6.8 KiB
Python
180 lines
6.8 KiB
Python
# ///
|
|
# monitor_service.py
|
|
# 描述：节点状态监控服务，定时检测API和微信状态
|
|
# 作者：AI Generated
|
|
# 创建日期：2026-04-05
|
|
# ///
|
|
|
|
import asyncio
|
|
from datetime import datetime
|
|
from typing import Dict, Any, Optional
|
|
import httpx
|
|
|
|
from config import get_settings, WebhookConfig
|
|
from services.node_manager import node_manager
|
|
from services.log_service import log_service
|
|
from database_service import get_db_service
|
|
|
|
|
|
class MonitorService:
|
|
def __init__(self):
|
|
self._running = False
|
|
self._task: Optional[asyncio.Task] = None
|
|
self._last_status: Dict[str, Dict[str, str]] = {}
|
|
|
|
async def start(self):
|
|
if self._running:
|
|
return
|
|
self._running = True
|
|
self._task = asyncio.create_task(self._monitor_loop())
|
|
|
|
async def stop(self):
|
|
self._running = False
|
|
if self._task:
|
|
self._task.cancel()
|
|
try:
|
|
await self._task
|
|
except asyncio.CancelledError:
|
|
pass
|
|
|
|
async def _monitor_loop(self):
|
|
settings = get_settings()
|
|
log_service.info("监控服务启动", "Monitor")
|
|
while self._running:
|
|
try:
|
|
await self._check_all_nodes()
|
|
except Exception as e:
|
|
log_service.error(f"监控循环错误: {e}", "Monitor")
|
|
await asyncio.sleep(settings.monitor.check_interval)
|
|
|
|
async def _check_all_nodes(self):
|
|
settings = get_settings()
|
|
if not settings.monitor.enabled:
|
|
return
|
|
|
|
for node in await node_manager.get_all_nodes():
|
|
await self._check_node(node.node_id)
|
|
|
|
async def _check_node(self, node_id: str):
|
|
node = await node_manager.get_node(node_id)
|
|
if not node:
|
|
return
|
|
|
|
api_status_changed = False
|
|
wechat_status_changed = False
|
|
old_api_status = self._last_status.get(node_id, {}).get("api_status")
|
|
old_wechat_status = self._last_status.get(node_id, {}).get("wechat_status")
|
|
|
|
api_status = await node_manager.check_node_status(node_id)
|
|
if old_api_status is not None and old_api_status != api_status:
|
|
api_status_changed = True
|
|
self._last_status.setdefault(node_id, {})["api_status"] = api_status
|
|
|
|
wechat_status = node.wechat_status
|
|
if old_wechat_status is not None and old_wechat_status != wechat_status:
|
|
wechat_status_changed = True
|
|
self._last_status.setdefault(node_id, {})["wechat_status"] = wechat_status
|
|
|
|
if api_status_changed:
|
|
event = "node_online" if self._last_status[node_id]["api_status"] == "active" else "node_offline"
|
|
msg = f"节点 {node.name} ({node_id}) API状态变为 {self._last_status[node_id]['api_status']}"
|
|
log_service.info(msg, "Monitor")
|
|
if event == "node_offline":
|
|
log_service.error(msg, "Monitor")
|
|
await self._send_webhook(event, {
|
|
"node_id": node_id,
|
|
"node_name": node.name,
|
|
"api_status": self._last_status[node_id]["api_status"],
|
|
"message": msg
|
|
})
|
|
|
|
if wechat_status_changed:
|
|
event = "wechat_online" if wechat_status == "online" else "wechat_offline"
|
|
msg = f"节点 {node.name} ({node_id}) 微信状态变为 {wechat_status}"
|
|
log_service.info(msg, "Monitor")
|
|
if event == "wechat_offline":
|
|
log_service.warning(msg, "Monitor")
|
|
await self._send_webhook(event, {
|
|
"node_id": node_id,
|
|
"node_name": node.name,
|
|
"wechat_status": wechat_status,
|
|
"message": msg
|
|
})
|
|
|
|
async def _send_webhook(self, event: str, data: Dict[str, Any]):
|
|
settings = get_settings()
|
|
webhook_config: WebhookConfig = settings.webhook
|
|
|
|
db_service = get_db_service()
|
|
webhooks = []
|
|
if db_service:
|
|
webhooks = db_service.get_enabled_webhooks()
|
|
|
|
if not webhooks:
|
|
if not webhook_config.enabled or not webhook_config.urls:
|
|
return
|
|
webhooks = [{"url": url, "format": webhook_config.format} for url in webhook_config.urls]
|
|
|
|
event_names = {
|
|
"api_error": "API错误",
|
|
"wechat_offline": "微信掉线",
|
|
"wechat_online": "微信上线",
|
|
"node_offline": "节点离线",
|
|
"node_online": "节点在线"
|
|
}
|
|
|
|
event_name = event_names.get(event, event)
|
|
node_name = data.get('node_name', 'N/A')
|
|
status_info = data.get('wechat_status') or data.get('api_status') or 'N/A'
|
|
timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
|
|
|
for webhook in webhooks:
|
|
url = webhook.url if hasattr(webhook, 'url') else webhook.get('url')
|
|
format_type = webhook.format if hasattr(webhook, 'format') else webhook.get('format', 'bark')
|
|
event_types = webhook.event_types if hasattr(webhook, 'event_types') else []
|
|
|
|
if event_types and event not in event_types:
|
|
continue
|
|
|
|
if format_type == "bark":
|
|
title = f"WXAuto告警 - {event_name}"
|
|
content = f"{node_name}\n状态: {status_info}\n时间: {timestamp}"
|
|
payload = {
|
|
"title": title,
|
|
"body": content,
|
|
"icon": "https://is3-ssl.mzstatic.com/image/thumb/Purple122/v4/8e/1a/b5/8e1ab5a2-09c7-2c06-ae03-2056b0348d2c/AppIcon-Production-20x20@2x.png/0x0ss.png"
|
|
}
|
|
else:
|
|
content = f"""【WXAuto Center 告警】
|
|
事件类型: {event_name}
|
|
节点名称: {node_name}
|
|
状态信息: {status_info}
|
|
时间: {timestamp}"""
|
|
payload = {
|
|
"msgtype": "text",
|
|
"text": {
|
|
"content": content
|
|
}
|
|
}
|
|
|
|
for attempt in range(webhook_config.retry_times):
|
|
try:
|
|
async with httpx.AsyncClient(timeout=webhook_config.timeout) as client:
|
|
response = await client.post(url, json=payload)
|
|
if response.status_code == 200:
|
|
log_service.success(f"Webhook通知发送成功: {event} -> {url}", "Monitor")
|
|
break
|
|
except Exception as e:
|
|
log_service.warning(f"Webhook发送失败 (尝试 {attempt + 1}): {e}", "Monitor")
|
|
else:
|
|
log_service.error(f"Webhook通知发送失败 (已重试{webhook_config.retry_times}次): {event} -> {url}", "Monitor")
|
|
|
|
def get_last_status(self, node_id: str) -> Dict[str, str]:
|
|
return self._last_status.get(node_id, {})
|
|
|
|
def get_all_status(self) -> Dict[str, Dict[str, str]]:
|
|
return self._last_status.copy()
|
|
|
|
|
|
monitor_service = MonitorService()
|