document-management/backend/migrations/versions/0024_manage_permission_matrix.py

"""管理权限矩阵化：sys_role 增加管理权限位，sys_user 移除旧管理权限列。

Revision ID: 0024_manage_permission_matrix
Revises: 0023_admin_permissions
Create Date: 2026-09-03

变更要点：
1. sys_role 增加 manage_category_edit / manage_org_edit / manage_user_edit
   TINYINT(1) NOT NULL DEFAULT 0 与 managed_org_id BIGINT NULL：
   管理权限改挂角色，判定读"生效角色"（用户显式角色，未挂角色回退默认角色）；
2. 存量迁移：把活跃 ADMIN 用户旧有的 (三权限位, 锚点) 组合收敛到生效角色——
   生效角色仅被一组占用时直接置位；被多组占用时克隆角色（复制权限矩阵 10 行，
   命名追加 -M{n} 后缀）并回填该组用户 role_id；
3. sys_user 删除 perm_category_edit / perm_org_edit / perm_user_edit /
   managed_org_id 四列（判定改读生效角色，用户级配置下线）；
4. 升级幂等：加列先探测；数据迁移以 sys_user.perm_category_edit 存在为前提，
   列已删（已迁移）则整体跳过；中断重跑可续（已置位/已克隆的组自动跳过）。
"""

from __future__ import annotations

from collections.abc import Sequence

import sqlalchemy as sa
from alembic import op

revision: str = "0024_manage_permission_matrix"
down_revision: str | None = "0023_admin_permissions"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None

_MANAGE_FLAG_COLUMNS = (
    "manage_category_edit",
    "manage_org_edit",
    "manage_user_edit",
)
_LEGACY_FLAG_COLUMNS = (
    "perm_category_edit",
    "perm_org_edit",
    "perm_user_edit",
)
_NAME_SUFFIX_BUDGET = 61  # role_name 宽 64，预留 "-M" + 两位序号


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


def _migrate_legacy_permissions(bind: sa.engine.Connection) -> None:
    """把 sys_user 旧管理位收敛到生效角色（置位或克隆），随后删旧列。"""
    default_role_id = bind.execute(
        sa.text(
            "SELECT id FROM sys_role WHERE is_default = 1 AND is_deleted = 0 "
            "AND status = 'ENABLED' ORDER BY sort_no ASC, id ASC"
        )
    ).scalar()

    role_rows = bind.execute(
        sa.text(
            "SELECT id, role_name, sort_no, status, remark, "
            "manage_category_edit, manage_org_edit, manage_user_edit, managed_org_id "
            "FROM sys_role WHERE is_deleted = 0"
        )
    ).mappings().all()
    roles = {row["id"]: dict(row) for row in role_rows}

    def effective_role_id(user: dict) -> int | None:
        role_id = user["role_id"]
        if role_id is not None:
            role = roles.get(role_id)
            if role is not None and role["status"] == "ENABLED":
                return role_id
        return default_role_id

    user_rows = bind.execute(
        sa.text(
            "SELECT id, role_id, perm_category_edit, perm_org_edit, "
            "perm_user_edit, managed_org_id FROM sys_user "
            "WHERE role_code = 'ADMIN' AND is_deleted = 0"
        )
    ).mappings().all()

    # 组键 = (生效角色, 三权限位, 锚点) → 组内用户 id 列表
    groups: dict[tuple[int, int, int, int, int | None], list[int]] = {}
    for row in user_rows:
        eff_id = effective_role_id(dict(row))
        if eff_id is None:
            continue
        key = (
            eff_id,
            int(bool(row["perm_category_edit"])),
            int(bool(row["perm_org_edit"])),
            int(bool(row["perm_user_edit"])),
            row["managed_org_id"],
        )
        groups.setdefault(key, []).append(row["id"])

    for (eff_id, cat, org, usr, anchor), member_ids in groups.items():
        role = roles[eff_id]
        current = (
            int(bool(role["manage_category_edit"])),
            int(bool(role["manage_org_edit"])),
            int(bool(role["manage_user_edit"])),
            role["managed_org_id"],
        )
        target = (cat, org, usr, anchor)
        if current == target:
            continue
        # 占用该生效角色的全部组数（含已一致的组）：唯一占用直接置位，否则克隆
        occupancy = {key for key in groups if key[0] == eff_id}
        if len(occupancy) == 1:
            bind.execute(
                sa.text(
                    "UPDATE sys_role SET manage_category_edit = :cat, "
                    "manage_org_edit = :org, manage_user_edit = :usr, "
                    "managed_org_id = :anchor, row_version = row_version + 1 "
                    "WHERE id = :rid"
                ),
                {"cat": cat, "org": org, "usr": usr, "anchor": anchor, "rid": eff_id},
            )
            role.update(
                manage_category_edit=cat,
                manage_org_edit=org,
                manage_user_edit=usr,
                managed_org_id=anchor,
            )
            continue
        new_name = _free_clone_name(bind, role["role_name"])
        bind.execute(
            sa.text(
                "INSERT INTO sys_role (role_name, sort_no, status, remark, is_default, "
                "manage_category_edit, manage_org_edit, manage_user_edit, managed_org_id, "
                "row_version) VALUES (:name, :sort_no, 'ENABLED', :remark, 0, "
                ":cat, :org, :usr, :anchor, 0)"
            ),
            {
                "name": new_name,
                "sort_no": role["sort_no"],
                "remark": role["remark"],
                "cat": cat,
                "org": org,
                "usr": usr,
                "anchor": anchor,
            },
        )
        new_id = bind.execute(
            sa.text(
                "SELECT id FROM sys_role WHERE role_name = :name AND is_deleted = 0"
            ),
            {"name": new_name},
        ).scalar_one()
        bind.execute(
            sa.text(
                "INSERT INTO sys_role_permission (role_id, security_level, dept_scope, "
                "can_view, can_download, can_edit, can_delete, can_config_permission, "
                "can_bind_attachment, can_unbind_attachment) "
                "SELECT :new_id, security_level, dept_scope, can_view, can_download, "
                "can_edit, can_delete, can_config_permission, can_bind_attachment, "
                "can_unbind_attachment FROM sys_role_permission "
                "WHERE role_id = :old_id AND is_deleted = 0"
            ),
            {"new_id": new_id, "old_id": eff_id},
        )
        for user_id in member_ids:
            bind.execute(
                sa.text("UPDATE sys_user SET role_id = :nid WHERE id = :uid"),
                {"nid": new_id, "uid": user_id},
            )
        roles[new_id] = {
            "id": new_id,
            "role_name": new_name,
            "sort_no": role["sort_no"],
            "status": "ENABLED",
            "remark": role["remark"],
            "manage_category_edit": cat,
            "manage_org_edit": org,
            "manage_user_edit": usr,
            "managed_org_id": anchor,
        }


def _free_clone_name(bind: sa.engine.Connection, base_name: str) -> str:
    """为克隆角色取空闲名：截断原名后追加 -M{n} 后缀，避开活跃角色重名。"""
    stem = base_name[:_NAME_SUFFIX_BUDGET]
    taken = {
        row[0]
        for row in bind.execute(
            sa.text("SELECT role_name FROM sys_role WHERE is_deleted = 0")
        )
    }
    suffix = 1
    while suffix < 100:
        candidate = f"{stem}-M{suffix}"
        if candidate not in taken:
            return candidate
        suffix += 1
    raise RuntimeError(f"无法为角色“{base_name}”生成空闲的克隆名")


def upgrade() -> None:
    existing = _columns_of("sys_role")
    for name in _MANAGE_FLAG_COLUMNS:
        if name not in existing:
            op.add_column(
                "sys_role",
                sa.Column(
                    name,
                    sa.Boolean(),
                    nullable=False,
                    server_default=sa.text("0"),
                ),
            )
    if "managed_org_id" not in existing:
        op.add_column(
            "sys_role",
            sa.Column("managed_org_id", sa.BigInteger(), nullable=True),
        )

    user_columns = _columns_of("sys_user")
    if "perm_category_edit" in user_columns:
        _migrate_legacy_permissions(op.get_bind())

    user_columns = _columns_of("sys_user")
    if "managed_org_id" in user_columns:
        op.drop_column("sys_user", "managed_org_id")
    for name in _LEGACY_FLAG_COLUMNS:
        if name in user_columns:
            op.drop_column("sys_user", name)


def downgrade() -> None:
    # 旧列为纯用户级配置，删列后无法从角色位反推原值，仅重建结构不恢复数据。
    user_columns = _columns_of("sys_user")
    for name in _LEGACY_FLAG_COLUMNS:
        if name not in user_columns:
            op.add_column(
                "sys_user",
                sa.Column(
                    name,
                    sa.Boolean(),
                    nullable=False,
                    server_default=sa.text("0"),
                ),
            )
    if "managed_org_id" not in user_columns:
        op.add_column(
            "sys_user",
            sa.Column("managed_org_id", sa.BigInteger(), nullable=True),
        )
    existing = _columns_of("sys_role")
    if "managed_org_id" in existing:
        op.drop_column("sys_role", "managed_org_id")
    for name in reversed(_MANAGE_FLAG_COLUMNS):
        if name in existing:
            op.drop_column("sys_role", name)