zhifu-platform/scripts/prepare_frontend_accounts.py

import asyncio
import json
import secrets
from datetime import UTC, datetime, timedelta
from uuid import UUID, uuid4

from anyio import Path
from sqlalchemy.ext.asyncio import async_sessionmaker

from zhifu.core.config import Settings
from zhifu.core.database import create_engine
from zhifu.modules.after_sales.models import BusinessEvent, ExchangeRule, OrderProjection
from zhifu.modules.billing.models import BillingGrant
from zhifu.modules.identity.access_models import Principal, StoreGrant
from zhifu.modules.identity.auth import password_hasher
from zhifu.modules.identity.models import Store, Tenant


async def main():
    settings = Settings()
    if settings.environment == "production":
        raise ValueError("开发账户只能在 development 或 test 环境创建")
    engine = create_engine(settings)
    sessions = async_sessionmaker(engine, expire_on_commit=False)
    tenant_id = UUID("ac700000-2026-4000-8000-000000000001")
    store_id = UUID("ac700000-2026-4000-8000-000000000002")
    target = Path(".work/frontend-accounts.json")
    # 数据保留在独立开发商家中；重复运行只更新该商家的演示订单观察时间。
    existing = (
        json.loads(await target.read_text(encoding="utf-8")) if await target.exists() else None
    )
    password = existing["password"] if existing else secrets.token_urlsafe(18)
    roles = {
        "customer01": (
            "CUSTOMER",
            ["orders:read", "tickets:read", "after_sales:create",
             "after_sales:confirm", "after_sales:cancel"],
        ),
        "staff01": (
            "STAFF",
            ["orders:read", "tickets:read", "after_sales:create", "after_sales:cancel",
             "conversations:read", "conversations:reply", "conversations:manage"],
        ),
        "supervisor01": ("STAFF", ["orders:read", "tickets:read", "approvals:decide"]),
        "admin01": (
            "STAFF",
            ["orders:read", "tickets:read", "rules:write", "knowledge:write", "billing:read",
             "billing:request", "billing:approve", "stores:manage", "staff:manage",
             "channels:manage", "agent:publish"],
        ),
        "finance02": ("STAFF", ["billing:read", "billing:request", "billing:approve"]),
    }
    actors = {}
    now = datetime.now(UTC).replace(tzinfo=None)
    try:
        async with sessions.begin() as session:
            tenant = await session.get(Tenant, tenant_id.bytes)
            if tenant is None:
                session.add(Tenant(id=tenant_id.bytes, name="前端开发独立商家"))
                await session.flush()
                session.add(Store(
                    id=store_id.bytes, tenant_id=tenant_id.bytes, name="前端开发配件店"
                ))
                await session.flush()
            elif existing is None:
                raise ValueError("开发商家已存在，但本地账户文件缺失；请使用已有开发账户")
            for index, (username, (kind, permissions)) in enumerate(roles.items(), start=10):
                actor_id = UUID(f"ac700000-2026-4000-8000-{index:012d}")
                actors[username] = str(actor_id)
                principal = await session.get(Principal, actor_id.bytes)
                if principal is None:
                    session.add(Principal(
                        id=actor_id.bytes, tenant_id=tenant_id.bytes, username=username,
                        password_hash=password_hasher.hash(password), kind=kind,
                    ))
                    await session.flush()
                    for permission in permissions:
                        session.add(StoreGrant(
                            tenant_id=tenant_id.bytes, principal_id=actor_id.bytes,
                            store_id=store_id.bytes, permission=permission,
                        ))
                        if permission.startswith("billing:"):
                            session.add(BillingGrant(
                                tenant_id=tenant_id.bytes, principal_id=actor_id.bytes,
                                permission=permission,
                            ))
                elif username == "admin01":
                    # 开发管理员补充本次管理接口权限，其他账户授权保持原有配置。
                    for permission in (
                        "stores:manage", "staff:manage", "channels:manage", "agent:publish"
                    ):
                        if await session.get(
                            StoreGrant,
                            (tenant_id.bytes, actor_id.bytes, store_id.bytes, permission),
                        ) is None:
                            session.add(StoreGrant(
                                tenant_id=tenant_id.bytes, principal_id=actor_id.bytes,
                                store_id=store_id.bytes, permission=permission,
                            ))
            await session.flush()
            order_id = UUID("ac700000-2026-4000-8000-000000000003")
            order = await session.get(OrderProjection, order_id.bytes)
            if order is None:
                session.add(OrderProjection(
                    id=order_id.bytes, tenant_id=tenant_id.bytes, store_id=store_id.bytes,
                    subject_id=UUID(actors["customer01"]).bytes,
                    source_system="frontend-development", source_version=1,
                    item_id="keyboard-01", product_type="keyboard", available_quantity=10,
                    signed_at=now - timedelta(days=1), observed_at=now,
                ))
            else:
                order.observed_at = now
            rule = await session.get(ExchangeRule, (tenant_id.bytes, store_id.bytes, 1))
            if rule is None:
                session.add(ExchangeRule(
                    tenant_id=tenant_id.bytes, store_id=store_id.bytes, version=1,
                    window_days=7, product_types=["keyboard"],
                    citation="签收七天内的键盘故障支持申请换货，以当前审批与履约结果为准。",
                ))
            session.add(BusinessEvent(
                id=uuid4().bytes, tenant_id=tenant_id.bytes, actor_id=UUID(int=0).bytes,
                aggregate_id=tenant_id.bytes, aggregate_version=1,
                event_type="development.frontend_accounts", request_id=str(uuid4()),
                payload={"source": "prepare_frontend_accounts", "accounts": list(roles)},
                occurred_at=now,
            ))
        await target.parent.mkdir(exist_ok=True)
        await target.write_text(
            json.dumps(
                {"tenant_id": str(tenant_id), "store_id": str(store_id),
                 "order_id": str(order_id), "password": password, "actors": actors},
                ensure_ascii=False, indent=2,
            ),
            encoding="utf-8",
        )
        print("开发账户信息已保存至 .work/frontend-accounts.json；未输出密码。")
    finally:
        await engine.dispose()


asyncio.run(main())