107 lines
3.8 KiB
Python
107 lines
3.8 KiB
Python
#!/usr/bin/env python
|
||
# -*- coding: utf-8 -*-
|
||
|
||
"""
|
||
用户操作模型模块
|
||
包含用户操作记录相关的数据模型
|
||
"""
|
||
|
||
import enum
|
||
from sqlalchemy import Column, Integer, String, Text, DateTime, Enum, ForeignKey, JSON, Boolean
|
||
from sqlalchemy.orm import relationship
|
||
from data.models.base import BaseModel
|
||
|
||
class OperationType(enum.Enum):
|
||
"""
|
||
操作类型枚举
|
||
"""
|
||
CREATE = 'create' # 创建
|
||
UPDATE = 'update' # 更新
|
||
DELETE = 'delete' # 删除
|
||
ENABLE = 'enable' # 启用
|
||
DISABLE = 'disable' # 禁用
|
||
IMPORT = 'import' # 导入
|
||
EXPORT = 'export' # 导出
|
||
EXECUTE = 'execute' # 执行
|
||
CANCEL = 'cancel' # 取消
|
||
PAUSE = 'pause' # 暂停
|
||
RESUME = 'resume' # 恢复
|
||
|
||
class TargetType(enum.Enum):
|
||
"""
|
||
操作目标类型枚举
|
||
"""
|
||
TASK = 'task' # 任务
|
||
TASK_VERSION = 'task_version' # 任务版本
|
||
SUBTASK = 'subtask' # 子任务
|
||
COMPONENT = 'component' # 组件
|
||
|
||
class UserOperation(BaseModel):
|
||
"""
|
||
用户操作模型
|
||
记录用户对系统的操作
|
||
"""
|
||
__tablename__ = 'user_operations'
|
||
|
||
user_id = Column(String(100), nullable=False, comment='用户ID')
|
||
user_name = Column(String(100), nullable=True, comment='用户名称')
|
||
operation_type = Column(Enum(OperationType), nullable=False, comment='操作类型')
|
||
target_type = Column(Enum(TargetType), nullable=False, comment='目标类型')
|
||
target_id = Column(Integer, nullable=False, comment='目标ID')
|
||
target_name = Column(String(100), nullable=True, comment='目标名称')
|
||
operation_time = Column(DateTime, nullable=False, comment='操作时间')
|
||
operation_ip = Column(String(50), nullable=True, comment='操作IP')
|
||
operation_details = Column(JSON, nullable=True, comment='操作详情（JSON格式）')
|
||
operation_result = Column(Boolean, nullable=False, default=True, comment='操作结果（成功/失败）')
|
||
error_message = Column(Text, nullable=True, comment='错误信息')
|
||
|
||
def __repr__(self):
|
||
return f"<UserOperation(id={self.id}, user_id='{self.user_id}', operation_type='{self.operation_type}', target_type='{self.target_type}', target_id={self.target_id})>"
|
||
|
||
@classmethod
|
||
def get_by_user(cls, user_id, page=1, per_page=20):
|
||
"""
|
||
获取用户的操作记录
|
||
"""
|
||
return cls.query.filter(
|
||
cls.user_id == user_id,
|
||
cls.is_deleted == False
|
||
).order_by(cls.operation_time.desc()).paginate(page=page, per_page=per_page)
|
||
|
||
@classmethod
|
||
def get_by_target(cls, target_type, target_id, page=1, per_page=20):
|
||
"""
|
||
获取目标对象的操作记录
|
||
"""
|
||
return cls.query.filter(
|
||
cls.target_type == target_type,
|
||
cls.target_id == target_id,
|
||
cls.is_deleted == False
|
||
).order_by(cls.operation_time.desc()).paginate(page=page, per_page=per_page)
|
||
|
||
@classmethod
|
||
def log_operation(cls, user_id, user_name, operation_type, target_type, target_id, target_name=None,
|
||
operation_details=None, operation_ip=None, operation_result=True, error_message=None):
|
||
"""
|
||
记录用户操作
|
||
"""
|
||
import datetime
|
||
from config.database import db_session
|
||
|
||
operation = cls(
|
||
user_id=user_id,
|
||
user_name=user_name,
|
||
operation_type=operation_type,
|
||
target_type=target_type,
|
||
target_id=target_id,
|
||
target_name=target_name,
|
||
operation_time=datetime.datetime.now(),
|
||
operation_ip=operation_ip,
|
||
operation_details=operation_details,
|
||
operation_result=operation_result,
|
||
error_message=error_message
|
||
)
|
||
|
||
db_session.add(operation)
|
||
db_session.commit()
|
||
return operation |