document-management/backend/migrations/versions/0025_move_audit_actions.py

"""审计动作扩展：sys_audit_log 的 action_type CHECK 加入 MOVE_DOCUMENT / MOVE_ATTACHMENT。

Revision ID: 0025_move_audit_actions
Revises: 0024_manage_permission_matrix
Create Date: 2026-09-03

变更要点：
1. 跨页调整顺序功能（契约1.18）新增审计动作 MOVE_DOCUMENT / MOVE_ATTACHMENT
   （0015 的 AUDIT_ACTIONS_V7 全集 + 2 = 38 项，target_type 白名单不变）；
2. 重建 sys_audit_log 的 audit_action_type CHECK 约束；
3. 达梦兼容：约束名在 DM 下渲染为大写（CK_SYS_AUDIT_LOG_AUDIT_ACTION_TYPE），
   探测按小写比较后 drop 原始逻辑名（naming convention 渲染最终名）；
4. 升级幂等：drop-if-exists + 重建（与 Flyway V10 并行执行结果一致，可重复跑）；
   降级拒绝已存在 MOVE 审计记录的库。
"""

from __future__ import annotations

from collections.abc import Sequence

import sqlalchemy as sa
from alembic import op

revision: str = "0025_move_audit_actions"
down_revision: str | None = "0024_manage_permission_matrix"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None


# 0015 的全集 + 本期新增动作
AUDIT_ACTIONS_V8 = (
    "LOGIN",
    "LOGOUT",
    "VIEW_DOCUMENT",
    "DOWNLOAD_DOCUMENT",
    "UPLOAD_DOCUMENT",
    "BATCH_IMPORT",
    "EDIT_DOCUMENT",
    "DELETE_DOCUMENT",
    "CREATE_CATEGORY",
    "EDIT_CATEGORY",
    "DELETE_CATEGORY",
    "CHANGE_PERMISSION",
    "BIND_ATTACHMENT",
    "UNBIND_ATTACHMENT",
    "RESTORE_DOCUMENT",
    "CREATE_ORGANIZATION",
    "EDIT_ORGANIZATION",
    "DELETE_ORGANIZATION",
    "CREATE_USER",
    "EDIT_USER",
    "DELETE_USER",
    "RESET_USER_PASSWORD",
    "REORDER_DOCUMENT",
    "REORDER_ATTACHMENT",
    "REORDER_CATEGORY",
    "CREATE_ROLE",
    "EDIT_ROLE",
    "DELETE_ROLE",
    "CHANGE_ROLE_PERMISSION",
    "CHANGE_ROLE_USER",
    "DOCUMENT_PASSWORD_SET",
    "DOCUMENT_PASSWORD_REMOVE",
    "CREATE_IP_ENTRY",
    "EDIT_IP_ENTRY",
    "DELETE_IP_ENTRY",
    "ASSIGN_ORG_IP",
    "MOVE_DOCUMENT",
    "MOVE_ATTACHMENT",
)
AUDIT_ACTIONS_V7 = AUDIT_ACTIONS_V8[:-2]

ACTION_CONSTRAINT = "audit_action_type"


def _check_sql(values: tuple[str, ...]) -> str:
    # 安全性：values 为本文件内硬编码的枚举常量，非用户输入，无注入风险
    inline = ", ".join(f"'{v}'" for v in values)
    return f"action_type IN ({inline})"


def _drop_action_check() -> None:
    # op.drop_constraint 按 metadata.naming_convention 前缀化约束名，
    # 须传原始约束名渲染最终名；达梦返回的约束名为大写，统一小写比较。
    existing = {
        item["name"].lower()
        for item in sa.inspect(op.get_bind()).get_check_constraints("sys_audit_log")
    }
    if f"ck_sys_audit_log_{ACTION_CONSTRAINT}" in existing:
        op.drop_constraint(ACTION_CONSTRAINT, "sys_audit_log", type_="check")


def upgrade() -> None:
    _drop_action_check()
    op.create_check_constraint(
        ACTION_CONSTRAINT,
        "sys_audit_log",
        _check_sql(AUDIT_ACTIONS_V8),
    )


def downgrade() -> None:
    move_rows = op.get_bind().scalar(
        sa.text(
            "SELECT COUNT(*) FROM sys_audit_log "
            "WHERE action_type IN ('MOVE_DOCUMENT', 'MOVE_ATTACHMENT')"
        )
    )
    if move_rows:
        raise RuntimeError(
            "0025已存在跨页移动审计记录，拒绝有损降级；"
            "请先备份并按数据治理流程处理"
        )
    _drop_action_check()
    op.create_check_constraint(
        ACTION_CONSTRAINT,
        "sys_audit_log",
        _check_sql(AUDIT_ACTIONS_V7),
    )