document-management/backend/migrations/versions/0009_reorder_audit_action.py

"""方案拖动排序：审计动作CHECK扩展。

Revision ID: 0009_reorder_audit_action
Revises: 0008_document_sort_no
Create Date: 2026-08-18

变更要点：
1. sys_audit_log 的 action_type CHECK 约束扩展，覆盖拖动排序新增的
   REORDER_DOCUMENT / REORDER_ATTACHMENT 两个审计动作。
2. 降级保护：已产生排序审计记录时拒绝有损降级。
"""

from __future__ import annotations

from collections.abc import Sequence

from alembic import op
import sqlalchemy as sa

revision: str = "0009_reorder_audit_action"
down_revision: str | None = "0008_document_sort_no"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None


# 既有审计动作（与 0004 对齐）+ 本期新增动作
AUDIT_ACTIONS_V3 = (
    "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",
)
AUDIT_ACTIONS_V2 = AUDIT_ACTIONS_V3[:22]

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 _existing_constraint_names(table: str) -> set[str]:
    inspector = sa.inspect(op.get_bind())
    return {item["name"] for item in inspector.get_check_constraints(table)}


def _drop_action_check(existing: set[str]) -> None:
    # op.drop_constraint 会按 metadata.naming_convention 再次前缀化约束名
    # （"ck_%(table_name)s_%(constraint_name)s"），因此必须传入原始约束名，
    # 让 SQLAlchemy 渲染出最终名；数据库实际存的也是渲染后的最终名。
    final_name = f"ck_sys_audit_log_{ACTION_CONSTRAINT}"
    if final_name in existing:
        op.drop_constraint(ACTION_CONSTRAINT, "sys_audit_log", type_="check")


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


def downgrade() -> None:
    bind = op.get_bind()
    reorder_rows = bind.scalar(
        sa.text(
            "SELECT COUNT(*) FROM sys_audit_log "
            "WHERE action_type IN ('REORDER_DOCUMENT','REORDER_ATTACHMENT')"
        )
    )
    if reorder_rows:
        raise RuntimeError(
            "0009已产生拖动排序相关审计记录，拒绝有损降级；"
            "请先备份并按数据治理流程处理"
        )

    _drop_action_check(_existing_constraint_names("sys_audit_log"))
    op.create_check_constraint(
        ACTION_CONSTRAINT,
        "sys_audit_log",
        _check_sql(AUDIT_ACTIONS_V2),
    )