from pydantic import BaseModel, Field class SetUserRoleRequest(BaseModel): role: str sub_role: str | None = None org_id: str | None = None region_id: str | None = None class SetUserStatusRequest(BaseModel): status: str = "active" # active | disabled class UserClassificationRequest(BaseModel): """用户 认证状态/所属/账号类型 分类字段。""" certification_status: str | None = None # uncertified|pending|certified|rejected certification_time: str | None = None affiliation: str | None = None # independent|park park_id: str | None = None park_name: str | None = None account_type: str | None = None # opc_default|park_staff|platform_staff|provider|enterprise class UpdateUserRequest(BaseModel): nickname: str | None = None role: str | None = None sub_role: str | None = None org_id: str | None = None region_id: str | None = None class UserCreateRequest(BaseModel): """管理员建号:账号由系统随机生成(不可指定),手机号必填且全局唯一。""" phone: str # 手机号(必填,11 位大陆手机号) password: str nickname: str = "" role: str = "opc_member" sub_role: str | None = None org_id: str | None = None region_id: str | None = None class BindPhoneRequest(BaseModel): """为已有用户绑定/更换手机号。""" phone: str class RolePermissionRequest(BaseModel): permissions: list[str] = [] async def _validate_role_assignment(db: Database, actor: dict, role: str, sub_role: str | None) -> None: """白名单 + 提权防护:拒绝非法角色组合,及无权者创建/提升特权角色。""" key = role if not sub_role else f"{role}|{sub_role}" if not await db.roles.role_exists(role): raise HTTPException(status_code=400, detail=f"无效角色: {role}") if sub_role and not await db.roles.role_exists(f"{role}|{sub_role}"): raise HTTPException(status_code=400, detail=f"无效角色组合: {key}") # 目标角色若授予「配置角色权限」,仅持该权限者(超管)可创建/分配 if "action:role.grant_perm" in await db.roles.permissions_for(role, sub_role) and \ "action:role.grant_perm" not in actor.get("permissions", []): raise HTTPException( status_code=403, detail="无权创建/分配特权角色(仅超级管理员)", ) class OpcCertReviewRequest(BaseModel): status: str = "certified" # certified | rejected comment: str = "" class ParkAdmissionReviewRequest(BaseModel): status: str = "approved" # approved | rejected | reviewing comment: str = "" class ParkTransferReviewRequest(BaseModel): status: str = "approved" # approved | rejected | reviewing comment: str = ""