"""管理权限矩阵化：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)