271 lines
7.5 KiB
Python
271 lines
7.5 KiB
Python
"""
|
|
操作日志服务层
|
|
"""
|
|
from typing import Optional, List, Dict, Any
|
|
from datetime import datetime
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from app.crud.operation_log import operation_log_crud
|
|
from app.schemas.operation_log import OperationLogCreate
|
|
|
|
|
|
class OperationLogService:
|
|
"""操作日志服务类"""
|
|
|
|
async def get_log(self, db: AsyncSession, log_id: int) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
获取操作日志详情
|
|
|
|
Args:
|
|
db: 数据库会话
|
|
log_id: 日志ID
|
|
|
|
Returns:
|
|
日志信息
|
|
"""
|
|
log = await operation_log_crud.get(db, log_id)
|
|
if not log:
|
|
return None
|
|
|
|
return {
|
|
"id": log.id,
|
|
"operator_id": log.operator_id,
|
|
"operator_name": log.operator_name,
|
|
"operator_ip": log.operator_ip,
|
|
"module": log.module,
|
|
"operation_type": log.operation_type,
|
|
"method": log.method,
|
|
"url": log.url,
|
|
"params": log.params,
|
|
"result": log.result,
|
|
"error_msg": log.error_msg,
|
|
"duration": log.duration,
|
|
"user_agent": log.user_agent,
|
|
"extra_data": log.extra_data,
|
|
"created_at": log.created_at,
|
|
}
|
|
|
|
async def get_logs(
|
|
self,
|
|
db: AsyncSession,
|
|
*,
|
|
skip: int = 0,
|
|
limit: int = 20,
|
|
operator_id: Optional[int] = None,
|
|
operator_name: Optional[str] = None,
|
|
module: Optional[str] = None,
|
|
operation_type: Optional[str] = None,
|
|
result: Optional[str] = None,
|
|
start_time: Optional[datetime] = None,
|
|
end_time: Optional[datetime] = None,
|
|
keyword: Optional[str] = None
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
获取操作日志列表
|
|
|
|
Args:
|
|
db: 数据库会话
|
|
skip: 跳过条数
|
|
limit: 返回条数
|
|
operator_id: 操作人ID
|
|
operator_name: 操作人姓名
|
|
module: 模块名称
|
|
operation_type: 操作类型
|
|
result: 操作结果
|
|
start_time: 开始时间
|
|
end_time: 结束时间
|
|
keyword: 关键词
|
|
|
|
Returns:
|
|
日志列表和总数
|
|
"""
|
|
items, total = await operation_log_crud.get_multi(
|
|
db,
|
|
skip=skip,
|
|
limit=limit,
|
|
operator_id=operator_id,
|
|
operator_name=operator_name,
|
|
module=module,
|
|
operation_type=operation_type,
|
|
result=result,
|
|
start_time=start_time,
|
|
end_time=end_time,
|
|
keyword=keyword
|
|
)
|
|
|
|
return {
|
|
"items": [
|
|
{
|
|
"id": item.id,
|
|
"operator_id": item.operator_id,
|
|
"operator_name": item.operator_name,
|
|
"operator_ip": item.operator_ip,
|
|
"module": item.module,
|
|
"operation_type": item.operation_type,
|
|
"method": item.method,
|
|
"url": item.url,
|
|
"result": item.result,
|
|
"error_msg": item.error_msg,
|
|
"duration": item.duration,
|
|
"created_at": item.created_at,
|
|
}
|
|
for item in items
|
|
],
|
|
"total": total
|
|
}
|
|
|
|
async def create_log(
|
|
self,
|
|
db: AsyncSession,
|
|
obj_in: OperationLogCreate
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
创建操作日志
|
|
|
|
Args:
|
|
db: 数据库会话
|
|
obj_in: 创建数据
|
|
|
|
Returns:
|
|
创建的日志信息
|
|
"""
|
|
import json
|
|
|
|
# 转换为字典
|
|
obj_in_data = obj_in.model_dump()
|
|
|
|
# 处理复杂类型
|
|
if obj_in_data.get("extra_data"):
|
|
obj_in_data["extra_data"] = json.loads(obj_in.extra_data.model_dump_json()) if isinstance(obj_in.extra_data, dict) else obj_in.extra_data
|
|
|
|
log = await operation_log_crud.create(db, obj_in=obj_in_data)
|
|
|
|
return {
|
|
"id": log.id,
|
|
"operator_name": log.operator_name,
|
|
"module": log.module,
|
|
"operation_type": log.operation_type,
|
|
}
|
|
|
|
async def get_statistics(
|
|
self,
|
|
db: AsyncSession,
|
|
*,
|
|
start_time: Optional[datetime] = None,
|
|
end_time: Optional[datetime] = None
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
获取操作日志统计信息
|
|
|
|
Args:
|
|
db: 数据库会话
|
|
start_time: 开始时间
|
|
end_time: 结束时间
|
|
|
|
Returns:
|
|
统计信息
|
|
"""
|
|
return await operation_log_crud.get_statistics(
|
|
db,
|
|
start_time=start_time,
|
|
end_time=end_time
|
|
)
|
|
|
|
async def get_operator_top(
|
|
self,
|
|
db: AsyncSession,
|
|
*,
|
|
limit: int = 10,
|
|
start_time: Optional[datetime] = None,
|
|
end_time: Optional[datetime] = None
|
|
) -> List[Dict[str, Any]]:
|
|
"""
|
|
获取操作排行榜
|
|
|
|
Args:
|
|
db: 数据库会话
|
|
limit: 返回条数
|
|
start_time: 开始时间
|
|
end_time: 结束时间
|
|
|
|
Returns:
|
|
操作排行列表
|
|
"""
|
|
return await operation_log_crud.get_operator_top(
|
|
db,
|
|
limit=limit,
|
|
start_time=start_time,
|
|
end_time=end_time
|
|
)
|
|
|
|
async def delete_old_logs(self, db: AsyncSession, *, days: int = 90) -> Dict[str, Any]:
|
|
"""
|
|
删除旧日志
|
|
|
|
Args:
|
|
db: 数据库会话
|
|
days: 保留天数
|
|
|
|
Returns:
|
|
删除结果
|
|
"""
|
|
count = await operation_log_crud.delete_old_logs(db, days=days)
|
|
return {
|
|
"deleted_count": count,
|
|
"message": f"已删除 {count} 条 {days} 天前的日志"
|
|
}
|
|
|
|
async def export_logs(
|
|
self,
|
|
db: AsyncSession,
|
|
*,
|
|
start_time: Optional[datetime] = None,
|
|
end_time: Optional[datetime] = None,
|
|
operator_id: Optional[int] = None,
|
|
module: Optional[str] = None,
|
|
operation_type: Optional[str] = None
|
|
) -> List[Dict[str, Any]]:
|
|
"""
|
|
导出操作日志
|
|
|
|
Args:
|
|
db: 数据库会话
|
|
start_time: 开始时间
|
|
end_time: 结束时间
|
|
operator_id: 操作人ID
|
|
module: 模块名称
|
|
operation_type: 操作类型
|
|
|
|
Returns:
|
|
日志列表
|
|
"""
|
|
items, total = await operation_log_crud.get_multi(
|
|
db,
|
|
skip=0,
|
|
limit=10000, # 导出限制
|
|
operator_id=operator_id,
|
|
module=module,
|
|
operation_type=operation_type,
|
|
start_time=start_time,
|
|
end_time=end_time
|
|
)
|
|
|
|
return [
|
|
{
|
|
"操作人": item.operator_name,
|
|
"模块": item.module,
|
|
"操作类型": item.operation_type,
|
|
"请求方法": item.method,
|
|
"请求URL": item.url,
|
|
"操作结果": item.result,
|
|
"错误信息": item.error_msg or "",
|
|
"执行时长(毫秒)": item.duration or 0,
|
|
"操作时间": item.created_at.strftime("%Y-%m-%d %H:%M:%S"),
|
|
"操作IP": item.operator_ip or "",
|
|
}
|
|
for item in items
|
|
]
|
|
|
|
|
|
# 创建全局实例
|
|
operation_log_service = OperationLogService()
|