document-management/backend/migrations/versions/0031_user_sort_no.py

"""机构/用户/角色拖动排序（契约1.27）：sys_user 加排序号列、审计白名单扩展。

Revision ID: 0031_user_sort_no
Revises: 0030_role_edit_permission
Create Date: 2026-09-11

变更要点：
1. sys_user 新增 sort_no 整型排序号（默认 0；机构/角色已有同名列，仅用户缺），
   用户列表排序由 real_name 改为 sort_no,id，拖拽重排按步长10重编号；
2. 重建 sys_audit_log 的 audit_action_type CHECK 白名单（0029 的 42 项 + 3 = 45 项：
   REORDER_ORGANIZATION / REORDER_USER / REORDER_ROLE）；
3. 达梦兼容：约束名大写探测小写比较后 drop（同 0028/0029 做法）；
4. 升级幂等：加列先探测、CHECK drop-if-exists 后重建（与 Flyway V16 并行互不冲突）；
5. 降级拒绝已存在三个重排审计动作记录的库。
"""

from __future__ import annotations

from collections.abc import Sequence

import sqlalchemy as sa
from alembic import op

revision: str = "0031_user_sort_no"
down_revision: str | None = "0030_role_edit_permission"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None

# 0029 的 V10 全集（42 项）+ 本期新增动作
AUDIT_ACTIONS_V11 = (
    "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",
    "BACKUP_CREATE",
    "PREVIEW_CACHE_CLEAR",
    "FORCE_OFFLINE",
    "BACKUP_TEST_TARGET",
    "REORDER_ORGANIZATION",
    "REORDER_USER",
    "REORDER_ROLE",
)
AUDIT_ACTIONS_V10 = AUDIT_ACTIONS_V11[:-3]

ACTION_CONSTRAINT = "audit_action_type"

REORDER_ACTIONS = (
    "REORDER_ORGANIZATION",
    "REORDER_USER",
    "REORDER_ROLE",
)


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


def _existing_check_names() -> set[str]:
    # 达梦返回的约束名为大写，统一小写比较（同 0028/0029）。
    return {
        item["name"].lower()
        for item in sa.inspect(op.get_bind()).get_check_constraints("sys_audit_log")
    }


def _drop_checks(names: set[str]) -> None:
    if f"ck_sys_audit_log_{ACTION_CONSTRAINT}" in names:
        op.drop_constraint(ACTION_CONSTRAINT, "sys_audit_log", type_="check")


def _existing_user_columns() -> set[str]:
    inspector = sa.inspect(op.get_bind())
    return {
        column["name"].lower()
        for column in inspector.get_columns("sys_user")
    }


def _has_reorder_audit_records() -> bool:
    placeholders = ", ".join(f":action_{i}" for i in range(len(REORDER_ACTIONS)))
    params = {
        f"action_{i}": action for i, action in enumerate(REORDER_ACTIONS)
    }
    result = op.get_bind().execute(
        sa.text(
            f"SELECT COUNT(*) FROM sys_audit_log "
            f"WHERE action_type IN ({placeholders})"
        ),
        params,
    )
    return bool(result.scalar())


def upgrade() -> None:
    # 1. sys_user.sort_no 加列（先探测后执行，幂等）
    if "sort_no" not in _existing_user_columns():
        op.add_column(
            "sys_user",
            sa.Column(
                "sort_no",
                sa.Integer(),
                nullable=False,
                server_default=sa.text("0"),
            ),
        )

    # 2. 审计动作白名单重建（drop-if-exists + 重建）
    existing = _existing_check_names()
    _drop_checks(existing)
    op.create_check_constraint(
        ACTION_CONSTRAINT,
        "sys_audit_log",
        _check_sql(AUDIT_ACTIONS_V11),
    )


def downgrade() -> None:
    if _has_reorder_audit_records():
        raise RuntimeError("已存在机构/用户/角色重排审计记录，拒绝有损降级")

    if "sort_no" in _existing_user_columns():
        op.drop_column("sys_user", "sort_no")

    existing = _existing_check_names()
    _drop_checks(existing)
    op.create_check_constraint(
        ACTION_CONSTRAINT,
        "sys_audit_log",
        _check_sql(AUDIT_ACTIONS_V10),
    )