document-management/backend/migrations/versions/0012_role_management.py

"""角色管理与权限矩阵：三张新表 + 用户角色外键 + 审计CHECK扩展。

Revision ID: 0012_role_management
Revises: 0011_reorder_category_audit_action
Create Date: 2026-08-20

变更要点：
1. 新建 sys_role / sys_role_dept / sys_role_permission 三张表：
   - sys_role 含 default_marker 虚拟生成列 + 唯一约束，保证全局至多一个默认角色；
   - sys_role_dept 角色可访问部门列表（矩阵 ASSIGNED 维度判定依据）；
   - sys_role_permission 权限矩阵（密级 × 部门范围 × 7 操作布尔列）。
2. sys_user 新增 role_id 可空外键（NULL → 套用默认角色）。
3. sys_audit_log 的 action_type / target_type CHECK 约束扩展，
   覆盖角色管理相关的新枚举值。
4. 降级保护：已产生角色数据/审计记录时拒绝有损降级。
"""

from __future__ import annotations

from collections.abc import Sequence

from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import mysql

revision: str = "0012_role_management"
down_revision: str | None = "0011_reorder_category_audit_action"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None


# 既有审计动作（与 0011 对齐）+ 本期新增动作
AUDIT_ACTIONS_V5 = (
    "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",
)
AUDIT_ACTIONS_V4 = AUDIT_ACTIONS_V5[:-5]
AUDIT_TARGETS_V3 = (
    "AUTH",
    "CATEGORY",
    "DOCUMENT",
    "ATTACHMENT",
    "PERMISSION",
    "ATTACHMENT_BINDING",
    "ORGANIZATION",
    "USER",
    "ROLE",
)
AUDIT_TARGETS_V2 = AUDIT_TARGETS_V3[:-1]

ACTION_CONSTRAINT = "audit_action_type"
TARGET_CONSTRAINT = "audit_target_type"


def _business_columns() -> list[sa.Column]:
    return [
        sa.Column("id", sa.BigInteger(), autoincrement=True, nullable=False),
        sa.Column("created_by", sa.BigInteger(), nullable=True),
        sa.Column("updated_by", sa.BigInteger(), nullable=True),
        sa.Column(
            "created_at",
            mysql.DATETIME(fsp=3),
            server_default=sa.text("CURRENT_TIMESTAMP(3)"),
            nullable=False,
        ),
        sa.Column(
            "updated_at",
            mysql.DATETIME(fsp=3),
            server_default=sa.text("CURRENT_TIMESTAMP(3)"),
            nullable=False,
        ),
        sa.Column("deleted_at", mysql.DATETIME(fsp=3), nullable=True),
        sa.Column(
            "is_deleted",
            sa.Boolean(),
            server_default=sa.text("0"),
            nullable=False,
        ),
        sa.Column(
            "row_version",
            sa.Integer(),
            server_default=sa.text("0"),
            nullable=False,
        ),
        sa.PrimaryKeyConstraint("id"),
    ]


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 _drop_audit_checks(existing: set[str]) -> None:
    # op.drop_constraint 会按 metadata.naming_convention 再次前缀化约束名
    # （"ck_%(table_name)s_%(constraint_name)s"），因此必须传入原始约束名，
    # 让 SQLAlchemy 渲染出最终名；数据库实际存的也是渲染后的最终名。
    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 upgrade() -> None:
    # 1) 角色本体
    op.create_table(
        "sys_role",
        *_business_columns(),
        sa.Column("role_name", sa.String(length=64), nullable=False),
        sa.Column(
            "sort_no",
            sa.Integer(),
            server_default=sa.text("0"),
            nullable=False,
        ),
        sa.Column(
            "status",
            sa.String(length=20),
            server_default=sa.text("'ENABLED'"),
            nullable=False,
        ),
        sa.Column("remark", sa.String(length=255), nullable=True),
        sa.Column(
            "is_default",
            sa.Boolean(),
            server_default=sa.text("0"),
            nullable=False,
        ),
        sa.Column(
            "active_marker",
            sa.Integer(),
            sa.Computed("IF(is_deleted = 0, 1, NULL)", persisted=False),
            nullable=True,
        ),
        sa.Column(
            "default_marker",
            sa.Integer(),
            sa.Computed(
                "IF(is_default = 1 AND is_deleted = 0, 1, NULL)",
                persisted=False,
            ),
            nullable=True,
        ),
        sa.CheckConstraint(
            "status IN ('ENABLED', 'DISABLED')",
            name="role_status",
        ),
        sa.UniqueConstraint(
            "role_name",
            "active_marker",
            name="uq_sys_role_role_name_active",
        ),
        sa.UniqueConstraint("default_marker", name="uq_sys_role_default_active"),
    )
    op.create_index(
        "ix_sys_role_status_active",
        "sys_role",
        ["status", "is_deleted"],
    )

    # 2) 角色可访问部门列表
    op.create_table(
        "sys_role_dept",
        *_business_columns(),
        sa.Column("role_id", sa.BigInteger(), nullable=False),
        sa.Column("organization_id", sa.BigInteger(), nullable=False),
        sa.Column(
            "active_marker",
            sa.Integer(),
            sa.Computed("IF(is_deleted = 0, 1, NULL)", persisted=False),
            nullable=True,
        ),
        sa.ForeignKeyConstraint(
            ["role_id"],
            ["sys_role.id"],
            name="fk_sys_role_dept_role_id_sys_role",
            ondelete="RESTRICT",
        ),
        sa.ForeignKeyConstraint(
            ["organization_id"],
            ["sys_organization.id"],
            name="fk_sys_role_dept_organization_id_sys_organization",
            ondelete="RESTRICT",
        ),
        sa.UniqueConstraint(
            "role_id",
            "organization_id",
            "active_marker",
            name="uq_sys_role_dept_active",
        ),
    )
    op.create_index(
        "ix_sys_role_dept_role_active",
        "sys_role_dept",
        ["role_id", "is_deleted"],
    )
    op.create_index(
        "ix_sys_role_dept_org_active",
        "sys_role_dept",
        ["organization_id", "is_deleted"],
    )

    # 3) 角色权限矩阵（每角色固定 15 行：5 密级 × 3 部门范围）
    op.create_table(
        "sys_role_permission",
        *_business_columns(),
        sa.Column("role_id", sa.BigInteger(), nullable=False),
        sa.Column("security_level", sa.String(length=32), nullable=False),
        sa.Column("dept_scope", sa.String(length=32), nullable=False),
        sa.Column(
            "can_view",
            sa.Boolean(),
            server_default=sa.text("0"),
            nullable=False,
        ),
        sa.Column(
            "can_download",
            sa.Boolean(),
            server_default=sa.text("0"),
            nullable=False,
        ),
        sa.Column(
            "can_edit",
            sa.Boolean(),
            server_default=sa.text("0"),
            nullable=False,
        ),
        sa.Column(
            "can_delete",
            sa.Boolean(),
            server_default=sa.text("0"),
            nullable=False,
        ),
        sa.Column(
            "can_config_permission",
            sa.Boolean(),
            server_default=sa.text("0"),
            nullable=False,
        ),
        sa.Column(
            "can_bind_attachment",
            sa.Boolean(),
            server_default=sa.text("0"),
            nullable=False,
        ),
        sa.Column(
            "can_unbind_attachment",
            sa.Boolean(),
            server_default=sa.text("0"),
            nullable=False,
        ),
        sa.Column(
            "active_marker",
            sa.Integer(),
            sa.Computed("IF(is_deleted = 0, 1, NULL)", persisted=False),
            nullable=True,
        ),
        sa.CheckConstraint(
            "security_level IN "
            "('PUBLIC', 'INTERNAL', 'SECRET', 'CONFIDENTIAL', 'TOP_SECRET')",
            name="role_permission_security_level",
        ),
        sa.CheckConstraint(
            "dept_scope IN ('OWN', 'ASSIGNED', 'OTHER')",
            name="role_permission_dept_scope",
        ),
        sa.ForeignKeyConstraint(
            ["role_id"],
            ["sys_role.id"],
            name="fk_sys_role_permission_role_id_sys_role",
            ondelete="RESTRICT",
        ),
        sa.UniqueConstraint(
            "role_id",
            "security_level",
            "dept_scope",
            "active_marker",
            name="uq_sys_role_permission_active",
        ),
    )
    op.create_index(
        "ix_sys_role_permission_role_active",
        "sys_role_permission",
        ["role_id", "is_deleted"],
    )

    # 4) 用户挂角色（NULL → 默认角色）
    op.add_column("sys_user", sa.Column("role_id", sa.BigInteger(), nullable=True))
    op.create_foreign_key(
        "fk_sys_user_role_id_sys_role",
        "sys_user",
        "sys_role",
        ["role_id"],
        ["id"],
        ondelete="RESTRICT",
    )
    op.create_index(
        "ix_sys_user_role_active",
        "sys_user",
        ["role_id", "is_deleted"],
    )

    # 5) 审计 CHECK 扩展
    _drop_audit_checks(_existing_constraint_names("sys_audit_log"))
    op.create_check_constraint(
        ACTION_CONSTRAINT,
        "sys_audit_log",
        _check_sql(AUDIT_ACTIONS_V5),
    )
    op.create_check_constraint(
        TARGET_CONSTRAINT,
        "sys_audit_log",
        _target_check_sql(AUDIT_TARGETS_V3),
    )


def downgrade() -> None:
    bind = op.get_bind()
    role_rows = bind.scalar(sa.text("SELECT COUNT(*) FROM sys_role"))
    user_role_rows = bind.scalar(
        sa.text("SELECT COUNT(*) FROM sys_user WHERE role_id IS NOT NULL")
    )
    audit_rows = bind.scalar(
        sa.text(
            "SELECT COUNT(*) FROM sys_audit_log "
            "WHERE action_type IN ("
            "'CREATE_ROLE','EDIT_ROLE','DELETE_ROLE',"
            "'CHANGE_ROLE_PERMISSION','CHANGE_ROLE_USER') "
            "OR target_type = 'ROLE'"
        )
    )
    if role_rows or user_role_rows or audit_rows:
        raise RuntimeError(
            "0012已产生角色管理相关数据或审计记录，拒绝有损降级；"
            "请先备份并按数据治理流程处理"
        )

    _drop_audit_checks(_existing_constraint_names("sys_audit_log"))
    op.create_check_constraint(
        ACTION_CONSTRAINT,
        "sys_audit_log",
        _check_sql(AUDIT_ACTIONS_V4),
    )
    op.create_check_constraint(
        TARGET_CONSTRAINT,
        "sys_audit_log",
        _target_check_sql(AUDIT_TARGETS_V2),
    )

    op.drop_index("ix_sys_user_role_active", table_name="sys_user")
    op.drop_constraint("fk_sys_user_role_id_sys_role", "sys_user", type_="foreignkey")
    op.drop_column("sys_user", "role_id")

    op.drop_table("sys_role_permission")
    op.drop_table("sys_role_dept")
    op.drop_table("sys_role")