zhifu-platform/scripts/prepare_platform_data.py

import asyncio
import json
import secrets
from pathlib import Path
from uuid import UUID

from sqlalchemy.ext.asyncio import async_sessionmaker

from zhifu.core.config import Settings
from zhifu.core.database import create_engine
from zhifu.modules.identity.auth import now, password_hasher
from zhifu.modules.identity.models import Tenant
from zhifu.modules.platform.models import PlatformAccount, TenantRegistration


async def main():
    settings = Settings()
    if settings.environment == "production":
        raise ValueError("开发平台账户仅允许在 development 或 test 环境创建")
    target = Path(".work/platform-accounts.json")
    existing = (
        json.loads(await asyncio.to_thread(target.read_text, encoding="utf-8"))
        if await asyncio.to_thread(target.exists) else None
    )
    password = existing["password"] if existing else secrets.token_urlsafe(24)
    actor_id = UUID("ac700000-2026-4000-8000-000000000050")
    tenant_id = UUID("ac700000-2026-4000-8000-000000000001")
    engine = create_engine(settings)
    try:
        sessions = async_sessionmaker(engine, expire_on_commit=False)
        async with sessions.begin() as session:
            account = await session.get(PlatformAccount, actor_id.bytes)
            if account is None:
                session.add(PlatformAccount(
                    id=actor_id.bytes, username="platform-admin",
                    password_hash=password_hasher.hash(password),
                    status="ACTIVE", permissions=["platform:manage"], created_at=now(),
                ))
            elif existing is None:
                raise ValueError("平台账户已存在，本地账户凭据文件缺失")
            tenant = await session.get(Tenant, tenant_id.bytes)
            if tenant is None:
                raise ValueError("请先准备前端开发商家账户")
            registration = await session.get(TenantRegistration, tenant_id.bytes)
            if registration is None:
                session.add(TenantRegistration(
                    tenant_id=tenant_id.bytes, contact="开发环境接入管理员",
                    status="COMPLETE", created_at=now(),
                ))
            tenant_name = tenant.name
        target.parent.mkdir(exist_ok=True)
        await asyncio.to_thread(target.write_text, json.dumps({
            "username": "platform-admin", "password": password, "actor_id": str(actor_id),
            "tenant_id": str(tenant_id), "tenant_name": tenant_name,
        }, ensure_ascii=False, indent=2), encoding="utf-8")
        print("平台开发账户已准备，凭据保存在 .work/platform-accounts.json，未输出密码。")
    finally:
        await engine.dispose()


if __name__ == "__main__":
    asyncio.run(main())