document-management/backend/dms/services/admin_permission.py

"""管理类权限判定：分类/机构/用户编辑开关与管理范围锚点。

语义（管理权限矩阵化后）：
- 三管理位（manage_category_edit / manage_org_edit / manage_user_edit）挂在
  角色上，判定读"生效角色"（用户显式角色，未挂角色回退默认角色），
  且仅对 ADMIN 类别生效；分类编辑全局生效，机构/用户编辑受管理范围约束；
- 角色上 managed_org_id 为管理范围锚点：NULL=不受范围约束（全局），否则
  机构/用户编辑限定锚点部门及其下级子树（沿启用边BFS）；
- 内置 admin 账号恒为全权超管，不受开关与范围约束。
"""

from __future__ import annotations

from sqlalchemy import select

from dms.common.enums import EnabledStatus, RoleCode
from dms.common.errors import ForbiddenError
from dms.models import Organization, Role, User
from dms.security.auth_context import get_auth_context

SUPER_ADMIN_USERNAME = "admin"

# managed_org_id 哨兵值 0 = "SELF"（锚点跟随用户所在部门）：
# 0 不是合法机构ID，机构/用户编辑范围按各用户自身 organization_id 收敛。
MANAGED_ORG_SELF_ANCHOR = 0
MANAGED_ORG_SELF = "SELF"


def effective_manage_role(session, user: User) -> Role | None:
    """生效角色：显式角色（活跃且启用）优先，否则回退默认角色（活跃且启用）。"""
    if user.role_id is not None:
        role = session.get(Role, user.role_id)
        if (
            role is not None
            and not role.is_deleted
            and role.status == EnabledStatus.ENABLED.value
        ):
            return role
    return session.scalar(
        select(Role).where(
            Role.is_default.is_(True),
            Role.is_deleted.is_(False),
            Role.status == EnabledStatus.ENABLED.value,
        )
    )


def is_super_admin(user: User) -> bool:
    """内置 admin 账号恒为超管，不受三管理位与管理范围约束。"""
    return user.username == SUPER_ADMIN_USERNAME


def _has_manage_flag(session, user: User, flag: str) -> bool:
    if user.role_code != RoleCode.ADMIN.value:
        return False
    role = effective_manage_role(session, user)
    return role is not None and bool(getattr(role, flag))


def can_edit_categories(session, user: User) -> bool:
    if is_super_admin(user):
        return True
    return _has_manage_flag(session, user, "manage_category_edit")


def can_edit_organizations(session, user: User) -> bool:
    if is_super_admin(user):
        return True
    return _has_manage_flag(session, user, "manage_org_edit")


def can_edit_users(session, user: User) -> bool:
    if is_super_admin(user):
        return True
    return _has_manage_flag(session, user, "manage_user_edit")


def require_category_editor(session) -> User:
    """取当前认证用户并校验分类编辑权限。"""
    actor = session.get(User, get_auth_context().user_id)
    if actor is None or not can_edit_categories(session, actor):
        raise ForbiddenError("无分类编辑权限")
    return actor


def require_organization_editor(session) -> User:
    """取当前认证用户并校验机构编辑权限（范围按目标另行判定）。"""
    actor = session.get(User, get_auth_context().user_id)
    if actor is None or not can_edit_organizations(session, actor):
        raise ForbiddenError("无机构编辑权限")
    return actor


def require_user_editor(session) -> User:
    """取当前认证用户并校验用户编辑权限（范围按目标另行判定）。"""
    actor = session.get(User, get_auth_context().user_id)
    if actor is None or not can_edit_users(session, actor):
        raise ForbiddenError("无用户编辑权限")
    return actor


def managed_scope_ids(session, user: User) -> set[int] | None:
    """操作者可管理的机构ID集合；None 表示不受范围约束（全局）。

    super_admin 与生效角色锚点为 NULL 的管理员不受约束；锚点为 SELF
    哨兵（0）时按用户自身所在部门收敛（无部门则无可管理范围）；其余
    按锚点部门子树（仅沿启用边）收敛。
    """
    if is_super_admin(user):
        return None
    role = effective_manage_role(session, user)
    anchor = role.managed_org_id if role is not None else None
    if anchor is None:
        return None
    if anchor == MANAGED_ORG_SELF_ANCHOR:
        if user.organization_id is None:
            return set()
        anchor = user.organization_id
    organizations = session.scalars(
        select(Organization).where(Organization.is_deleted.is_(False))
    ).all()
    children: dict[int | None, list[int]] = {}
    for item in organizations:
        if item.status == EnabledStatus.ENABLED.value:
            children.setdefault(item.parent_id, []).append(item.id)
    scope = {anchor}
    pending = [anchor]
    while pending:
        current = pending.pop()
        for child_id in children.get(current, []):
            if child_id not in scope:
                scope.add(child_id)
                pending.append(child_id)
    return scope


def ensure_org_in_managed_scope(session, actor: User, org_id: int | None) -> None:
    """校验目标机构在操作者管理范围内。

    org_id 为 None 表示顶级机构（或无部门用户）：仅全局管理员可操作。
    """
    scope = managed_scope_ids(session, actor)
    if scope is None:
        return
    if org_id is None or org_id not in scope:
        raise ForbiddenError("目标超出管理范围")