document-management/backend/migrations/versions/0021_remove_assigned_scope.py

"""移除"可访问部门"维度：ASSIGNED 矩阵行删除 + sys_role_dept 删表。

Revision ID: 0021_remove_assigned_scope
Revises: 0020_sub_plan_to_attachment
Create Date: 2026-08-31

变更要点：
1. 部门范围由 OWN/ASSIGNED/OTHER 收敛为 OWN/OTHER，"其他部门"涵盖
   所属部门子树之外的全部部门（含无归属文档）；
2. sys_role_permission 中全部 dept_scope='ASSIGNED' 行物理删除（含软删行），
   存量"可访问部门权限"数据按需求清除；
3. dept_scope CHECK 约束换名收敛：旧名 ck_sys_role_permission_role_permission_dept_scope
   DROP IF EXISTS，新名 ck_sys_role_permission_dept_scope_v2 ADD IF NOT EXISTS
   （按名幂等，与 Flyway V6 一致）；
4. sys_role_dept 角色可访问部门表 DROP IF EXISTS；
5. 降级：重建 sys_role_dept 空表、恢复旧 CHECK，并为存活角色补种
   ASSIGNED 默认全否行以还原 15 行矩阵不变量（原有勾选数据不可恢复）。
"""

from __future__ import annotations

from collections.abc import Sequence

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

revision: str = "0021_remove_assigned_scope"
down_revision: str | None = "0020_sub_plan_to_attachment"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None

_OLD_CHECK = "ck_sys_role_permission_role_permission_dept_scope"
_NEW_CHECK = "ck_sys_role_permission_dept_scope_v2"
_NEW_CHECK_DDL = (
    "ALTER TABLE sys_role_permission ADD CONSTRAINT " + _NEW_CHECK + " "
    "CHECK (dept_scope IN ('OWN', 'OTHER'))"
)
_OLD_CHECK_DDL = (
    "ALTER TABLE sys_role_permission ADD CONSTRAINT " + _OLD_CHECK + " "
    "CHECK (dept_scope IN ('OWN', 'ASSIGNED', 'OTHER'))"
)


def _constraint_exists(name: str) -> bool:
    bind = op.get_bind()
    return bool(
        bind.scalar(
            sa.text(
                "SELECT COUNT(*) FROM information_schema.TABLE_CONSTRAINTS "
                "WHERE TABLE_SCHEMA = DATABASE() "
                "AND TABLE_NAME = 'sys_role_permission' AND CONSTRAINT_NAME = :name"
            ),
            {"name": name},
        )
    )


def _table_exists(name: str) -> bool:
    bind = op.get_bind()
    return bool(
        bind.scalar(
            sa.text(
                "SELECT COUNT(*) FROM information_schema.TABLES "
                "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :name"
            ),
            {"name": name},
        )
    )


def _run(statement: str) -> None:
    op.get_bind().execute(sa.text(statement))


def _create_role_dept_table() -> None:
    op.create_table(
        "sys_role_dept",
        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.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",
        ),
        sa.PrimaryKeyConstraint("id"),
    )
    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"],
    )


def upgrade() -> None:
    # 1) 存量 ASSIGNED 权限行物理删除（含软删行；新 CHECK 前置条件）
    _run("DELETE FROM sys_role_permission WHERE dept_scope = 'ASSIGNED'")

    # 2) CHECK 约束收敛为 OWN/OTHER（按名幂等）
    if _constraint_exists(_OLD_CHECK):
        _run(f"ALTER TABLE sys_role_permission DROP CHECK {_OLD_CHECK}")
    if not _constraint_exists(_NEW_CHECK):
        _run(_NEW_CHECK_DDL)

    # 3) 角色可访问部门表下线
    _run("DROP TABLE IF EXISTS sys_role_dept")


def downgrade() -> None:
    # 1) 恢复旧 CHECK（先删新名，避免取值集合冲突）
    if _constraint_exists(_NEW_CHECK):
        _run(f"ALTER TABLE sys_role_permission DROP CHECK {_NEW_CHECK}")
    if not _constraint_exists(_OLD_CHECK):
        _run(_OLD_CHECK_DDL)

    # 2) 重建角色可访问部门空表（原数据已在升级时删除，不可恢复）
    if not _table_exists("sys_role_dept"):
        _create_role_dept_table()

    # 3) 为存活角色补种 ASSIGNED 默认全否行，还原"每角色 15 行"不变量
    _run(
        """
        INSERT INTO sys_role_permission
            (role_id, security_level, dept_scope, created_by, updated_by)
        SELECT r.id, lv.security_level, 'ASSIGNED', r.created_by, r.updated_by
        FROM sys_role r
        JOIN (
            SELECT 'PUBLIC' AS security_level
            UNION ALL SELECT 'INTERNAL'
            UNION ALL SELECT 'SECRET'
            UNION ALL SELECT 'CONFIDENTIAL'
            UNION ALL SELECT 'TOP_SECRET'
        ) lv
        WHERE r.is_deleted = 0
          AND NOT EXISTS (
              SELECT 1 FROM sys_role_permission p
              WHERE p.role_id = r.id
                AND p.security_level = lv.security_level
                AND p.dept_scope = 'ASSIGNED'
          )
        """
    )