998211c483
- 从服务器拉取完整代码 - 按框架规范整理项目结构 - 配置 Drone CI 测试环境部署 - 包含后端(FastAPI)、前端(Vue3)、管理端 技术栈: Vue3 + TypeScript + FastAPI + MySQL
185 lines
5.6 KiB
Python
185 lines
5.6 KiB
Python
"""
|
||
系统日志 API
|
||
提供日志查询、筛选、详情查看等功能
|
||
"""
|
||
import logging
|
||
from typing import Optional
|
||
from datetime import datetime
|
||
from fastapi import APIRouter, Depends, HTTPException, Query
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.core.deps import get_db, get_current_user
|
||
from app.models.user import User
|
||
from app.schemas.base import ResponseModel
|
||
from app.schemas.system_log import (
|
||
SystemLogCreate,
|
||
SystemLogResponse,
|
||
SystemLogQuery,
|
||
SystemLogListResponse
|
||
)
|
||
from app.services.system_log_service import system_log_service
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
router = APIRouter(prefix="/admin/logs")
|
||
|
||
|
||
@router.get("", response_model=ResponseModel[SystemLogListResponse])
|
||
async def get_system_logs(
|
||
level: Optional[str] = Query(None, description="日志级别筛选"),
|
||
type: Optional[str] = Query(None, description="日志类型筛选"),
|
||
user: Optional[str] = Query(None, description="用户筛选"),
|
||
keyword: Optional[str] = Query(None, description="关键词搜索"),
|
||
start_date: Optional[datetime] = Query(None, description="开始日期"),
|
||
end_date: Optional[datetime] = Query(None, description="结束日期"),
|
||
page: int = Query(1, ge=1, description="页码"),
|
||
page_size: int = Query(20, ge=1, le=100, description="每页数量"),
|
||
db: AsyncSession = Depends(get_db),
|
||
current_user: User = Depends(get_current_user)
|
||
):
|
||
"""
|
||
获取系统日志列表
|
||
支持按级别、类型、用户、关键词、日期范围筛选
|
||
仅管理员可访问
|
||
"""
|
||
try:
|
||
# 权限检查：仅管理员可查看系统日志
|
||
if current_user.role != "admin":
|
||
raise HTTPException(status_code=403, detail="无权限访问系统日志")
|
||
|
||
# 构建查询参数
|
||
query_params = SystemLogQuery(
|
||
level=level,
|
||
type=type,
|
||
user=user,
|
||
keyword=keyword,
|
||
start_date=start_date,
|
||
end_date=end_date,
|
||
page=page,
|
||
page_size=page_size
|
||
)
|
||
|
||
# 查询日志
|
||
logs, total = await system_log_service.get_logs(db, query_params)
|
||
|
||
# 计算总页数
|
||
total_pages = (total + page_size - 1) // page_size
|
||
|
||
# 转换为响应格式
|
||
log_responses = [SystemLogResponse.model_validate(log) for log in logs]
|
||
|
||
response_data = SystemLogListResponse(
|
||
items=log_responses,
|
||
total=total,
|
||
page=page,
|
||
page_size=page_size,
|
||
total_pages=total_pages
|
||
)
|
||
|
||
return ResponseModel(
|
||
code=200,
|
||
message="获取系统日志成功",
|
||
data=response_data
|
||
)
|
||
|
||
except HTTPException:
|
||
raise
|
||
except Exception as e:
|
||
logger.error(f"获取系统日志失败: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=f"获取系统日志失败: {str(e)}")
|
||
|
||
|
||
@router.get("/{log_id}", response_model=ResponseModel[SystemLogResponse])
|
||
async def get_log_detail(
|
||
log_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
current_user: User = Depends(get_current_user)
|
||
):
|
||
"""
|
||
获取日志详情
|
||
仅管理员可访问
|
||
"""
|
||
try:
|
||
# 权限检查
|
||
if current_user.role != "admin":
|
||
raise HTTPException(status_code=403, detail="无权限访问系统日志")
|
||
|
||
# 查询日志
|
||
log = await system_log_service.get_log_by_id(db, log_id)
|
||
|
||
if not log:
|
||
raise HTTPException(status_code=404, detail="日志不存在")
|
||
|
||
return ResponseModel(
|
||
code=200,
|
||
message="获取日志详情成功",
|
||
data=SystemLogResponse.model_validate(log)
|
||
)
|
||
|
||
except HTTPException:
|
||
raise
|
||
except Exception as e:
|
||
logger.error(f"获取日志详情失败: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=f"获取日志详情失败: {str(e)}")
|
||
|
||
|
||
@router.post("", response_model=ResponseModel[SystemLogResponse])
|
||
async def create_system_log(
|
||
log_data: SystemLogCreate,
|
||
db: AsyncSession = Depends(get_db)
|
||
):
|
||
"""
|
||
创建系统日志（内部API，供系统各模块调用）
|
||
注意：此接口不需要用户认证，但应该只供内部调用
|
||
"""
|
||
try:
|
||
log = await system_log_service.create_log(db, log_data)
|
||
|
||
return ResponseModel(
|
||
code=200,
|
||
message="创建日志成功",
|
||
data=SystemLogResponse.model_validate(log)
|
||
)
|
||
|
||
except Exception as e:
|
||
logger.error(f"创建日志失败: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=f"创建日志失败: {str(e)}")
|
||
|
||
|
||
@router.delete("/cleanup")
|
||
async def cleanup_old_logs(
|
||
before_days: int = Query(90, ge=1, description="删除多少天之前的日志"),
|
||
db: AsyncSession = Depends(get_db),
|
||
current_user: User = Depends(get_current_user)
|
||
):
|
||
"""
|
||
清理旧日志
|
||
仅管理员可访问
|
||
"""
|
||
try:
|
||
# 权限检查
|
||
if current_user.role != "admin":
|
||
raise HTTPException(status_code=403, detail="无权限执行此操作")
|
||
|
||
# 计算截止日期
|
||
from datetime import timedelta
|
||
before_date = datetime.now() - timedelta(days=before_days)
|
||
|
||
# 删除旧日志
|
||
deleted_count = await system_log_service.delete_logs_before_date(db, before_date)
|
||
|
||
return ResponseModel(
|
||
code=200,
|
||
message=f"成功清理 {deleted_count} 条日志",
|
||
data={"deleted_count": deleted_count}
|
||
)
|
||
|
||
except HTTPException:
|
||
raise
|
||
except Exception as e:
|
||
logger.error(f"清理日志失败: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=f"清理日志失败: {str(e)}")
|
||
|
||
|
||
|