refactor: users.role 归一化为默认端口标记

1. 创建 normalize_user_roles.py 迁移脚本
2. 8 个旧角色用户(developer/enterprise/investor/provider)归一化为 opc_member
3. account_types.py 注释明确 role 语义:仅用于默认端口和前端展示
4. 权限判断完全依赖 capabilities 叠加制,role 仅作回退安全网
This commit is contained in:
Pine
2026-09-03 09:20:24 +08:00
parent d851b54755
commit 5b229f3de8
2 changed files with 60 additions and 1 deletions
+5 -1
View File
@@ -6,7 +6,11 @@
carrier 载体方 —— 创业载体/园区运营方
opc_member OPC —— 超级个体(含企业管理员能力:is_admin 只是企业成员标记,不是身份)
注意:「企业管理员」不是身份,是 OPC 之上的能力叠加(company_members.is_admin);
users.role 字段语义:默认端口标记(operator/carrier/opc_member),仅用于登录后
默认进入哪个端口和前端展示。权限判断完全依赖 JWT 中的 capabilities 叠加制能力集合,
role 字段本身不参与权限判定(role_allowed 中 role 仅作回退安全网)。
注意:「企业管理员」不是身份,是 OPC 之上的能力叠加(organization_members.role=admin + org.type=enterprise);
「甲方企业/投资人/服务商」等是历史业务概念,不属于当前身份体系,统一归为 OPC 基础身份。
子角色(sub_role)已停用:权限一律按账号类型解析,不再读子角色。
"""
+55
View File
@@ -0,0 +1,55 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""数据迁移:归一化 users.role 字段。
把旧的角色值(enterprise/provider/government/investor/developer/service 等)
归一化为三种账号类型之一(operator/carrier/opc_member)。
用法:
cd code/server-core
uv run python scripts/db/normalize_user_roles.py
"""
from __future__ import annotations
import asyncio
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from sqlalchemy import select, update
from app.infrastructure.db import AsyncSessionLocal
from app.infrastructure.models import User
from app.domain.account_types import account_type
async def migrate() -> None:
async with AsyncSessionLocal() as session:
users = (await session.scalars(select(User))).all()
stats = {"total": 0, "changed": 0, "unchanged": 0}
changes = []
for u in users:
stats["total"] += 1
atype = account_type(u.role)
if u.role != atype:
changes.append((u.id, u.username, u.role, atype))
u.role = atype
stats["changed"] += 1
else:
stats["unchanged"] += 1
await session.commit()
print(f"归一化完成: {stats}")
print(f" - 总用户: {stats['total']}")
print(f" - 已变更: {stats['changed']}")
print(f" - 未变更: {stats['unchanged']}")
if changes:
print("\n变更明细:")
for uid, username, old, new in changes:
print(f" {uid} ({username}): {old} -> {new}")
if __name__ == "__main__":
asyncio.run(migrate())