fix: 添加 pymysql 依赖 + Edge 兼容路由

- requirements.txt 添加 pymysql、cos-python-sdk-v5
- 新增 /api/ai/alert/edge/report 和 /edge/resolve 路由（无认证）
- 使 Edge 设备可直接上报到 FastAPI 服务

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

app/main.py

@@ -27,7 +27,7 @@ from app.services.ai_analyzer import trigger_async_analysis
 from app.services.notification_service import get_notification_service
 from app.services.device_service import get_device_service
 from app.utils.logger import logger
-from app.routers import yudao_alert_router, yudao_auth_router, yudao_aiot_alarm_router, yudao_aiot_edge_router, yudao_aiot_storage_router
+from app.routers import yudao_alert_router, yudao_auth_router, yudao_aiot_alarm_router, yudao_aiot_edge_router, yudao_aiot_storage_router, edge_compat_router
 from app.yudao_compat import yudao_exception_handler
 import json
 
@@ -83,19 +83,18 @@ app.include_router(yudao_aiot_alarm_router)
 app.include_router(yudao_aiot_edge_router)
 app.include_router(yudao_aiot_storage_router)
 
+# ==================== Edge 兼容路由 ====================
+# Edge 设备使用 /api/ai/alert/edge/* 路径上报（与 WVP 一致），无需认证
+app.include_router(edge_compat_router)
+
 # 注册芋道格式异常处理器
 app.add_exception_handler(HTTPException, yudao_exception_handler)
 
-# ==================== 静态文件（本地截图） ====================
+# ==================== 静态文件 ====================
 _uploads_dir = Path("uploads")
 _uploads_dir.mkdir(parents=True, exist_ok=True)
 app.mount("/uploads", StaticFiles(directory=str(_uploads_dir)), name="uploads")
 
-# Edge 本地截图目录（COS 未配置时的回退访问路径）
-_edge_captures_dir = Path(r"C:\Users\16337\PycharmProjects\ai_edge\data\captures")
-if _edge_captures_dir.exists():
-    app.mount("/captures", StaticFiles(directory=str(_edge_captures_dir)), name="captures")
-
 
 def get_alert_svc():
     return alert_service

app/routers/__init__.py

@@ -9,6 +9,7 @@ from app.routers.yudao_alert import router as yudao_alert_router
 from app.routers.yudao_aiot_alarm import router as yudao_aiot_alarm_router
 from app.routers.yudao_aiot_edge import router as yudao_aiot_edge_router
 from app.routers.yudao_aiot_storage import router as yudao_aiot_storage_router
+from app.routers.edge_compat import router as edge_compat_router
 
 __all__ = [
     "yudao_auth_router",
@@ -16,4 +17,5 @@ __all__ = [
     "yudao_aiot_alarm_router",
     "yudao_aiot_edge_router",
     "yudao_aiot_storage_router",
+    "edge_compat_router",
 ]

app/routers/edge_compat.py

@@ -0,0 +1,71 @@
+"""
+Edge 设备兼容路由
+
+Edge 设备使用 /api/ai/alert/edge/report 和 /api/ai/alert/edge/resolve 路径上报告警，
+该路径与 WVP 端点一致。本模块提供相同路径的路由，无需认证，
+使 Edge 设备可以直接上报到 FastAPI 服务。
+"""
+
+import asyncio
+from datetime import datetime
+from fastapi import APIRouter, Depends
+from typing import Optional
+
+from app.yudao_compat import YudaoResponse
+from app.services.alarm_event_service import get_alarm_event_service, AlarmEventService
+from app.services.notification_service import get_notification_service
+from app.schemas import EdgeAlarmReport, EdgeAlarmResolve
+from app.utils.logger import logger
+
+router = APIRouter(prefix="/api/ai/alert", tags=["Edge-兼容路由"])
+
+
+@router.post("/edge/report")
+async def edge_alarm_report(
+    report: EdgeAlarmReport,
+    service: AlarmEventService = Depends(get_alarm_event_service),
+):
+    """
+    Edge 告警上报（无认证）
+
+    与 /admin-api/aiot/alarm/edge/report 功能相同，
+    但不要求认证，供 Edge 设备直接调用。
+    """
+    alarm = service.create_from_edge_report(report.model_dump())
+
+    if alarm is None:
+        return YudaoResponse.error(500, "告警创建失败")
+
+    # WebSocket 通知
+    try:
+        notification_svc = get_notification_service()
+        notification_svc.notify_sync("new_alert", alarm.to_dict())
+    except Exception:
+        pass
+
+    return YudaoResponse.success({
+        "alarmId": alarm.alarm_id,
+        "created": True,
+    })
+
+
+@router.post("/edge/resolve")
+async def edge_alarm_resolve(
+    resolve: EdgeAlarmResolve,
+    service: AlarmEventService = Depends(get_alarm_event_service),
+):
+    """
+    Edge 告警结束通知（无认证）
+
+    与 /admin-api/aiot/alarm/edge/resolve 功能相同，
+    但不要求认证，供 Edge 设备直接调用。
+    """
+    success = service.resolve_alarm(
+        alarm_id=resolve.alarm_id,
+        duration_ms=resolve.duration_ms,
+        last_frame_time=resolve.last_frame_time,
+        resolve_type=resolve.resolve_type,
+    )
+    if not success:
+        return YudaoResponse.error(404, "告警不存在")
+    return YudaoResponse.success(True)

requirements.txt

@@ -12,3 +12,5 @@ paho-mqtt==2.1.0
 python-dotenv==1.0.1
 websockets==12.0
 redis>=5.0.0
+pymysql>=1.1.0
+cos-python-sdk-v5>=1.9.30