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:
+39
-39
@@ -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,
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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"}
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user