Yuzhiran Dev
62 lines
2.3 KiB
Python
62 lines
2.3 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, Request, Query
|
|
from sqlalchemy.orm import Session
|
|
from typing import List, Optional
|
|
from datetime import datetime, timedelta
|
|
|
|
from ..database import get_db
|
|
from ..models import AuditLog, User
|
|
from ..schemas import AuditLogResponse
|
|
from .auth import get_current_admin
|
|
|
|
router = APIRouter(prefix="/api/audit", tags=["audit"])
|
|
|
|
@router.get("/logs", response_model=List[AuditLogResponse])
|
|
def list_audit_logs(
|
|
request: Request,
|
|
db: Session = Depends(get_db),
|
|
admin_user: User = Depends(get_current_admin),
|
|
username: Optional[str] = Query(None, description="按用户名筛选"),
|
|
action: Optional[str] = Query(None, description="按操作类型筛选"),
|
|
resource_type: Optional[str] = Query(None, description="按资源类型筛选"),
|
|
start_date: Optional[datetime] = Query(None, description="开始时间"),
|
|
end_date: Optional[datetime] = Query(None, description="结束时间"),
|
|
limit: int = Query(100, ge=1, le=1000, description="返回数量限制")
|
|
):
|
|
"""查询审计日志（管理员）"""
|
|
query = db.query(AuditLog)
|
|
|
|
if username:
|
|
query = query.filter(AuditLog.username == username)
|
|
if action:
|
|
query = query.filter(AuditLog.action == action)
|
|
if resource_type:
|
|
query = query.filter(AuditLog.resource_type == resource_type)
|
|
if start_date:
|
|
query = query.filter(AuditLog.created_at >= start_date)
|
|
if end_date:
|
|
query = query.filter(AuditLog.created_at <= end_date)
|
|
|
|
logs = query.order_by(AuditLog.created_at.desc()).limit(limit).all()
|
|
return logs
|
|
|
|
@router.get("/logs/actions")
|
|
def list_audit_actions(
|
|
request: Request,
|
|
db: Session = Depends(get_db),
|
|
admin_user: bool = Depends(get_current_admin)
|
|
):
|
|
"""获取所有已记录的操作类型列表（用于筛选）"""
|
|
actions = db.query(AuditLog.action).distinct().all()
|
|
return [a[0] for a in actions if a[0]]
|
|
|
|
@router.get("/logs/users")
|
|
def list_audit_users(
|
|
request: Request,
|
|
db: Session = Depends(get_db),
|
|
admin_user: User = Depends(get_current_admin),
|
|
limit: int = Query(50, ge=1, le=200)
|
|
):
|
|
"""获取最近产生审计记录的用户列表"""
|
|
users = db.query(AuditLog.username).distinct().order_by(AuditLog.created_at.desc()).limit(limit).all()
|
|
return [u[0] for u in users if u[0]]
|