Spaces:
Sleeping
Sleeping
File size: 2,890 Bytes
551658a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 | from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from typing import List
from database import get_db
from models.operation_log import OperationLog
from models.user import User
from schemas.operation_log import OperationLogResponse, OperationLogCreate
from routers.auth import get_current_user, get_current_superadmin, is_admin_user
router = APIRouter()
# 记录操作日志的辅助函数
def create_operation_log(db: Session, user: User, module: str, operation_type: str, operation_detail: str = None, before_data: str = None, after_data: str = None, company_code: str = None):
"""创建操作日志"""
log = OperationLog(
user_id=user.id,
username=user.username,
company_code=company_code,
module=module,
operation_type=operation_type,
operation_detail=operation_detail,
before_data=before_data,
after_data=after_data
)
db.add(log)
db.commit()
db.refresh(log)
return log
# 获取操作日志列表
@router.get("/", response_model=List[OperationLogResponse])
async def get_operation_logs(
module: str = None,
operation_type: str = None,
current_user: User = Depends(get_current_user),
db: Session = Depends(get_db)
):
"""获取操作日志列表，支持按模块和操作类型过滤"""
query = db.query(OperationLog)
# 数据权限分级：普通用户只能看到自己公司的日志，admin和超级管理员可以看到所有日志
if not is_admin_user(current_user) and not current_user.is_superadmin and current_user.company_code:
query = query.filter(OperationLog.company_code == current_user.company_code)
if module:
query = query.filter(OperationLog.module == module)
if operation_type:
query = query.filter(OperationLog.operation_type == operation_type)
# 只返回前10条，并按照日期倒序排列
logs = query.order_by(OperationLog.created_at.desc()).limit(10).all()
return logs
# 获取指定模块的操作日志
@router.get("/module/{module}", response_model=List[OperationLogResponse])
async def get_operation_logs_by_module(
module: str,
current_user: User = Depends(get_current_user),
db: Session = Depends(get_db)
):
"""获取指定模块的操作日志"""
query = db.query(OperationLog).filter(
OperationLog.module == module
)
# 数据权限分级：普通用户只能看到自己公司的日志，admin和超级管理员可以看到所有日志
if not is_admin_user(current_user) and not current_user.is_superadmin and current_user.company_code:
query = query.filter(OperationLog.company_code == current_user.company_code)
# 只返回前10条，并按照日期倒序排列
logs = query.order_by(OperationLog.created_at.desc()).limit(10).all()
return logs
|