lt
2bd3c83cf5
核心功能: - 新增 JWT 认证系统，支持管理员登录/登出 - 前后端合并为单一 FastAPI 应用 (端口 8001) - 系统概览页：6 个统计卡片，点击跳转筛选 - 选题管理页：批量操作 (刷新/创作/优化)，时间列展示 - 系统日志页：整合日志查看功能 - 用户管理页：管理员可创建/删除用户 - 移动端适配：响应式布局，底部导航栏 - 标题居中显示 技术改进: - 添加 generated_at 字段支持创作时间记录 - 状态更新时自动同步 updated_at - 所有 API 路由添加 JWT 认证保护 - 前端 authFetch 封装自动附加 Token - 升级 FastAPI 0.136, Pydantic 2.13 等依赖 修复: - 修复 API 500 错误 (数据库列缺失) - 修复 formatRelativeTime 未定义错误 - 修复登录 Token 存储和自动附加逻辑
73 lines
2.8 KiB
Python
73 lines
2.8 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
|
|
from ..schemas import AuditLogResponse
|
|
from .auth import verify_token
|
|
|
|
router = APIRouter(prefix="/api/audit", tags=["audit"])
|
|
|
|
def get_current_admin(request: Request, db: Session = Depends(get_db)):
|
|
"""依赖项：验证管理员权限"""
|
|
auth_header = request.headers.get("Authorization")
|
|
if not auth_header or not auth_header.startswith("Bearer "):
|
|
raise HTTPException(status_code=401, detail="未提供认证令牌")
|
|
token = auth_header.split(" ")[1]
|
|
user = verify_token(token, db)
|
|
if user.role != "admin":
|
|
raise HTTPException(status_code=403, detail="需要管理员权限")
|
|
return user
|
|
|
|
@router.get("/logs", response_model=List[AuditLogResponse])
|
|
def list_audit_logs(
|
|
request: Request,
|
|
db: Session = Depends(get_db),
|
|
admin_user: bool = 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: bool = 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]]
|