zhifu-platform/scripts/prepare_storefront_data.py

import asyncio
import json
import secrets
from uuid import UUID

import httpx
from anyio import Path as AsyncPath
from sqlalchemy.ext.asyncio import async_sessionmaker

from zhifu.core.config import Settings
from zhifu.core.database import create_engine
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  # noqa: F401
from zhifu.modules.storefront.models import CatalogCategory, CatalogProduct


async def main():
    settings = Settings()
    if settings.environment == "production":
        raise ValueError("开发商品只能在 development 或 test 环境创建")
    accounts = json.loads(
        await AsyncPath(".work/frontend-accounts.json").read_text(encoding="utf-8")
    )
    tenant_id, store_id = UUID(accounts["tenant_id"]).bytes, UUID(accounts["store_id"]).bytes
    asset_dir = AsyncPath("frontend/public/products")
    await asset_dir.mkdir(parents=True, exist_ok=True)
    products = [
        ("keyboard-01", "keyboard", "办公键盘",
         "适合日常文字输入的桌面键盘。使用前连接电脑并检查输入功能。",
         {"款式": "标准款", "分类": "键盘"}, 17479949,
         "https://www.pexels.com/photo/computer-keyboard-and-a-mouse-on-the-desk-17479949/"),
        ("mouse-01", "mouse", "无线鼠标", "日常桌面操作用鼠标。首次使用请依据随附说明完成连接。",
         {"款式": "标准款", "分类": "鼠标"}, 28297695,
         "https://www.pexels.com/photo/modern-workspace-with-computer-accessories-28297695/"),
        ("desk-01", "desk", "桌面配件", "整理电脑周边的桌面配件，适合搭配键盘和鼠标使用。",
         {"款式": "标准款", "分类": "桌面配件"}, 15372901,
         "https://www.pexels.com/photo/computer-accessories-on-mouse-pad-on-desk-15372901/"),
    ]
    # 下载官方原图资源，不进行图像修改；来源与授权记录保存在仓库文档中。
    async with httpx.AsyncClient(timeout=30, follow_redirects=True) as client:
        for code, _, _, _, _, photo, _ in products:
            target = asset_dir / f"{code}.jpg"
            if not await target.exists():
                response = await client.get(
                    f"https://images.pexels.com/photos/{photo}/pexels-photo-{photo}.jpeg"
                )
                response.raise_for_status()
                if not response.headers.get("content-type", "").startswith("image/"):
                    raise ValueError("图片来源返回了非图片内容")
                await target.write_bytes(response.content)
    credentials_path = AsyncPath(".work/storefront-channel.json")
    credentials_exist = await credentials_path.exists()
    credentials = json.loads(
        await credentials_path.read_text(encoding="utf-8")
    ) if credentials_exist else {
        "tenant_id": accounts["tenant_id"], "username": "channel-service01",
        "password": secrets.token_urlsafe(24), "store_id": accounts["store_id"],
    }
    engine = create_engine(settings)
    sessions = async_sessionmaker(engine, expire_on_commit=False)
    try:
        async with sessions.begin() as session:
            if await session.get(Store, store_id) is None:
                raise ValueError("请先准备前端开发账户")
            categories = [("keyboard", "键盘"), ("mouse", "鼠标"), ("desk", "桌面配件")]
            for position, (code, name) in enumerate(categories):
                if await session.get(CatalogCategory, (tenant_id, store_id, code)) is None:
                    session.add(CatalogCategory(tenant_id=tenant_id, store_id=store_id,
                                                code=code, name=name, position=position))
            await session.flush()
            for code, category, name, description, specifications, _, source in products:
                if await session.get(CatalogProduct, (tenant_id, store_id, code)) is None:
                    session.add(CatalogProduct(
                        tenant_id=tenant_id, store_id=store_id, id=code, category_code=category,
                        name=name, description=description, specifications=specifications,
                        delivery="配送安排以店铺订单约定为准", image_url=f"/products/{code}.jpg",
                        image_source=source, status="ACTIVE",
                    ))
            actor_id = UUID("ac700000-2026-4000-8000-000000000020").bytes
            actor = await session.get(Principal, actor_id)
            if actor is None:
                session.add(Principal(
                    id=actor_id, tenant_id=tenant_id, username=credentials["username"],
                    password_hash=password_hasher.hash(credentials["password"]), kind="SERVICE",
                ))
                await session.flush()
                for permission in ["identities:verify", "tickets:read"]:
                    session.add(StoreGrant(tenant_id=tenant_id, store_id=store_id,
                                           principal_id=actor_id, permission=permission))
            elif not credentials_exist:
                raise ValueError("渠道账户已存在但本地凭据缺失，请恢复原凭据")
            customer_id = UUID("ac700000-2026-4000-8000-000000000021").bytes
            if await session.get(Principal, customer_id) is None:
                session.add(Principal(
                    id=customer_id, tenant_id=tenant_id, username="storefront-customer02",
                    password_hash=password_hasher.hash(accounts["password"]), kind="CUSTOMER",
                ))
                await session.flush()
                for permission in ["orders:read", "tickets:read"]:
                    session.add(StoreGrant(tenant_id=tenant_id, store_id=store_id,
                                           principal_id=customer_id, permission=permission))
        await credentials_path.write_text(
            json.dumps(credentials, ensure_ascii=False, indent=2), encoding="utf-8"
        )
        print("3 个分类、3 个商品、渠道服务和隔离验证客户已保存；保留已有经营资料。")
    finally:
        await engine.dispose()


asyncio.run(main())