fix: 路由顺序导致 422 + 彻底消除 Header(...) 隐患

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 <noreply@anthropic.com>
This commit is contained in:
2026-06-12 11:49:53 +08:00
parent a5ac25a01d
commit 8a8ae4ed57
3 changed files with 92 additions and 92 deletions
+49 -49
View File
@@ -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"}
)