document-management/backend/dms/models/audit_log.py

"""追加写审计日志模型。"""

from __future__ import annotations

from datetime import datetime, timezone

from sqlalchemy import BigInteger, CheckConstraint, Index, JSON, String, Text, text
from sqlalchemy.dialects.mysql import DATETIME
from sqlalchemy.orm import Mapped, mapped_column

from dms.extensions import db


class AuditLog(db.Model):
    __tablename__ = "sys_audit_log"
    # action_type 不设 CHECK 白名单（0035 移除）：审计动作是持续扩展的枚举，
    # DB 约束导致新动作写入失败并连带业务操作 500，合法性由应用层字典约束；
    # 前端字典无匹配时回退显示英文原码。
    __table_args__ = (
        CheckConstraint(
            "target_type IS NULL OR target_type IN ("
            "'AUTH', 'CATEGORY', 'DOCUMENT', 'ATTACHMENT', "
            "'PERMISSION', 'ATTACHMENT_BINDING', 'ORGANIZATION', 'USER', "
            "'ROLE', 'IP_ENTRY', 'BACKUP'"
            ")",
            name="audit_target_type",
        ),
        CheckConstraint(
            "operation_result IN ('SUCCESS', 'FAILURE')",
            name="audit_operation_result",
        ),
        Index("ix_sys_audit_log_created_at", "created_at"),
        Index("ix_sys_audit_log_user_created", "user_id", "created_at"),
        Index(
            "ix_sys_audit_log_organization_created",
            "organization_id",
            "created_at",
        ),
        Index("ix_sys_audit_log_action_created", "action_type", "created_at"),
        Index(
            "ix_sys_audit_log_target_created",
            "target_type",
            "target_id",
            "created_at",
        ),
        Index(
            "ix_sys_audit_log_result_created",
            "operation_result",
            "created_at",
        ),
        Index("ix_sys_audit_log_request_id", "request_id"),
    )

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
    user_id: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
    username: Mapped[str | None] = mapped_column(String(64), nullable=True)
    real_name: Mapped[str | None] = mapped_column(String(64), nullable=True)
    organization_id: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
    organization_name: Mapped[str | None] = mapped_column(String(128), nullable=True)
    action_type: Mapped[str] = mapped_column(String(32), nullable=False)
    target_type: Mapped[str | None] = mapped_column(String(32), nullable=True)
    target_id: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
    target_name: Mapped[str | None] = mapped_column(String(255), nullable=True)
    operation_result: Mapped[str] = mapped_column(String(20), nullable=False)
    failure_reason: Mapped[str | None] = mapped_column(String(1000), nullable=True)
    client_ip: Mapped[str | None] = mapped_column(String(64), nullable=True)
    user_agent: Mapped[str | None] = mapped_column(String(500), nullable=True)
    request_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
    operation_detail: Mapped[dict[str, object] | None] = mapped_column(
        JSON,
        nullable=True,
    )
    created_at: Mapped[datetime] = mapped_column(
        DATETIME(fsp=3),
        nullable=False,
        # 应用侧 UTC 落库：server_default CURRENT_TIMESTAMP 为服务器本地时区，
        # 与全系统 UTC 口径（序列化追加 Z）不一致会导致前端显示快 8 小时
        default=lambda: datetime.now(timezone.utc).replace(tzinfo=None),
        server_default=text("CURRENT_TIMESTAMP(3)"),
    )