From 8a8ae4ed57f2172a93183f60a7b7c3132e4de9f9 Mon Sep 17 00:00:00 2001 From: v6ole Date: Fri, 12 Jun 2026 11:49:53 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E8=B7=AF=E7=94=B1=E9=A1=BA=E5=BA=8F?= =?UTF-8?q?=E5=AF=BC=E8=87=B4=20422=20+=20=E5=BD=BB=E5=BA=95=E6=B6=88?= =?UTF-8?q?=E9=99=A4=20Header(...)=20=E9=9A=90=E6=82=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. devices.py: /tags 和 /export/csv 移到 /{device_id}(int) 之前， 避免 FastAPI 将 'tags'/'export' 当作 device_id 解析失败返回 422 2. audit.py: /logs/export/csv 移到 /logs/{log_id}(int) 之前，同上 3. auth.py: Header(...) → Header(None)，统一为手动 401 返回， 消除全局最后一个 Header(...) 导致的 422 隐患 Co-Authored-By: Claude --- backend/app/api/v1/audit.py | 78 ++++++++++++++-------------- backend/app/api/v1/auth.py | 8 +-- backend/app/api/v1/devices.py | 98 +++++++++++++++++------------------ 3 files changed, 92 insertions(+), 92 deletions(-) diff --git a/backend/app/api/v1/audit.py b/backend/app/api/v1/audit.py index d3f578c..ed1c3b4 100644 --- a/backend/app/api/v1/audit.py +++ b/backend/app/api/v1/audit.py @@ -49,45 +49,6 @@ def get_audit_logs( } -@router.get("/logs/{log_id}") -def get_audit_log_detail( - log_id: int, - db: Session = Depends(get_db), - _: dict = Depends(require_permission('*')), -): - """获取单条审计日志详情""" - log = db.query(AuditLog).filter(AuditLog.id == log_id).first() - if not log: - from fastapi import HTTPException - raise HTTPException(status_code=404, detail="日志不存在") - return _fmt(log, detail=True) - - -@router.get("/stats") -def get_audit_stats( - days: int = Query(7, ge=1, le=90), - db: Session = Depends(get_db), - _: dict = Depends(require_permission('*')), -): - """审计日志统计（最近N天）""" - from datetime import timedelta - from sqlalchemy import func - since = datetime.utcnow() - timedelta(days=days) - rows = ( - db.query(AuditLog.action_type, AuditLog.status, func.count().label("cnt")) - .filter(AuditLog.action_time >= since) - .group_by(AuditLog.action_type, AuditLog.status) - .all() - ) - total = db.query(func.count(AuditLog.id)).filter(AuditLog.action_time >= since).scalar() - by_type = {} - for row in rows: - if row.action_type not in by_type: - by_type[row.action_type] = {"success": 0, "failed": 0, "error": 0} - by_type[row.action_type][row.status] = row.cnt - return {"total": total, "days": days, "by_type": by_type} - - @router.get("/logs/export/csv") def export_audit_logs( start_time: Optional[datetime] = Query(None), @@ -130,6 +91,45 @@ def export_audit_logs( ) +@router.get("/logs/{log_id}") +def get_audit_log_detail( + log_id: int, + db: Session = Depends(get_db), + _: dict = Depends(require_permission('*')), +): + """获取单条审计日志详情""" + log = db.query(AuditLog).filter(AuditLog.id == log_id).first() + if not log: + from fastapi import HTTPException + raise HTTPException(status_code=404, detail="日志不存在") + return _fmt(log, detail=True) + + +@router.get("/stats") +def get_audit_stats( + days: int = Query(7, ge=1, le=90), + db: Session = Depends(get_db), + _: dict = Depends(require_permission('*')), +): + """审计日志统计（最近N天）""" + from datetime import timedelta + from sqlalchemy import func + since = datetime.utcnow() - timedelta(days=days) + rows = ( + db.query(AuditLog.action_type, AuditLog.status, func.count().label("cnt")) + .filter(AuditLog.action_time >= since) + .group_by(AuditLog.action_type, AuditLog.status) + .all() + ) + total = db.query(func.count(AuditLog.id)).filter(AuditLog.action_time >= since).scalar() + by_type = {} + for row in rows: + if row.action_type not in by_type: + by_type[row.action_type] = {"success": 0, "failed": 0, "error": 0} + by_type[row.action_type][row.status] = row.cnt + return {"total": total, "days": days, "by_type": by_type} + + def _fmt(r: AuditLog, detail: bool = False) -> dict: base = { "id": r.id, diff --git a/backend/app/api/v1/auth.py b/backend/app/api/v1/auth.py index 3d72568..7ba59e3 100644 --- a/backend/app/api/v1/auth.py +++ b/backend/app/api/v1/auth.py @@ -84,12 +84,12 @@ def callback(body: CallbackRequest, db: Session = Depends(get_db)): @router.get("/permissions") def get_my_permissions( - authorization: str = Header(..., alias="Authorization"), + authorization: str = Header(None, alias="Authorization"), db: Session = Depends(get_db) ): """获取当前用户的权限码列表""" from app.middleware.permission_middleware import get_role_permissions - if not authorization.startswith("Bearer "): + if not authorization or not authorization.startswith("Bearer "): raise HTTPException(status_code=401, detail="未授权") token = authorization[7:] payload = verify_token(token) @@ -102,11 +102,11 @@ def get_my_permissions( @router.get("/profile") def get_profile( - authorization: str = Header(..., alias="Authorization"), + authorization: str = Header(None, alias="Authorization"), db: Session = Depends(get_db) ): """获取当前用户信息""" - if not authorization.startswith("Bearer "): + if not authorization or not authorization.startswith("Bearer "): raise HTTPException(status_code=401, detail="未授权") token = authorization[7:] payload = verify_token(token) diff --git a/backend/app/api/v1/devices.py b/backend/app/api/v1/devices.py index 784a8af..5bb9ff3 100644 --- a/backend/app/api/v1/devices.py +++ b/backend/app/api/v1/devices.py @@ -338,6 +338,55 @@ def export_replacements( ) +@router.get("/tags") +def get_all_tags(db: Session = Depends(get_db)): + """获取所有不重复的设备标签""" + from sqlalchemy import func as _func + rows = db.query(ONUDevice.tags).filter( + ONUDevice.tags.isnot(None), ONUDevice.tags != '' + ).all() + tags = set() + for (tag_str,) in rows: + for t in tag_str.split(','): + t = t.strip() + if t: + tags.add(t) + return sorted(tags) + + +@router.get("/export/csv") +def export_devices_csv( + region: Optional[str] = Query(None), + school_name: Optional[str] = Query(None), + db: Session = Depends(get_db), + _: dict = Depends(require_permission('device.view')), +): + """导出设备列表为 CSV""" + from sqlalchemy import func as _func + + query = db.query(ONUDevice) + if region: + query = query.filter(ONUDevice.region == region) + if school_name: + query = query.filter(ONUDevice.school_name == school_name) + devices = query.order_by(ONUDevice.region, ONUDevice.school_name).all() + + output = io.StringIO() + writer = csv.writer(output) + writer.writerow(["MAC地址", "区域", "学校", "楼宇", "场所类型", "房间号", "端口", "型号", "LOID", "距离(m)", "备注"]) + for d in devices: + writer.writerow([d.mac_address, d.region or "", d.school_name or "", d.building or "", + d.place_type or "", d.room_number or "", d.port_id or "", d.model or "", + d.loid or "", d.distance_m or "", d.notes or ""]) + + output.seek(0) + return StreamingResponse( + iter([output.getvalue()]), + media_type="text/csv", + headers={"Content-Disposition": "attachment; filename=onu_devices.csv"} + ) + + @router.get("/{device_id}", response_model=ONUDeviceResponse) def get_device( device_id: int, @@ -701,52 +750,3 @@ def get_optical_power_history( "recorded_at": r.recorded_at.isoformat() if r.recorded_at else None} for r in rows ] - - -@router.get("/tags") -def get_all_tags(db: Session = Depends(get_db)): - """获取所有不重复的设备标签""" - from sqlalchemy import func as _func - rows = db.query(ONUDevice.tags).filter( - ONUDevice.tags.isnot(None), ONUDevice.tags != '' - ).all() - tags = set() - for (tag_str,) in rows: - for t in tag_str.split(','): - t = t.strip() - if t: - tags.add(t) - return sorted(tags) - - -@router.get("/export/csv") -def export_devices_csv( - region: Optional[str] = Query(None), - school_name: Optional[str] = Query(None), - db: Session = Depends(get_db), - _: dict = Depends(require_permission('device.view')), -): - """导出设备列表为 CSV""" - from sqlalchemy import func as _func - - query = db.query(ONUDevice) - if region: - query = query.filter(ONUDevice.region == region) - if school_name: - query = query.filter(ONUDevice.school_name == school_name) - devices = query.order_by(ONUDevice.region, ONUDevice.school_name).all() - - output = io.StringIO() - writer = csv.writer(output) - writer.writerow(["MAC地址", "区域", "学校", "楼宇", "场所类型", "房间号", "端口", "型号", "LOID", "距离(m)", "备注"]) - for d in devices: - writer.writerow([d.mac_address, d.region or "", d.school_name or "", d.building or "", - d.place_type or "", d.room_number or "", d.port_id or "", d.model or "", - d.loid or "", d.distance_m or "", d.notes or ""]) - - output.seek(0) - return StreamingResponse( - iter([output.getvalue()]), - media_type="text/csv", - headers={"Content-Disposition": "attachment; filename=onu_devices.csv"} - )