document-management/backend/migrations/versions/0004_organization_user_maintenance.py

"""机构与用户维护：复合唯一索引 + 审计CHECK扩展。

Revision ID: 0004_organization_user_maintenance
Revises: 0003_q3_business_alignment_and_content_search
Create Date: 2026-08-12

变更要点：
1. sys_organization / doc_category 引入 active_marker 虚拟生成列，
   将 org_code / category_code 的唯一性约束改造为"仅活跃行唯一"，
   支持软删除后重建同名条目，与既有 attachment_binding / permission 表保持一致。
2. sys_audit_log 的 action_type / target_type CHECK 约束扩展，
   覆盖机构与用户维护相关的新枚举值。
"""

from __future__ import annotations

from collections.abc import Sequence

from alembic import op
import sqlalchemy as sa

revision: str = "0004_organization_user_maintenance"
down_revision: str | None = "0003_q3_business_alignment_and_content_search"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None


# 既有审计动作（与 0002 对齐）+ 本期新增动作
AUDIT_ACTIONS_V2 = (
    "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",
)
AUDIT_ACTIONS_V1 = AUDIT_ACTIONS_V2[:15]
AUDIT_TARGETS_V2 = (
    "AUTH",
    "CATEGORY",
    "DOCUMENT",
    "ATTACHMENT",
    "PERMISSION",
    "ATTACHMENT_BINDING",
    "ORGANIZATION",
    "USER",
)
AUDIT_TARGETS_V1 = AUDIT_TARGETS_V2[:6]

ACTION_CONSTRAINT = "audit_action_type"
TARGET_CONSTRAINT = "audit_target_type"


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


def _target_check_sql(values: tuple[str, ...]) -> str:
    # 安全性：values 为本文件内硬编码的枚举常量，非用户输入，无注入风险
    inline = ", ".join(f"'{v}'" for v in values)
    return f"(target_type IS NULL OR target_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 _existing_index_names(table: str) -> set[str]:
    inspector = sa.inspect(op.get_bind())
    return {item["name"] for item in inspector.get_indexes(table)}


def _existing_columns(table: str) -> set[str]:
    inspector = sa.inspect(op.get_bind())
    return {item["name"] for item in inspector.get_columns(table)}


def _drop_audit_checks(existing: set[str]) -> None:
    # op.drop_constraint 会按 metadata.naming_convention 再次前缀化约束名
    # （"ck_%(table_name)s_%(constraint_name)s"），因此必须传入原始约束名，
    # 让 SQLAlchemy 渲染出最终名；数据库实际存的也是渲染后的最终名。
    # 这里用最终名判断是否实际存在，避免被二次前缀化导致MySQL 3821。
    for raw_name, final_name in (
        (ACTION_CONSTRAINT, f"ck_sys_audit_log_{ACTION_CONSTRAINT}"),
        (TARGET_CONSTRAINT, f"ck_sys_audit_log_{TARGET_CONSTRAINT}"),
    ):
        if final_name in existing:
            op.drop_constraint(raw_name, "sys_audit_log", type_="check")


# 复合唯一索引迁移逻辑 --------------------------------------------------

def _migrate_unique_to_active(
    table: str,
    code_column: str,
    old_unique_index: str,
    new_unique_index: str,
) -> None:
    """将单列唯一索引升级为 (code, active_marker) 复合唯一索引。"""
    columns = _existing_columns(table)
    indexes = _existing_index_names(table)

    if "active_marker" not in columns:
        op.execute(
            sa.text(
                f"ALTER TABLE `{table}` ADD COLUMN `active_marker` "
                f"INT GENERATED ALWAYS AS (IF(is_deleted = 0, 1, NULL)) VIRTUAL"
            )
        )

    if old_unique_index in indexes:
        op.drop_index(old_unique_index, table_name=table)

    if new_unique_index not in indexes:
        op.create_index(
            new_unique_index,
            table,
            [code_column, "active_marker"],
            unique=True,
        )


def _revert_unique_to_simple(
    table: str,
    code_column: str,
    old_unique_index: str,
    new_unique_index: str,
) -> None:
    indexes = _existing_index_names(table)
    if new_unique_index in indexes:
        op.drop_index(new_unique_index, table_name=table)
    if old_unique_index not in indexes:
        op.create_index(
            old_unique_index,
            table,
            [code_column],
            unique=True,
        )
    columns = _existing_columns(table)
    if "active_marker" in columns:
        op.drop_column(table, "active_marker")


# 升级 / 降级 ----------------------------------------------------------

def upgrade() -> None:
    # 1) 复合唯一索引（机构 + 分类同步改造）
    _migrate_unique_to_active(
        table="sys_organization",
        code_column="org_code",
        old_unique_index="uq_sys_organization_org_code",
        new_unique_index="uq_sys_organization_org_code_active",
    )
    _migrate_unique_to_active(
        table="doc_category",
        code_column="category_code",
        old_unique_index="uq_doc_category_category_code",
        new_unique_index="uq_doc_category_category_code_active",
    )

    # 2) 审计 CHECK 扩展
    _drop_audit_checks(_existing_constraint_names("sys_audit_log"))
    op.create_check_constraint(
        ACTION_CONSTRAINT,
        "sys_audit_log",
        _check_sql(AUDIT_ACTIONS_V2),
    )
    op.create_check_constraint(
        TARGET_CONSTRAINT,
        "sys_audit_log",
        _target_check_sql(AUDIT_TARGETS_V2),
    )


def downgrade() -> None:
    bind = op.get_bind()
    new_action_rows = bind.scalar(
        sa.text(
            "SELECT COUNT(*) FROM sys_audit_log "
            "WHERE action_type IN ("
            "'CREATE_ORGANIZATION','EDIT_ORGANIZATION','DELETE_ORGANIZATION',"
            "'CREATE_USER','EDIT_USER','DELETE_USER','RESET_USER_PASSWORD')"
        )
    )
    new_target_rows = bind.scalar(
        sa.text(
            "SELECT COUNT(*) FROM sys_audit_log "
            "WHERE target_type IN ('ORGANIZATION','USER')"
        )
    )
    if new_action_rows or new_target_rows:
        raise RuntimeError(
            "0004已产生机构/用户维护相关审计记录，拒绝有损降级；"
            "请先备份并按数据治理流程处理"
        )

    _drop_audit_checks(_existing_constraint_names("sys_audit_log"))
    op.create_check_constraint(
        ACTION_CONSTRAINT,
        "sys_audit_log",
        _check_sql(AUDIT_ACTIONS_V1),
    )
    op.create_check_constraint(
        TARGET_CONSTRAINT,
        "sys_audit_log",
        _target_check_sql(AUDIT_TARGETS_V1),
    )

    _revert_unique_to_simple(
        table="sys_organization",
        code_column="org_code",
        old_unique_index="uq_sys_organization_org_code",
        new_unique_index="uq_sys_organization_org_code_active",
    )
    _revert_unique_to_simple(
        table="doc_category",
        code_column="category_code",
        old_unique_index="uq_doc_category_category_code",
        new_unique_index="uq_doc_category_category_code_active",
    )