document-management/backend/migrations/versions/0023_admin_permissions.py

"""管理类权限：sys_user 增加分类/机构/用户编辑权限位与管理范围锚点部门。

Revision ID: 0023_admin_permissions
Revises: 0022_login_lockout
Create Date: 2026-09-02

变更要点：
1. sys_user 增加 perm_category_edit / perm_org_edit / perm_user_edit
   TINYINT(1) NOT NULL DEFAULT 0（管理类权限开关，仅 ADMIN 类别生效）；
2. sys_user 增加 managed_org_id BIGINT NULL（管理范围锚点部门，
   NULL=不受范围约束即全局；机构/用户编辑权限限定锚点部门及其下级）；
3. 存量初始化：ADMIN 类别用户三权限默认开启；admin 账号锚点保持 NULL（全局），
   其余 ADMIN 用户锚点置为所在部门；
4. 升级幂等：列已存在时跳过加列；数据初始化可重复执行（结果一致）。
"""

from __future__ import annotations

from collections.abc import Sequence

import sqlalchemy as sa
from alembic import op

revision: str = "0023_admin_permissions"
down_revision: str | None = "0022_login_lockout"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None

_TABLE = "sys_user"
_FLAG_COLUMNS = ("perm_category_edit", "perm_org_edit", "perm_user_edit")


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


def upgrade() -> None:
    existing = _existing_columns()
    for name in _FLAG_COLUMNS:
        if name not in existing:
            op.add_column(
                _TABLE,
                sa.Column(
                    name,
                    sa.Boolean(),
                    nullable=False,
                    server_default=sa.text("0"),
                ),
            )
    if "managed_org_id" not in existing:
        op.add_column(
            _TABLE,
            sa.Column("managed_org_id", sa.BigInteger(), nullable=True),
        )
    # 存量初始化：管理员类别默认开启三权限；admin 账号锚点 NULL（全局），其余锚点=所在部门
    op.execute(
        "UPDATE sys_user SET perm_category_edit = 1, perm_org_edit = 1, "
        "perm_user_edit = 1 WHERE role_code = 'ADMIN' AND is_deleted = 0"
    )
    op.execute(
        "UPDATE sys_user SET managed_org_id = organization_id "
        "WHERE role_code = 'ADMIN' AND is_deleted = 0 "
        "AND username <> 'admin' AND organization_id IS NOT NULL"
    )


def downgrade() -> None:
    existing = _existing_columns()
    if "managed_org_id" in existing:
        op.drop_column(_TABLE, "managed_org_id")
    for name in reversed(_FLAG_COLUMNS):
        if name in existing:
            op.drop_column(_TABLE, name)