168 lines
4.5 KiB
Python
168 lines
4.5 KiB
Python
"""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,
|
|
OPERATOR_PERMISSIONS,
|
|
VIEWER_PERMISSIONS,
|
|
WRITE_PERMISSIONS,
|
|
normalize_permissions,
|
|
)
|
|
|
|
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 when the role has global data scope (built-in admin)."""
|
|
record = get_cached_role(role)
|
|
if record is not None:
|
|
return bool(record.is_admin)
|
|
# Fallback before cache is warm / for unit tests.
|
|
return str(role or "") == ROLE_ADMIN
|
|
|
|
|
|
def can_write(role: str | None) -> bool:
|
|
record = get_cached_role(role)
|
|
if record is not None:
|
|
if record.is_admin:
|
|
return True
|
|
return any(code in WRITE_PERMISSIONS for code in record.permissions)
|
|
return str(role or "") in (ROLE_ADMIN, ROLE_OPERATOR)
|
|
|
|
|
|
def can_manage_users(role: str | None) -> bool:
|
|
return has_permission(role, "users.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 code in OPERATOR_PERMISSIONS
|
|
if str(role or "") == ROLE_VIEWER:
|
|
return code in VIEWER_PERMISSIONS
|
|
return False
|
|
if record.is_admin:
|
|
return True
|
|
return code in record.permissions
|
|
|
|
|
|
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:
|
|
return list(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)
|
|
return normalize_permissions(codes)
|