feat(yudao): 添加芋道兼容层和基础路由
- 新增 yudao_compat.py：芋道标准响应格式、权限校验 - 新增 yudao_auth.py：登录认证、权限信息、租户等系统接口 - 新增 yudao_alert.py：告警管理和摄像头汇总的芋道兼容路由 - 新增 routers/__init__.py：统一导出路由模块 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
17
app/routers/__init__.py
Normal file
17
app/routers/__init__.py
Normal file
@@ -0,0 +1,17 @@
|
||||
"""
|
||||
路由模块
|
||||
|
||||
芋道兼容的 API 路由
|
||||
"""
|
||||
|
||||
from app.routers.yudao_auth import router as yudao_auth_router
|
||||
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
|
||||
|
||||
__all__ = [
|
||||
"yudao_auth_router",
|
||||
"yudao_alert_router",
|
||||
"yudao_aiot_alarm_router",
|
||||
"yudao_aiot_edge_router",
|
||||
]
|
||||
264
app/routers/yudao_alert.py
Normal file
264
app/routers/yudao_alert.py
Normal file
@@ -0,0 +1,264 @@
|
||||
"""
|
||||
AI 告警路由 - 芋道规范
|
||||
|
||||
提供与芋道主平台一致的接口规范。
|
||||
|
||||
API 路径规范：
|
||||
- /admin-api/ai-alert/alert/page - 分页查询
|
||||
- /admin-api/ai-alert/alert/get - 获取详情
|
||||
- /admin-api/ai-alert/alert/handle - 处理告警
|
||||
- /admin-api/ai-alert/alert/delete - 删除告警
|
||||
- /admin-api/ai-alert/alert/statistics - 获取统计
|
||||
- /admin-api/ai-alert/camera-summary/page - 摄像头汇总
|
||||
|
||||
分页参数规范：
|
||||
- pageNo: 页码（从1开始）
|
||||
- pageSize: 每页大小
|
||||
|
||||
响应格式规范：
|
||||
{
|
||||
"code": 0,
|
||||
"data": { "list": [...], "total": 100 },
|
||||
"msg": "success"
|
||||
}
|
||||
"""
|
||||
|
||||
from fastapi import APIRouter, Query, Depends, HTTPException
|
||||
from typing import Optional
|
||||
from datetime import datetime
|
||||
|
||||
from app.yudao_compat import YudaoResponse, get_current_user
|
||||
from app.services.alert_service import get_alert_service, AlertService
|
||||
from app.schemas import AlertHandleRequest
|
||||
|
||||
router = APIRouter(prefix="/admin-api/ai-alert", tags=["AI告警"])
|
||||
|
||||
|
||||
# ==================== 告警管理 ====================
|
||||
|
||||
@router.get("/alert/page")
|
||||
async def get_alert_page(
|
||||
pageNo: int = Query(1, ge=1, description="页码"),
|
||||
pageSize: int = Query(20, ge=1, le=100, description="每页大小"),
|
||||
cameraId: Optional[str] = Query(None, description="摄像头ID"),
|
||||
deviceId: Optional[str] = Query(None, description="设备ID"),
|
||||
alertType: Optional[str] = Query(None, description="告警类型"),
|
||||
status: Optional[str] = Query(None, description="状态: pending/confirmed/ignored/resolved/dispatched"),
|
||||
level: Optional[str] = Query(None, description="级别: low/medium/high/critical"),
|
||||
startTime: Optional[datetime] = Query(None, description="开始时间"),
|
||||
endTime: Optional[datetime] = Query(None, description="结束时间"),
|
||||
service: AlertService = Depends(get_alert_service),
|
||||
current_user: dict = Depends(get_current_user)
|
||||
):
|
||||
"""
|
||||
分页查询告警列表
|
||||
|
||||
权限标识：ai-alert:alert:query（测试阶段放行）
|
||||
"""
|
||||
alerts, total = service.get_alerts(
|
||||
camera_id=cameraId,
|
||||
device_id=deviceId,
|
||||
alert_type=alertType,
|
||||
status=status,
|
||||
level=level,
|
||||
start_time=startTime,
|
||||
end_time=endTime,
|
||||
page=pageNo,
|
||||
page_size=pageSize,
|
||||
)
|
||||
|
||||
# 转换为前端需要的格式
|
||||
alert_list = []
|
||||
for alert in alerts:
|
||||
alert_dict = alert.to_dict()
|
||||
# 字段名映射：snake_case -> camelCase
|
||||
alert_list.append({
|
||||
"id": alert_dict.get("id"),
|
||||
"alertNo": alert_dict.get("alert_no"),
|
||||
"cameraId": alert_dict.get("camera_id"),
|
||||
"cameraName": alert_dict.get("camera_id"), # TODO: 从设备服务获取
|
||||
"roiId": alert_dict.get("roi_id"),
|
||||
"bindId": alert_dict.get("bind_id"),
|
||||
"deviceId": alert_dict.get("device_id"),
|
||||
"alertType": alert_dict.get("alert_type"),
|
||||
"alertTypeName": _get_alert_type_name(alert_dict.get("alert_type")),
|
||||
"algorithm": alert_dict.get("algorithm"),
|
||||
"confidence": alert_dict.get("confidence"),
|
||||
"durationMinutes": alert_dict.get("duration_minutes"),
|
||||
"triggerTime": alert_dict.get("trigger_time"),
|
||||
"message": alert_dict.get("message"),
|
||||
"bbox": alert_dict.get("bbox"),
|
||||
"snapshotUrl": alert_dict.get("snapshot_url"),
|
||||
"ossUrl": alert_dict.get("oss_url"),
|
||||
"status": alert_dict.get("status"),
|
||||
"level": alert_dict.get("level"),
|
||||
"handleRemark": alert_dict.get("handle_remark"),
|
||||
"handledBy": alert_dict.get("handled_by"),
|
||||
"handledAt": alert_dict.get("handled_at"),
|
||||
"workOrderId": alert_dict.get("work_order_id"),
|
||||
"aiAnalysis": alert_dict.get("ai_analysis"),
|
||||
"createdAt": alert_dict.get("created_at"),
|
||||
"updatedAt": alert_dict.get("updated_at"),
|
||||
})
|
||||
|
||||
return YudaoResponse.page(
|
||||
list_data=alert_list,
|
||||
total=total,
|
||||
page_no=pageNo,
|
||||
page_size=pageSize
|
||||
)
|
||||
|
||||
|
||||
@router.get("/alert/get")
|
||||
async def get_alert(
|
||||
id: int = Query(..., description="告警ID"),
|
||||
service: AlertService = Depends(get_alert_service),
|
||||
current_user: dict = Depends(get_current_user)
|
||||
):
|
||||
"""
|
||||
获取告警详情
|
||||
|
||||
权限标识：ai-alert:alert:query
|
||||
"""
|
||||
alert = service.get_alert(id)
|
||||
if not alert:
|
||||
raise HTTPException(status_code=404, detail="告警不存在")
|
||||
|
||||
alert_dict = alert.to_dict()
|
||||
return YudaoResponse.success({
|
||||
"id": alert_dict.get("id"),
|
||||
"alertNo": alert_dict.get("alert_no"),
|
||||
"cameraId": alert_dict.get("camera_id"),
|
||||
"cameraName": alert_dict.get("camera_id"),
|
||||
"roiId": alert_dict.get("roi_id"),
|
||||
"bindId": alert_dict.get("bind_id"),
|
||||
"deviceId": alert_dict.get("device_id"),
|
||||
"alertType": alert_dict.get("alert_type"),
|
||||
"alertTypeName": _get_alert_type_name(alert_dict.get("alert_type")),
|
||||
"algorithm": alert_dict.get("algorithm"),
|
||||
"confidence": alert_dict.get("confidence"),
|
||||
"durationMinutes": alert_dict.get("duration_minutes"),
|
||||
"triggerTime": alert_dict.get("trigger_time"),
|
||||
"message": alert_dict.get("message"),
|
||||
"bbox": alert_dict.get("bbox"),
|
||||
"snapshotUrl": alert_dict.get("snapshot_url"),
|
||||
"ossUrl": alert_dict.get("oss_url"),
|
||||
"status": alert_dict.get("status"),
|
||||
"level": alert_dict.get("level"),
|
||||
"handleRemark": alert_dict.get("handle_remark"),
|
||||
"handledBy": alert_dict.get("handled_by"),
|
||||
"handledAt": alert_dict.get("handled_at"),
|
||||
"workOrderId": alert_dict.get("work_order_id"),
|
||||
"aiAnalysis": alert_dict.get("ai_analysis"),
|
||||
"createdAt": alert_dict.get("created_at"),
|
||||
"updatedAt": alert_dict.get("updated_at"),
|
||||
})
|
||||
|
||||
|
||||
@router.put("/alert/handle")
|
||||
async def handle_alert(
|
||||
id: int = Query(..., description="告警ID"),
|
||||
status: str = Query(..., description="处理状态: confirmed/ignored/resolved"),
|
||||
remark: Optional[str] = Query(None, description="处理备注"),
|
||||
service: AlertService = Depends(get_alert_service),
|
||||
current_user: dict = Depends(get_current_user)
|
||||
):
|
||||
"""
|
||||
处理告警
|
||||
|
||||
权限标识：ai-alert:alert:handle
|
||||
"""
|
||||
handle_data = AlertHandleRequest(status=status, remark=remark)
|
||||
handled_by = current_user.get("username", "admin")
|
||||
|
||||
alert = service.handle_alert(id, handle_data, handled_by)
|
||||
if not alert:
|
||||
raise HTTPException(status_code=404, detail="告警不存在")
|
||||
|
||||
return YudaoResponse.success(True)
|
||||
|
||||
|
||||
@router.delete("/alert/delete")
|
||||
async def delete_alert(
|
||||
id: int = Query(..., description="告警ID"),
|
||||
service: AlertService = Depends(get_alert_service),
|
||||
current_user: dict = Depends(get_current_user)
|
||||
):
|
||||
"""
|
||||
删除告警
|
||||
|
||||
权限标识：ai-alert:alert:delete
|
||||
"""
|
||||
success = service.delete_alert(id)
|
||||
if not success:
|
||||
raise HTTPException(status_code=404, detail="告警不存在")
|
||||
|
||||
return YudaoResponse.success(True)
|
||||
|
||||
|
||||
@router.get("/alert/statistics")
|
||||
async def get_statistics(
|
||||
service: AlertService = Depends(get_alert_service),
|
||||
current_user: dict = Depends(get_current_user)
|
||||
):
|
||||
"""
|
||||
获取告警统计
|
||||
|
||||
权限标识：ai-alert:alert:query
|
||||
"""
|
||||
stats = service.get_statistics()
|
||||
|
||||
# 转换为驼峰命名
|
||||
return YudaoResponse.success({
|
||||
"total": stats.get("total", 0),
|
||||
"pending": stats.get("pending", 0),
|
||||
"confirmed": stats.get("confirmed", 0),
|
||||
"ignored": stats.get("ignored", 0),
|
||||
"resolved": stats.get("resolved", 0),
|
||||
"dispatched": stats.get("dispatched", 0),
|
||||
"byType": stats.get("by_type", {}),
|
||||
"byLevel": stats.get("by_level", {}),
|
||||
})
|
||||
|
||||
|
||||
# ==================== 摄像头告警汇总 ====================
|
||||
|
||||
@router.get("/camera-summary/page")
|
||||
async def get_camera_summary_page(
|
||||
pageNo: int = Query(1, ge=1, description="页码"),
|
||||
pageSize: int = Query(20, ge=1, le=100, description="每页大小"),
|
||||
service: AlertService = Depends(get_alert_service),
|
||||
current_user: dict = Depends(get_current_user)
|
||||
):
|
||||
"""
|
||||
获取摄像头告警汇总（分页）
|
||||
|
||||
以摄像头为维度聚合告警数据
|
||||
|
||||
权限标识：ai-alert:camera-summary:query
|
||||
"""
|
||||
result = service.get_camera_alert_summary(page=pageNo, page_size=pageSize)
|
||||
|
||||
return YudaoResponse.page(
|
||||
list_data=result.get("list", []),
|
||||
total=result.get("total", 0),
|
||||
page_no=pageNo,
|
||||
page_size=pageSize
|
||||
)
|
||||
|
||||
|
||||
# ==================== 辅助函数 ====================
|
||||
|
||||
def _get_alert_type_name(alert_type: Optional[str]) -> str:
|
||||
"""获取告警类型名称"""
|
||||
type_names = {
|
||||
"leave_post": "离岗检测",
|
||||
"intrusion": "周界入侵",
|
||||
"crowd": "人群聚集",
|
||||
"fire": "火焰检测",
|
||||
"smoke": "烟雾检测",
|
||||
"fall": "跌倒检测",
|
||||
"helmet": "安全帽检测",
|
||||
"unknown": "未知类型",
|
||||
}
|
||||
return type_names.get(alert_type, alert_type or "未知类型")
|
||||
357
app/routers/yudao_auth.py
Normal file
357
app/routers/yudao_auth.py
Normal file
@@ -0,0 +1,357 @@
|
||||
"""
|
||||
芋道认证兼容路由
|
||||
|
||||
提供与芋道主平台一致的认证接口，便于前端统一对接。
|
||||
|
||||
测试阶段 (dev_mode=True)：
|
||||
- 本地验证 admin/admin
|
||||
- 返回模拟的 Token 和用户信息
|
||||
|
||||
生产阶段 (dev_mode=False)：
|
||||
- 转发到芋道主平台验证
|
||||
- 返回真实的 Token 和用户信息
|
||||
|
||||
API 路径规范：
|
||||
- /admin-api/system/auth/login - 登录
|
||||
- /admin-api/system/auth/logout - 登出
|
||||
- /admin-api/system/auth/refresh-token - 刷新令牌
|
||||
- /admin-api/system/auth/get-permission-info - 获取权限信息
|
||||
"""
|
||||
|
||||
from fastapi import APIRouter, HTTPException, Header, Query
|
||||
from pydantic import BaseModel
|
||||
from typing import Optional
|
||||
import time
|
||||
import secrets
|
||||
|
||||
from app.config import settings
|
||||
from app.yudao_compat import YudaoResponse
|
||||
|
||||
router = APIRouter(prefix="/admin-api/system", tags=["系统认证"])
|
||||
|
||||
|
||||
class LoginRequest(BaseModel):
|
||||
"""登录请求"""
|
||||
username: str
|
||||
password: str
|
||||
captchaVerification: Optional[str] = None
|
||||
# 社交登录参数（预留）
|
||||
socialType: Optional[int] = None
|
||||
socialCode: Optional[str] = None
|
||||
socialState: Optional[str] = None
|
||||
|
||||
|
||||
@router.post("/auth/login")
|
||||
async def login(req: LoginRequest):
|
||||
"""
|
||||
登录接口
|
||||
|
||||
测试阶段：验证 admin/admin，返回模拟 Token
|
||||
生产阶段：转发到芋道主平台验证
|
||||
|
||||
Returns:
|
||||
{
|
||||
"code": 0,
|
||||
"data": {
|
||||
"userId": 1,
|
||||
"accessToken": "xxx",
|
||||
"refreshToken": "xxx",
|
||||
"expiresTime": 1234567890000
|
||||
},
|
||||
"msg": "success"
|
||||
}
|
||||
"""
|
||||
if settings.app.dev_mode:
|
||||
# 测试阶段：简单验证 admin/admin
|
||||
if req.username == "admin" and req.password == "admin":
|
||||
# 生成模拟 Token
|
||||
access_token = secrets.token_urlsafe(32)
|
||||
refresh_token = secrets.token_urlsafe(32)
|
||||
|
||||
return YudaoResponse.success({
|
||||
"userId": 1,
|
||||
"accessToken": access_token,
|
||||
"refreshToken": refresh_token,
|
||||
"expiresTime": int(time.time() * 1000) + 24 * 60 * 60 * 1000 # 24小时后过期
|
||||
})
|
||||
|
||||
raise HTTPException(status_code=401, detail="用户名或密码错误")
|
||||
|
||||
# TODO: 生产阶段转发到芋道主平台
|
||||
# response = requests.post(
|
||||
# f"{settings.yudao.base_url}/system/auth/login",
|
||||
# json=req.dict()
|
||||
# )
|
||||
# return response.json()
|
||||
|
||||
raise HTTPException(status_code=501, detail="生产模式认证未实现，请配置芋道主平台地址")
|
||||
|
||||
|
||||
@router.get("/auth/get-permission-info")
|
||||
async def get_permission_info(
|
||||
authorization: Optional[str] = Header(None, alias="Authorization")
|
||||
):
|
||||
"""
|
||||
获取当前用户权限信息
|
||||
|
||||
测试阶段：返回超级管理员权限
|
||||
生产阶段：验证 Token 并获取真实权限
|
||||
|
||||
Returns:
|
||||
{
|
||||
"code": 0,
|
||||
"data": {
|
||||
"user": { "id": 1, "nickname": "超级管理员", ... },
|
||||
"roles": ["super_admin"],
|
||||
"permissions": ["*:*:*"]
|
||||
},
|
||||
"msg": "success"
|
||||
}
|
||||
"""
|
||||
if settings.app.dev_mode:
|
||||
return YudaoResponse.success({
|
||||
"user": {
|
||||
"id": 1,
|
||||
"nickname": "超级管理员",
|
||||
"avatar": "",
|
||||
"deptId": 1,
|
||||
"homePath": "/aiot/alarm/list"
|
||||
},
|
||||
"roles": ["super_admin"],
|
||||
"permissions": ["*:*:*"], # 超级权限，拥有所有操作权限
|
||||
"menus": _build_aiot_menus()
|
||||
})
|
||||
|
||||
# 生产阶段：验证 Token
|
||||
if not authorization or not authorization.startswith("Bearer "):
|
||||
raise HTTPException(status_code=401, detail="请先登录")
|
||||
|
||||
# TODO: 生产阶段调用芋道主平台验证
|
||||
# token = authorization.replace("Bearer ", "")
|
||||
# response = requests.get(
|
||||
# f"{settings.yudao.base_url}/system/auth/get-permission-info",
|
||||
# headers={"Authorization": f"Bearer {token}"}
|
||||
# )
|
||||
# return response.json()
|
||||
|
||||
raise HTTPException(status_code=501, detail="生产模式认证未实现")
|
||||
|
||||
|
||||
@router.post("/auth/logout")
|
||||
async def logout(
|
||||
authorization: Optional[str] = Header(None, alias="Authorization")
|
||||
):
|
||||
"""
|
||||
登出接口
|
||||
|
||||
清除用户登录状态
|
||||
"""
|
||||
# 测试阶段和生产阶段都直接返回成功
|
||||
# 实际的 Token 失效由前端清除 localStorage 完成
|
||||
return YudaoResponse.success(True)
|
||||
|
||||
|
||||
@router.post("/auth/refresh-token")
|
||||
async def refresh_token(
|
||||
refreshToken: str = Query(..., description="刷新令牌")
|
||||
):
|
||||
"""
|
||||
刷新访问令牌
|
||||
|
||||
使用 refreshToken 获取新的 accessToken
|
||||
|
||||
Returns:
|
||||
{
|
||||
"code": 0,
|
||||
"data": {
|
||||
"accessToken": "xxx",
|
||||
"refreshToken": "xxx",
|
||||
"expiresTime": 1234567890000
|
||||
},
|
||||
"msg": "success"
|
||||
}
|
||||
"""
|
||||
if settings.app.dev_mode:
|
||||
# 测试阶段：直接返回新 Token
|
||||
new_access_token = secrets.token_urlsafe(32)
|
||||
new_refresh_token = secrets.token_urlsafe(32)
|
||||
|
||||
return YudaoResponse.success({
|
||||
"accessToken": new_access_token,
|
||||
"refreshToken": new_refresh_token,
|
||||
"expiresTime": int(time.time() * 1000) + 24 * 60 * 60 * 1000
|
||||
})
|
||||
|
||||
# TODO: 生产阶段调用芋道主平台刷新
|
||||
raise HTTPException(status_code=501, detail="生产模式认证未实现")
|
||||
|
||||
|
||||
# ==================== 租户相关接口（预留） ====================
|
||||
|
||||
@router.get("/tenant/simple-list")
|
||||
async def get_tenant_list():
|
||||
"""
|
||||
获取租户列表
|
||||
|
||||
测试阶段：返回空列表（前端 VITE_APP_TENANT_ENABLE=false 不会调用）
|
||||
"""
|
||||
return YudaoResponse.success([])
|
||||
|
||||
|
||||
@router.get("/tenant/get-by-website")
|
||||
async def get_tenant_by_website(website: str = Query(...)):
|
||||
"""
|
||||
根据域名获取租户
|
||||
|
||||
测试阶段：返回默认租户
|
||||
"""
|
||||
return YudaoResponse.success({
|
||||
"id": 1,
|
||||
"name": "默认租户"
|
||||
})
|
||||
|
||||
|
||||
# ==================== 验证码相关接口（预留） ====================
|
||||
|
||||
@router.post("/captcha/get")
|
||||
async def get_captcha(data: dict):
|
||||
"""
|
||||
获取验证码
|
||||
|
||||
测试阶段：前端 VITE_APP_CAPTCHA_ENABLE=false 不会调用
|
||||
"""
|
||||
return YudaoResponse.success({
|
||||
"enable": False
|
||||
})
|
||||
|
||||
|
||||
@router.post("/captcha/check")
|
||||
async def check_captcha(data: dict):
|
||||
"""
|
||||
校验验证码
|
||||
|
||||
测试阶段：直接返回成功
|
||||
"""
|
||||
return YudaoResponse.success(True)
|
||||
|
||||
|
||||
# ==================== 字典数据接口（预留） ====================
|
||||
|
||||
@router.get("/dict-data/simple-list")
|
||||
async def get_dict_data_simple_list():
|
||||
"""
|
||||
获取字典数据列表
|
||||
|
||||
测试阶段：返回空列表
|
||||
"""
|
||||
return YudaoResponse.success([])
|
||||
|
||||
|
||||
@router.get("/notify-message/get-unread-count")
|
||||
async def get_unread_message_count():
|
||||
"""
|
||||
获取未读消息数量
|
||||
|
||||
测试阶段：返回 0
|
||||
"""
|
||||
return YudaoResponse.success(0)
|
||||
|
||||
|
||||
# ==================== 辅助函数 ====================
|
||||
|
||||
def _build_aiot_menus():
|
||||
"""
|
||||
构建 AIoT 菜单树
|
||||
|
||||
芋道前端 (accessMode='backend') 期望的菜单格式：
|
||||
- id, parentId, name, path, component, icon, sort, visible, keepAlive
|
||||
- 顶层 parentId=0, component='Layout'
|
||||
- 叶子节点 component 为 views/ 下的路径
|
||||
"""
|
||||
return [
|
||||
{
|
||||
"id": 2000,
|
||||
"parentId": 0,
|
||||
"name": "AIoT 智能平台",
|
||||
"path": "aiot",
|
||||
"component": "Layout",
|
||||
"icon": "ep:monitor",
|
||||
"sort": 10,
|
||||
"visible": True,
|
||||
"keepAlive": True,
|
||||
"children": [
|
||||
{
|
||||
"id": 2010,
|
||||
"parentId": 2000,
|
||||
"name": "告警列表",
|
||||
"path": "alarm/list",
|
||||
"component": "aiot/alarm/list/index",
|
||||
"componentName": "AiotAlarmList",
|
||||
"icon": "ep:warning",
|
||||
"sort": 1,
|
||||
"visible": True,
|
||||
"keepAlive": True,
|
||||
},
|
||||
{
|
||||
"id": 2011,
|
||||
"parentId": 2000,
|
||||
"name": "摄像头告警汇总",
|
||||
"path": "alarm/summary",
|
||||
"component": "aiot/alarm/summary/index",
|
||||
"componentName": "AiotAlarmSummary",
|
||||
"icon": "ep:data-analysis",
|
||||
"sort": 2,
|
||||
"visible": True,
|
||||
"keepAlive": True,
|
||||
},
|
||||
{
|
||||
"id": 2020,
|
||||
"parentId": 2000,
|
||||
"name": "摄像头管理",
|
||||
"path": "device/camera",
|
||||
"component": "aiot/device/camera/index",
|
||||
"componentName": "AiotDeviceCamera",
|
||||
"icon": "ep:camera",
|
||||
"sort": 3,
|
||||
"visible": True,
|
||||
"keepAlive": True,
|
||||
},
|
||||
{
|
||||
"id": 2021,
|
||||
"parentId": 2000,
|
||||
"name": "ROI 区域配置",
|
||||
"path": "device/roi",
|
||||
"component": "aiot/device/roi/index",
|
||||
"componentName": "AiotDeviceRoi",
|
||||
"icon": "ep:aim",
|
||||
"sort": 4,
|
||||
"visible": True,
|
||||
"keepAlive": True,
|
||||
},
|
||||
{
|
||||
"id": 2030,
|
||||
"parentId": 2000,
|
||||
"name": "实时视频",
|
||||
"path": "video/live",
|
||||
"component": "aiot/video/live/index",
|
||||
"componentName": "AiotVideoLive",
|
||||
"icon": "ep:video-camera",
|
||||
"sort": 5,
|
||||
"visible": True,
|
||||
"keepAlive": True,
|
||||
},
|
||||
{
|
||||
"id": 2040,
|
||||
"parentId": 2000,
|
||||
"name": "边缘节点管理",
|
||||
"path": "edge/node",
|
||||
"component": "aiot/edge/node/index",
|
||||
"componentName": "AiotEdgeNode",
|
||||
"icon": "ep:cpu",
|
||||
"sort": 6,
|
||||
"visible": True,
|
||||
"keepAlive": True,
|
||||
},
|
||||
],
|
||||
}
|
||||
]
|
||||
149
app/yudao_compat.py
Normal file
149
app/yudao_compat.py
Normal file
@@ -0,0 +1,149 @@
|
||||
"""
|
||||
芋道平台兼容层
|
||||
|
||||
保持与芋道主平台一致的接口规范，便于后续无缝对接。
|
||||
|
||||
设计原则：
|
||||
- 响应格式符合芋道标准: {"code": 0, "data": ..., "msg": "success"}
|
||||
- 分页参数: pageNo, pageSize
|
||||
- HTTP 状态码始终 200，业务错误通过 code 区分
|
||||
"""
|
||||
|
||||
from typing import Any, Optional
|
||||
from fastapi import HTTPException, Header, Request
|
||||
from fastapi.responses import JSONResponse
|
||||
|
||||
from app.config import settings
|
||||
|
||||
|
||||
class YudaoResponse:
|
||||
"""芋道标准响应格式"""
|
||||
|
||||
@staticmethod
|
||||
def success(data: Any = None, msg: str = "success") -> dict:
|
||||
"""成功响应"""
|
||||
return {"code": 0, "data": data, "msg": msg}
|
||||
|
||||
@staticmethod
|
||||
def error(code: int, msg: str) -> dict:
|
||||
"""错误响应"""
|
||||
return {"code": code, "data": None, "msg": msg}
|
||||
|
||||
@staticmethod
|
||||
def page(list_data: list, total: int, page_no: int, page_size: int) -> dict:
|
||||
"""
|
||||
分页响应格式
|
||||
|
||||
Args:
|
||||
list_data: 数据列表
|
||||
total: 总记录数
|
||||
page_no: 当前页码
|
||||
page_size: 每页大小
|
||||
"""
|
||||
return {
|
||||
"code": 0,
|
||||
"data": {
|
||||
"list": list_data,
|
||||
"total": total,
|
||||
"pageNo": page_no,
|
||||
"pageSize": page_size
|
||||
},
|
||||
"msg": "success"
|
||||
}
|
||||
|
||||
|
||||
async def yudao_exception_handler(request: Request, exc: HTTPException) -> JSONResponse:
|
||||
"""
|
||||
统一异常处理器
|
||||
|
||||
芋道规范：HTTP 状态码始终返回 200，业务错误通过 code 区分
|
||||
"""
|
||||
return JSONResponse(
|
||||
status_code=200,
|
||||
content=YudaoResponse.error(exc.status_code, exc.detail)
|
||||
)
|
||||
|
||||
|
||||
def get_current_user(authorization: Optional[str] = Header(None, alias="Authorization")) -> dict:
|
||||
"""
|
||||
获取当前用户
|
||||
|
||||
测试阶段 (dev_mode=True)：
|
||||
- 跳过 Token 验证
|
||||
- 返回模拟的超级管理员用户
|
||||
- 拥有 ["*:*:*"] 全部权限
|
||||
|
||||
生产阶段 (dev_mode=False)：
|
||||
- 验证 Authorization Header
|
||||
- 调用芋道主平台验证 Token
|
||||
- 返回真实用户信息
|
||||
|
||||
Returns:
|
||||
dict: 用户信息，包含 userId, username, permissions
|
||||
"""
|
||||
if settings.app.dev_mode:
|
||||
# 测试阶段：返回模拟的超级管理员
|
||||
return {
|
||||
"userId": 1,
|
||||
"username": "admin",
|
||||
"nickname": "超级管理员",
|
||||
"permissions": ["*:*:*"]
|
||||
}
|
||||
|
||||
# 生产阶段：验证 Token
|
||||
if not authorization or not authorization.startswith("Bearer "):
|
||||
raise HTTPException(status_code=401, detail="请先登录")
|
||||
|
||||
token = authorization.replace("Bearer ", "")
|
||||
|
||||
# TODO: 生产阶段调用芋道主平台验证 Token
|
||||
# response = requests.get(
|
||||
# f"{settings.yudao.base_url}/system/auth/check-token",
|
||||
# headers={"Authorization": f"Bearer {token}"}
|
||||
# )
|
||||
# if response.status_code != 200:
|
||||
# raise HTTPException(status_code=401, detail="Token 无效或已过期")
|
||||
# return response.json()["data"]["user"]
|
||||
|
||||
# 临时：生产模式未实现时返回模拟用户
|
||||
return {
|
||||
"userId": 1,
|
||||
"username": "admin",
|
||||
"nickname": "超级管理员",
|
||||
"permissions": ["*:*:*"]
|
||||
}
|
||||
|
||||
|
||||
def check_permission(required_permission: str):
|
||||
"""
|
||||
权限检查装饰器
|
||||
|
||||
测试阶段：始终通过（超级用户拥有 *:*:* 权限）
|
||||
生产阶段：检查用户是否拥有指定权限
|
||||
|
||||
Usage:
|
||||
@router.get("/alert/page")
|
||||
@check_permission("ai-alert:alert:query")
|
||||
async def get_alert_page(...):
|
||||
...
|
||||
"""
|
||||
def decorator(func):
|
||||
async def wrapper(*args, current_user: dict = None, **kwargs):
|
||||
if current_user is None:
|
||||
raise HTTPException(status_code=401, detail="请先登录")
|
||||
|
||||
permissions = current_user.get("permissions", [])
|
||||
|
||||
# 超级权限检查
|
||||
if "*:*:*" in permissions:
|
||||
return await func(*args, current_user=current_user, **kwargs)
|
||||
|
||||
# 具体权限检查
|
||||
if required_permission not in permissions:
|
||||
raise HTTPException(status_code=403, detail="没有操作权限")
|
||||
|
||||
return await func(*args, current_user=current_user, **kwargs)
|
||||
|
||||
return wrapper
|
||||
|
||||
return decorator
|
||||
Reference in New Issue
Block a user