from __future__ import annotations
from fastapi import APIRouter, Request
from fastapi.responses import HTMLResponse
from pydantic import BaseModel, Field
from app.clients.pywechat_client import registry
from app.services.heartbeat_service import heartbeat_service
router = APIRouter(tags=["monitor"])
class HeartbeatRequest(BaseModel):
account_id: str = Field(..., description="渠道账号唯一标识")
account_name: str | None = Field(None)
tenant_code: str = Field("default", description="业务系统租户编码")
risk_level: int = Field(0, ge=0, le=100)
sent_count_today: int = Field(0, ge=0)
@router.post("/v1/heartbeat")
def heartbeat(request: HeartbeatRequest):
return heartbeat_service.heartbeat(request.account_id, request.model_dump(exclude_none=True))
@router.get("/v1/monitor/status")
def monitor_status():
clients = []
for client in registry.list_all():
state = client.get_state()
clients.append({
"account_id": state.account_id,
"account_name": state.account_name,
"login_status": state.login_status.value,
"online_status": state.online_status.value,
"last_heartbeat_at": state.last_heartbeat_at,
"risk_level": state.risk_level,
"sent_count_today": state.sent_count_today,
"error_message": state.error_message,
})
return {"clients": clients, "count": len(clients)}
@router.get("/monitor", response_class=HTMLResponse)
def monitor_page(request: Request):
html = """
pywechat 客户端监控
pywechat 客户端监控
"""
return HTMLResponse(content=html)