139 lines
4.0 KiB
Python
139 lines
4.0 KiB
Python
"""
|
|
操作日志服务
|
|
"""
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from fastapi import Request
|
|
from typing import Optional
|
|
import json
|
|
|
|
from app.models.log import OperationLog
|
|
from app.models.user import User
|
|
from app.core.enums import OperationType, ResourceType
|
|
|
|
|
|
class LogService:
|
|
"""操作日志服务"""
|
|
|
|
@staticmethod
|
|
async def log_operation(
|
|
db: AsyncSession,
|
|
operation_type: OperationType,
|
|
resource_type: ResourceType,
|
|
user: Optional[User] = None,
|
|
resource_id: Optional[int] = None,
|
|
detail: Optional[dict] = None,
|
|
request: Optional[Request] = None,
|
|
status: int = 1,
|
|
error_message: Optional[str] = None,
|
|
):
|
|
"""
|
|
记录操作日志
|
|
|
|
Args:
|
|
db: 数据库会话
|
|
operation_type: 操作类型
|
|
resource_type: 资源类型
|
|
user: 操作用户
|
|
resource_id: 资源ID
|
|
detail: 操作详情（字典格式）
|
|
request: FastAPI Request对象
|
|
status: 状态 0-失败 1-成功
|
|
error_message: 错误信息
|
|
"""
|
|
# 获取IP和User-Agent
|
|
ip_address = None
|
|
user_agent = None
|
|
if request:
|
|
ip_address = request.client.host if request.client else None
|
|
user_agent = request.headers.get("user-agent", "")
|
|
|
|
# 创建日志记录
|
|
log = OperationLog(
|
|
user_id=user.id if user else None,
|
|
username=user.username if user else "匿名",
|
|
operation_type=operation_type.value,
|
|
resource_type=resource_type.value,
|
|
resource_id=resource_id,
|
|
detail=json.dumps(detail, ensure_ascii=False) if detail else None,
|
|
ip_address=ip_address,
|
|
user_agent=user_agent,
|
|
status=status,
|
|
error_message=error_message,
|
|
)
|
|
|
|
db.add(log)
|
|
await db.commit()
|
|
|
|
@staticmethod
|
|
async def log_project_operation(
|
|
db: AsyncSession,
|
|
operation_type: OperationType,
|
|
project_id: int,
|
|
user: User,
|
|
detail: Optional[dict] = None,
|
|
request: Optional[Request] = None,
|
|
):
|
|
"""记录项目相关操作"""
|
|
await LogService.log_operation(
|
|
db=db,
|
|
operation_type=operation_type,
|
|
resource_type=ResourceType.PROJECT,
|
|
user=user,
|
|
resource_id=project_id,
|
|
detail=detail,
|
|
request=request,
|
|
)
|
|
|
|
@staticmethod
|
|
async def log_file_operation(
|
|
db: AsyncSession,
|
|
operation_type: OperationType,
|
|
project_id: int,
|
|
file_path: str,
|
|
user: User,
|
|
detail: Optional[dict] = None,
|
|
request: Optional[Request] = None,
|
|
):
|
|
"""记录文件相关操作"""
|
|
file_detail = {"project_id": project_id, "file_path": file_path}
|
|
if detail:
|
|
file_detail.update(detail)
|
|
|
|
await LogService.log_operation(
|
|
db=db,
|
|
operation_type=operation_type,
|
|
resource_type=ResourceType.FILE,
|
|
user=user,
|
|
resource_id=project_id, # 将project_id作为resource_id
|
|
detail=file_detail,
|
|
request=request,
|
|
)
|
|
|
|
@staticmethod
|
|
async def log_member_operation(
|
|
db: AsyncSession,
|
|
operation_type: OperationType,
|
|
project_id: int,
|
|
target_user_id: int,
|
|
user: User,
|
|
detail: Optional[dict] = None,
|
|
request: Optional[Request] = None,
|
|
):
|
|
"""记录成员相关操作"""
|
|
member_detail = {"project_id": project_id, "target_user_id": target_user_id}
|
|
if detail:
|
|
member_detail.update(detail)
|
|
|
|
await LogService.log_operation(
|
|
db=db,
|
|
operation_type=operation_type,
|
|
resource_type=ResourceType.MEMBER,
|
|
user=user,
|
|
resource_id=project_id,
|
|
detail=member_detail,
|
|
request=request,
|
|
)
|
|
|
|
|
|
log_service = LogService()
|