dy/backend/auth/roles.py

"""Role code helpers and an in-memory role registry backed by the roles table."""

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Iterable

from .permissions import (
    ALL_PERMISSIONS,
    DATA_SCOPE_ALL,
    OPERATOR_PERMISSIONS,
    VIEWER_PERMISSIONS,
    WRITE_PERMISSIONS,
    expand_legacy_bundles,
    permission_implies,
)


ROLE_ADMIN = "admin"
ROLE_OPERATOR = "operator"
ROLE_VIEWER = "viewer"

ALL_ROLES = (ROLE_ADMIN, ROLE_OPERATOR, ROLE_VIEWER)

ROLE_LABELS = {
    ROLE_ADMIN: "管理员",
    ROLE_OPERATOR: "运营",
    ROLE_VIEWER: "只读",
}


@dataclass
class RoleRecord:
    code: str
    label: str
    description: str = ""
    is_system: bool = False
    is_admin: bool = False
    permissions: list[str] = field(default_factory=list)


_ROLE_CACHE: dict[str, RoleRecord] = {}


def default_role_seeds() -> list[RoleRecord]:
    return [
        RoleRecord(
            code=ROLE_ADMIN,
            label=ROLE_LABELS[ROLE_ADMIN],
            description="拥有全部菜单、按钮与全局数据权限",
            is_system=True,
            is_admin=True,
            permissions=list(ALL_PERMISSIONS),
        ),
        RoleRecord(
            code=ROLE_OPERATOR,
            label=ROLE_LABELS[ROLE_OPERATOR],
            description="管理自己的账号、规则与私信（仅本人数据）",
            is_system=True,
            is_admin=False,
            permissions=list(OPERATOR_PERMISSIONS),
        ),
        RoleRecord(
            code=ROLE_VIEWER,
            label=ROLE_LABELS[ROLE_VIEWER],
            description="仅查看自己的业务数据，不可修改",
            is_system=True,
            is_admin=False,
            permissions=list(VIEWER_PERMISSIONS),
        ),
    ]


def set_role_cache(roles: Iterable[RoleRecord]) -> None:
    global _ROLE_CACHE
    _ROLE_CACHE = {role.code: role for role in roles}


def get_cached_role(code: str | None) -> RoleRecord | None:
    if not code:
        return None
    return _ROLE_CACHE.get(str(code))


def list_cached_roles() -> list[RoleRecord]:
    return list(_ROLE_CACHE.values())


def role_label(code: str | None) -> str:
    role = get_cached_role(code)
    if role:
        return role.label
    return ROLE_LABELS.get(str(code or ""), str(code or ""))


def is_admin(role: str | None) -> bool:
    """True for the built-in admin role (global admin flag)."""
    record = get_cached_role(role)
    if record is not None:
        return bool(record.is_admin)
    return str(role or "") == ROLE_ADMIN


def has_global_scope(role: str | None) -> bool:
    """True when the role may see all users' data (admin or data.scope_all)."""
    if is_admin(role):
        return True
    return has_permission(role, DATA_SCOPE_ALL)


def can_write(role: str | None) -> bool:
    record = get_cached_role(role)
    if record is not None:
        if record.is_admin:
            return True
        held = expand_legacy_bundles(set(record.permissions))
        return any(code in WRITE_PERMISSIONS for code in held)
    return str(role or "") in (ROLE_ADMIN, ROLE_OPERATOR)


def can_manage_users(role: str | None) -> bool:
    return has_permission(role, "users.manage")


def can_manage_roles(role: str | None) -> bool:
    return has_permission(role, "roles.manage")


def has_permission(role: str | None, permission: str) -> bool:
    code = str(permission or "").strip()
    if not code:
        return False
    record = get_cached_role(role)
    if record is None:
        if str(role or "") == ROLE_ADMIN:
            return True
        if str(role or "") == ROLE_OPERATOR:
            return permission_implies(expand_legacy_bundles(set(OPERATOR_PERMISSIONS)), code)
        if str(role or "") == ROLE_VIEWER:
            return permission_implies(set(VIEWER_PERMISSIONS), code)
        return False
    if record.is_admin:
        return True
    held = expand_legacy_bundles(set(record.permissions))
    return permission_implies(held, code)


def permissions_for_role(role: str | None) -> list[str]:
    record = get_cached_role(role)
    if record is None:
        if str(role or "") == ROLE_ADMIN:
            return list(ALL_PERMISSIONS)
        if str(role or "") == ROLE_OPERATOR:
            from .permissions import expand_paired_permissions

            return expand_paired_permissions(OPERATOR_PERMISSIONS)
        if str(role or "") == ROLE_VIEWER:
            return list(VIEWER_PERMISSIONS)
        return []
    if record.is_admin:
        return list(ALL_PERMISSIONS)
    return list(record.permissions)


def ensure_role(role: str) -> str:
    """Validate that a role code exists (cache or built-in fallback)."""
    code = str(role or "").strip()
    if not code:
        raise ValueError("角色不能为空")
    if get_cached_role(code) is not None:
        return code
    if code in ALL_ROLES:
        return code
    raise ValueError(f"无效角色: {code}")


def sanitize_role_permissions(
    codes: list[str] | None,
    *,
    force_all: bool = False,
) -> list[str]:
    if force_all:
        return list(ALL_PERMISSIONS)
    from .permissions import expand_paired_permissions

    return expand_paired_permissions(codes)