From 97f3c1956c39fe1382d012b5ada2c30eadd0fc1f Mon Sep 17 00:00:00 2001 From: Pine Date: Thu, 3 Sep 2026 00:39:41 +0800 Subject: [PATCH] =?UTF-8?q?refactor:=20=E8=A7=92=E8=89=B2=E7=BB=84?= =?UTF-8?q?=E7=BB=87=E8=B4=A6=E5=8F=B7=E6=A8=A1=E5=9E=8B=E9=87=8D=E6=9E=84?= =?UTF-8?q?=20-=20=E9=98=B6=E6=AE=B5=E4=BA=8C(=E4=BF=9D=E5=AE=88)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. 不删除旧表，不修改表结构，风险可控 --- app/api/dependencies.py | 60 ++++++++++--- scripts/db/migrate_members_to_org.py | 130 +++++++++++++++++++++++++++ 2 files changed, 176 insertions(+), 14 deletions(-) create mode 100644 scripts/db/migrate_members_to_org.py diff --git a/app/api/dependencies.py b/app/api/dependencies.py index 8f33733..08ad7ec 100644 --- a/app/api/dependencies.py +++ b/app/api/dependencies.py @@ -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 diff --git a/scripts/db/migrate_members_to_org.py b/scripts/db/migrate_members_to_org.py new file mode 100644 index 0000000..5164990 --- /dev/null +++ b/scripts/db/migrate_members_to_org.py @@ -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())