Pine
dbc29eee59
- park_members/company_members 多对多成员表（0028），users.affiliation 为派生缓存由 sync_user_affiliation 刷新 - 转园两级审核状态机（0029）：pending_from→pending_to→approved，支持用户/企业整体转园与运营端兜底 - 叠加制身份（0030）：身份恒为运营/载体/OPC，企业管理员为能力叠加非身份，去除 enterprise/carrier 直写角色 - rbac 新增 memberships/park-transfers/企业管理员工作台 /admin/ent/* 接口 Co-Authored-By: Claude <noreply@anthropic.com>
64 lines
2.6 KiB
Python
64 lines
2.6 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""领域层 · RBAC 判定规则（纯函数，不依赖 HTTP/数据库）。
|
|
|
|
四层架构：领域层定义核心业务模型与规则；接口层守卫（rbac.py 的 Depends
|
|
工厂）只负责调用这些规则并映射为 HTTP 异常，不承载判定语义。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
# 数据范围层级（越靠前权限越大）
|
|
SCOPE_ORDER = ("province", "city", "district")
|
|
|
|
|
|
def is_platform_admin(user: dict) -> bool:
|
|
"""平台管理方（平台端运营者）：拥有全权限，可执行全部平台管理操作。
|
|
与其它端口（投资人/OPC/政务/企业/服务商/载体）不关联。"""
|
|
return user.get("role") == "operator" or user.get("sub_role") in ("op_super_admin", "superadmin")
|
|
|
|
|
|
def role_allowed(user: dict, *roles: str) -> bool:
|
|
"""当前用户业务角色是否 ∈ 允许角色集合。
|
|
|
|
身份模型（叠加制，非割裂）：每个账号都自带「OPC 基础身份」——运营方(operator)、
|
|
载体方(carrier) 等只是在其上的业务角色叠加。因此凡要求 opc_member 的能力
|
|
（C 端通用能力：浏览/点赞/留言/报名/接单…），任意登录账号均具备。
|
|
"""
|
|
if "opc_member" in roles:
|
|
return True
|
|
# 叠加制能力：绑定园区管理员/企业管理员等叠加出的能力同样命中守卫
|
|
caps = set(user.get("capabilities") or [])
|
|
if caps and caps.intersection(set(roles)):
|
|
return True
|
|
return user.get("role") in set(roles)
|
|
|
|
|
|
def sub_role_allowed(user: dict, *sub_roles: str) -> bool:
|
|
"""当前用户子角色是否 ∈ 允许子角色集合（如 op_super_admin）。"""
|
|
return user.get("sub_role") in set(sub_roles)
|
|
|
|
|
|
def permission_granted(user: dict, perm: str) -> bool:
|
|
"""当前用户是否拥有权限码 perm。平台管理方(operator/op_super_admin)恒授予全权限。"""
|
|
if is_platform_admin(user):
|
|
return True
|
|
return perm in user.get("permissions", [])
|
|
|
|
|
|
def is_government(user: dict) -> bool:
|
|
return user.get("role") == "government"
|
|
|
|
|
|
def scope_level_ok(user: dict, min_level: str | None) -> bool:
|
|
"""政务数据范围层级校验：不低于 min_level。"""
|
|
if not user.get("region_id"):
|
|
return False
|
|
if not min_level or not user.get("scope_level"):
|
|
return True
|
|
cur = user["scope_level"]
|
|
return cur in SCOPE_ORDER and SCOPE_ORDER.index(cur) <= SCOPE_ORDER.index(min_level)
|
|
|
|
|
|
def region_in_scope(user: dict, region_id: str | None) -> bool:
|
|
"""目标区域是否落在当前用户数据范围内。"""
|
|
return bool(region_id) and region_id in user.get("scope_region_ids", [])
|