duxingchen
3286a11bc7
- 复制来源: /home/yueli/track @ 192c8ee (feature/ai-audit-update) - 组织隔离目标: LICA - 端口规划: 前端 8030 / 后端 8031 / 数据库 8032 - 已排除 deploy.sh、deploy_full.sh、docker-compose.prod.yml（IRIS 生产发布脚本） - 已排除工作区未提交改动，取干净的 192c8ee 状态
84 lines
3.8 KiB
Python
84 lines
3.8 KiB
Python
"""操作审计日志模型
|
||
|
||
设计参考 MOM(KCGL) 的 audit_logs，但按 Track 的技术栈与诉求做了取舍：
|
||
|
||
- 主键用 UUID（与库内其它表一致），而非 MOM 的自增 int。
|
||
- 增加 request_id：与 core/logging.py 的结构化日志打通 —— 凭一个 ID 就能把
|
||
「接口访问日志」和「审计记录」对上，排障时不用再猜。MOM 无此字段。
|
||
- 保留 module / action / target_* 的业务语义，使审计能按业务维度检索，
|
||
而不是只能按时间翻。
|
||
- 绝不记录请求体：登录等接口 body 含明文密码，一旦落库就成了长期泄露面。
|
||
|
||
与既有 task_logs 的分工：task_logs 是「任务流转轨迹」（有 task_id 非空约束，
|
||
只能挂在任务上，供流转树渲染）；本表是「操作审计」，覆盖登录、导出、
|
||
产品增删改、权限变更等与单个任务无关的动作，且额外记录来源 IP / UA / 耗时结果。
|
||
"""
|
||
import uuid
|
||
from datetime import datetime
|
||
|
||
from sqlalchemy import DateTime, Integer, String, Text
|
||
from sqlalchemy.dialects.postgresql import JSONB, UUID
|
||
from sqlalchemy.orm import Mapped, mapped_column
|
||
|
||
from app.models.base import Base
|
||
from app.core.time_utils import get_beijing_time
|
||
|
||
|
||
class AuditLog(Base):
|
||
__tablename__ = "audit_logs"
|
||
|
||
id: Mapped[uuid.UUID] = mapped_column(
|
||
UUID(as_uuid=True), primary_key=True, default=uuid.uuid4,
|
||
)
|
||
|
||
# ---- 操作人（逻辑外键 → MOM sys_user，仅存账号，无物理约束）----
|
||
user_id: Mapped[str | None] = mapped_column(
|
||
String(64), nullable=True, index=True, comment="操作人账号(逻辑外键→MOM)",
|
||
)
|
||
display_name: Mapped[str | None] = mapped_column(
|
||
String(100), nullable=True, comment="操作人显示名",
|
||
)
|
||
role: Mapped[str | None] = mapped_column(
|
||
String(50), nullable=True, comment="操作时角色快照",
|
||
)
|
||
|
||
# ---- 业务语义 ----
|
||
action: Mapped[str] = mapped_column(
|
||
String(50), nullable=False, index=True, comment="动作: create/update/delete/export/login/...",
|
||
)
|
||
module: Mapped[str] = mapped_column(
|
||
String(50), nullable=False, index=True, comment="业务模块: product/task/order/auth/print/...",
|
||
)
|
||
target_type: Mapped[str | None] = mapped_column(
|
||
String(50), nullable=True, comment="目标类型(表名或实体名)",
|
||
)
|
||
target_id: Mapped[str | None] = mapped_column(
|
||
String(100), nullable=True, index=True, comment="目标ID",
|
||
)
|
||
target_name: Mapped[str | None] = mapped_column(
|
||
String(200), nullable=True, comment="目标显示名(如产品身份证/工单号)",
|
||
)
|
||
details: Mapped[dict | None] = mapped_column(
|
||
JSONB, nullable=True, comment="变更详情 {old:{}, new:{}}；禁止写入密码等敏感字段",
|
||
)
|
||
|
||
# ---- 请求上下文（由中间件自动填充）----
|
||
ip_address: Mapped[str | None] = mapped_column(String(50), nullable=True, comment="来源IP")
|
||
user_agent: Mapped[str | None] = mapped_column(String(500), nullable=True, comment="浏览器UA")
|
||
method: Mapped[str | None] = mapped_column(String(10), nullable=True, comment="HTTP方法")
|
||
url: Mapped[str | None] = mapped_column(String(500), nullable=True, comment="请求路径")
|
||
status_code: Mapped[int | None] = mapped_column(Integer, nullable=True, comment="响应状态码")
|
||
error_message: Mapped[str | None] = mapped_column(Text, nullable=True, comment="错误信息(如有)")
|
||
|
||
# ---- 与结构化日志对账用 ----
|
||
request_id: Mapped[str | None] = mapped_column(
|
||
String(64), nullable=True, index=True, comment="关联 core/logging 的 request_id",
|
||
)
|
||
|
||
created_at: Mapped[datetime] = mapped_column(
|
||
DateTime(timezone=True), default=get_beijing_time, index=True, comment="操作时间",
|
||
)
|
||
|
||
def __repr__(self) -> str:
|
||
return f"<AuditLog {self.action} {self.module} by {self.user_id}>"
|