refactor: 角色组织账号模型重构 - 阶段二(保守)

1. 创建数据迁移脚本 migrate_members_to_org.py - ParkMember.admin 同步到 organization_members (type=carrier) - CompanyMember.is_admin 同步到 organization_members (type=enterprise) - 自动创建缺失的 organization 记录，幂等 2. 修改 _compute_capabilities，优先查 organization_members - carrier 能力：org_members.role=admin + org.type=carrier - enterprise 能力：org_members.role=admin + org.type=enterprise - 旧表 park_members/company_members 保留兼容 3. 不删除旧表，不修改表结构，风险可控
2026-09-03 00:39:41 +08:00
parent 6bb988cc3e
commit 97f3c1956c
2 changed files with 176 additions and 14 deletions
@@ -41,10 +41,12 @@ async def _compute_capabilities(db: Database, user: dict) -> list[str]:
    """叠加制能力集合（身份不互斥）：
    - opc_member：所有账号的基础能力；
    - operator：运营端角色；
-    - carrier：园区端角色 或 已绑定园区管理员（park_members.admin / operator_user_id）；
-    - enterprise：已绑定企业管理员（company_members.is_admin）。
+    - carrier：园区端角色 或 已绑定园区管理员（organization_members.role=admin + org.type=carrier，或旧表 park_members.admin / park_tenants.operator_user_id）；
+    - enterprise：已绑定企业管理员（organization_members.role=admin + org.type=enterprise，或旧表 company_members.is_admin）。
+
+    优先查统一的 organization_members 表，回退查旧表（park_members/company_members），保持兼容。
    """
-    from ..infrastructure.models import CompanyMember, ParkMember, ParkTenant
+    from ..infrastructure.models import CompanyMember, Organization, OrganizationMember, ParkMember, ParkTenant
    from sqlalchemy import select

    role = user.get("role", "")
@@ -54,19 +56,49 @@ async def _compute_capabilities(db: Database, user: dict) -> list[str]:
    if role in ("carrier", "park", "park_staff", "carrier_staff"):
        caps.append("carrier")
    else:
-        bound = await db.session.scalar(select(ParkMember).where(
-            ParkMember.user_id == user.get("id", ""), ParkMember.park_id != "",
-            ParkMember.member_type == "admin", ParkMember.status == "active").limit(1))
-        if not bound:
-            bound = await db.session.scalar(select(ParkTenant).where(
-                ParkTenant.operator_user_id == user.get("id", "")).limit(1))
-        if bound:
+        # 优先查统一表：organization_members.role=admin + org.type=carrier
+        admin_orgs = (await db.session.scalars(
+            select(Organization.id).join(
+                OrganizationMember, OrganizationMember.org_id == Organization.id
+            ).where(
+                OrganizationMember.user_id == user.get("id", ""),
+                OrganizationMember.role == "admin",
+                OrganizationMember.status == "active",
+                Organization.type == "carrier",
+            )
+        )).all()
+        if admin_orgs:
            caps.append("carrier")
-    ent = await db.session.scalar(select(CompanyMember).where(
-        CompanyMember.user_id == user.get("id", ""), CompanyMember.is_admin.is_(True),
-        CompanyMember.status == "active").limit(1))
-    if ent:
+        else:
+            # 回退查旧表
+            bound = await db.session.scalar(select(ParkMember).where(
+                ParkMember.user_id == user.get("id", ""), ParkMember.park_id != "",
+                ParkMember.member_type == "admin", ParkMember.status == "active").limit(1))
+            if not bound:
+                bound = await db.session.scalar(select(ParkTenant).where(
+                    ParkTenant.operator_user_id == user.get("id", "")).limit(1))
+            if bound:
+                caps.append("carrier")
+
+    # enterprise 能力：优先查统一表，回退查旧表
+    ent_orgs = (await db.session.scalars(
+        select(Organization.id).join(
+            OrganizationMember, OrganizationMember.org_id == Organization.id
+        ).where(
+            OrganizationMember.user_id == user.get("id", ""),
+            OrganizationMember.role == "admin",
+            OrganizationMember.status == "active",
+            Organization.type == "enterprise",
+        )
+    )).all()
+    if ent_orgs:
        caps.append("enterprise")
+    else:
+        ent = await db.session.scalar(select(CompanyMember).where(
+            CompanyMember.user_id == user.get("id", ""), CompanyMember.is_admin.is_(True),
+            CompanyMember.status == "active").limit(1))
+        if ent:
+            caps.append("enterprise")
    return caps


@@ -0,0 +1,130 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""数据迁移：把 ParkMember.admin 和 CompanyMember.is_admin 同步到 OrganizationMember。
+
+背景：当前存在三套成员关系表（organization_members / park_members / company_members），
+目标是统一到 organization_members。本脚本只做数据同步，不删除旧表，保留兼容。
+
+用法：
+    cd code/server-core
+    uv run python scripts/db/migrate_members_to_org.py
+
+逻辑：
+1. 遍历 ParkMember.member_type == 'admin' 的记录
+2. 确保对应的 park_tenant 有 organization 记录（type=carrier），没有则创建
+3. 在 organization_members 中创建/更新记录（role=admin, is_admin=True）
+4. 同样处理 CompanyMember.is_admin == True 的记录（type=enterprise）
+"""
+from __future__ import annotations
+
+import asyncio
+import sys
+from pathlib import Path
+
+# 确保项目根目录在 path 中
+sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+
+from sqlalchemy import select
+from app.infrastructure.database import async_engine, AsyncSessionLocal
+from app.infrastructure.models import (
+    Organization, OrganizationMember, ParkMember, CompanyMember, ParkTenant, ParkCompany,
+)
+from app.infrastructure.repositories import new_id, utcnow_iso
+
+
+async def migrate() -> None:
+    async with AsyncSessionLocal() as session:
+        now = utcnow_iso()
+        stats = {"park_admin": 0, "company_admin": 0, "org_created": 0, "member_created": 0, "member_skipped": 0}
+
+        # ── 1. 迁移 ParkMember.admin ──────────────────────────────────────
+        park_admins = (await session.scalars(
+            select(ParkMember).where(
+                ParkMember.member_type == "admin",
+                ParkMember.status == "active",
+            )
+        )).all()
+
+        for pm in park_admins:
+            stats["park_admin"] += 1
+            # 查找或创建对应的 organization（type=carrier）
+            org = await session.scalar(
+                select(Organization).where(Organization.id == pm.park_id)
+            )
+            if org is None:
+                # 查找 park_tenant 获取名称
+                tenant = await session.scalar(
+                    select(ParkTenant).where(ParkTenant.id == pm.park_id)
+                )
+                org_name = tenant.name if tenant else f"park-{pm.park_id}"
+                org = Organization(
+                    id=pm.park_id, name=org_name, type="carrier",
+                    region_id=None, parent_id=None, created_at=now,
+                )
+                session.add(org)
+                stats["org_created"] += 1
+                await session.flush()
+
+            # 检查是否已存在 organization_member
+            existing = await session.get(OrganizationMember, (pm.park_id, pm.user_id))
+            if existing:
+                stats["member_skipped"] += 1
+                continue
+
+            session.add(OrganizationMember(
+                org_id=pm.park_id, user_id=pm.user_id,
+                role="admin", is_admin=True, status="active", joined_at=now,
+            ))
+            stats["member_created"] += 1
+
+        # ── 2. 迁移 CompanyMember.is_admin ────────────────────────────────
+        company_admins = (await session.scalars(
+            select(CompanyMember).where(
+                CompanyMember.is_admin.is_(True),
+                CompanyMember.status == "active",
+            )
+        )).all()
+
+        for cm in company_admins:
+            stats["company_admin"] += 1
+            # 查找或创建对应的 organization（type=enterprise）
+            org = await session.scalar(
+                select(Organization).where(Organization.id == cm.company_id)
+            )
+            if org is None:
+                # 查找 park_company 获取名称
+                company = await session.scalar(
+                    select(ParkCompany).where(ParkCompany.id == cm.company_id)
+                )
+                org_name = company.name if company else f"company-{cm.company_id}"
+                org = Organization(
+                    id=cm.company_id, name=org_name, type="enterprise",
+                    region_id=None, parent_id=None, created_at=now,
+                )
+                session.add(org)
+                stats["org_created"] += 1
+                await session.flush()
+
+            # 检查是否已存在 organization_member
+            existing = await session.get(OrganizationMember, (cm.company_id, cm.user_id))
+            if existing:
+                stats["member_skipped"] += 1
+                continue
+
+            session.add(OrganizationMember(
+                org_id=cm.company_id, user_id=cm.user_id,
+                role="admin", is_admin=True, status="active", joined_at=now,
+            ))
+            stats["member_created"] += 1
+
+        await session.commit()
+        print(f"迁移完成: {stats}")
+        print(f"  - 园区管理员: {stats['park_admin']} 条")
+        print(f"  - 企业管理员: {stats['company_admin']} 条")
+        print(f"  - 新建组织: {stats['org_created']} 个")
+        print(f"  - 新建成员关系: {stats['member_created']} 条")
+        print(f"  - 已存在跳过: {stats['member_skipped']} 条")
+
+
+if __name__ == "__main__":
+    asyncio.run(migrate())