zhifu-platform/scripts/prepare_course_demo.py

import argparse
import asyncio
import csv
import hashlib
import json
import math
import os
import re
import secrets
import subprocess
import time
from datetime import UTC, datetime, timedelta
from pathlib import Path
from uuid import UUID, uuid4

import httpx
from dotenv import dotenv_values, set_key
from filelock import FileLock
from sqlalchemy import func, select
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, Ticket
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
from zhifu.modules.integrations.models import Connector, ExternalOperation
from zhifu.modules.storefront.models import CatalogProduct
from zhifu.oms.database import oms_engine
from zhifu.oms.models import Credential

ROOT = Path(__file__).resolve().parents[1]
DATA = ROOT / ".work" / "course-demo"
TENANT = UUID("ac700000-2026-4000-8000-000000000001")
STORE = UUID("ac700000-2026-4000-8000-000000000002")
TOKEN_ENV = "ZHIFU_COURSE_OMS_TOKEN"
CONNECTOR_URL = "http://oms:8100"
CUSTOMER_PERMISSIONS = (
    "orders:read", "tickets:read", "after_sales:create", "after_sales:confirm",
    "after_sales:cancel",
)


def require(condition, message):
    if not condition:
        raise ValueError(message)


def utc_now():
    return datetime.now(UTC).isoformat()


def read_json(path):
    return json.loads(path.read_text(encoding="utf-8"))


def protect_file(path):
    # Windows 使用当前用户 SID 设置访问权限，凭据文件仅当前用户和 SYSTEM 可读取。
    if os.name == "nt":
        result = subprocess.run(
            ["whoami", "/user", "/fo", "csv", "/nh"],
            check=True, capture_output=True, text=True, timeout=10,
        )
        sid = next(csv.reader(result.stdout.strip().splitlines()))[1]
        subprocess.run(
            ["icacls", str(path), "/inheritance:r", "/grant:r",
             f"*{sid}:(F)", "*S-1-5-18:(F)"],
            check=True, capture_output=True, timeout=10,
        )
    else:
        path.chmod(0o600)


def write_json(path, payload, *, private=False):
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = path.with_name(f".{path.name}.{uuid4().hex}.new")
    # 同目录临时文件与原子替换保证中断后仍能读取完整 JSON；临时文件也使用受限权限。
    descriptor = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
    with os.fdopen(descriptor, "w", encoding="utf-8") as stream:
        if private:
            protect_file(temporary)
        json.dump(payload, stream, ensure_ascii=False, indent=2)
        stream.write("\n")
        stream.flush()
        os.fsync(stream.fileno())
    temporary.replace(path)


def session_name(value):
    require(
        re.fullmatch(r"[a-zA-Z0-9][a-zA-Z0-9_-]{0,47}", value) is not None,
        "课程名称须为 1 至 48 个英文字母、数字、下划线或连字符，并以字母或数字开头",
    )
    require(
        value.upper() not in {"CON", "PRN", "AUX", "NUL"}
        and re.fullmatch(r"(?:COM|LPT)[1-9]", value.upper()) is None,
        "课程名称不能使用 Windows 保留文件名",
    )
    return value


def validate_scope(value):
    require(
        value["tenant_id"] == str(TENANT) and value["store_id"] == str(STORE),
        "课程数据仅允许使用固定开发商家和店铺",
    )


def validate_url(value):
    url = httpx.URL(value)
    require(
        url.scheme in {"http", "https"} and url.host in {"127.0.0.1", "localhost", "::1"}
        and not url.username and not url.password and not url.query and not url.fragment
        and url.path in {"", "/"},
        "课程接口地址必须为本机 HTTP 或 HTTPS 服务根地址",
    )
    return str(url).rstrip("/")


def load_manifest(name=None):
    path = DATA / "sessions" / f"{session_name(name)}.json" if name else DATA / "active.json"
    manifest = read_json(path)
    validate_scope(manifest)
    session_name(manifest["session"])
    require(manifest["schema_version"] == 1, "课程记录版本不受支持")
    validate_url(manifest["api_url"])
    validate_url(manifest["oms_url"])
    require(manifest["order"]["item_id"] == "keyboard-01", "课程订单商品不匹配")
    return manifest


def save_manifest(manifest, *, activate=False):
    write_json(DATA / "sessions" / f"{manifest['session']}.json", manifest)
    active = DATA / "active.json"
    if activate or (active.exists() and read_json(active)["session"] == manifest["session"]):
        write_json(active, manifest)


def credentials(accounts=None):
    path = DATA / "credentials.json"
    if path.exists():
        value = read_json(path)
        validate_scope(value)
        require(value["schema_version"] == 1, "课程凭据版本不受支持")
        protect_file(path)
        return value
    require(accounts is not None, "课程凭据不存在，请执行 prepare")
    value = {
        "schema_version": 1, "tenant_id": str(TENANT), "store_id": str(STORE),
        "customer_password": accounts["password"],
        "service": {
            "principal_id": str(uuid4()), "username": "course-order-sync",
            "password": secrets.token_urlsafe(32),
        },
        "oms_platform_token": secrets.token_urlsafe(32),
        "oms_fulfillment_token": secrets.token_urlsafe(32),
    }
    # 先保存凭据，再创建数据库记录；后续显式重试继续使用相同身份与令牌。
    write_json(path, value, private=True)
    return value


def prepare_files(args):
    accounts = read_json(ROOT / ".work" / "frontend-accounts.json")
    validate_scope(accounts)
    name = session_name(args.session or datetime.now().strftime("%Y%m%d-%H%M%S-%f"))
    path = DATA / "sessions" / f"{name}.json"
    secret = credentials(accounts)
    if path.exists():
        manifest = load_manifest(name)
        require(
            manifest["api_url"] == args.api_url and manifest["oms_url"] == args.oms_url,
            "已有课程接口地址与本次参数不同，请使用原有地址",
        )
    else:
        manifest = {
            "schema_version": 1, "session": name, "tenant_id": str(TENANT),
            "store_id": str(STORE), "created_at": utc_now(), "status": "PREPARING",
            "api_url": args.api_url, "oms_url": args.oms_url,
            "customer": {"principal_id": str(uuid4()), "username": f"course-{name}"},
            "order": {
                "order_id": str(uuid4()), "item_id": "keyboard-01", "product_type": "keyboard",
                "available_quantity": 1, "source_version": 1,
                "signed_at": (datetime.now(UTC) - timedelta(days=2)).isoformat(),
            },
            "staff": {name: identifier for name, identifier in accounts["actors"].items()
                      if name in {"staff01", "supervisor01", "admin01"}},
        }
        save_manifest(manifest)
    return manifest, secret


async def ensure_principal(session, actor, password, kind, permissions):
    identifier = UUID(actor["principal_id"]).bytes
    existing = await session.get(Principal, identifier, with_for_update=True)
    if existing is None:
        collision = await session.scalar(select(Principal).where(
            Principal.tenant_id == TENANT.bytes, Principal.username == actor["username"],
        ))
        require(collision is None, "课程账户名已经被其他身份使用")
        session.add(Principal(
            id=identifier, tenant_id=TENANT.bytes, username=actor["username"],
            password_hash=await asyncio.to_thread(password_hasher.hash, password), kind=kind,
        ))
        await session.flush()
        for permission in permissions:
            session.add(StoreGrant(
                tenant_id=TENANT.bytes, store_id=STORE.bytes,
                principal_id=identifier, permission=permission,
            ))
    else:
        require(
            existing.tenant_id == TENANT.bytes and existing.username == actor["username"]
            and existing.kind == kind and existing.status == "ACTIVE",
            "已有课程账户身份或状态与课程配置不一致",
        )
        require(
            await asyncio.to_thread(password_hasher.verify, password, existing.password_hash),
            "已有课程账户密码与本地受保护凭据不一致",
        )
        granted = set((await session.scalars(select(StoreGrant.permission).where(
            StoreGrant.tenant_id == TENANT.bytes, StoreGrant.store_id == STORE.bytes,
            StoreGrant.principal_id == identifier,
        ))).all())
        require(granted == set(permissions), "已有课程账户授权与课程要求不一致")


async def bootstrap(settings, manifest, secret):
    engine = create_engine(settings)
    oms = oms_engine()
    try:
        sessions = async_sessionmaker(engine, expire_on_commit=False)
        async with sessions.begin() as session:
            tenant = await session.get(Tenant, TENANT.bytes)
            store = await session.get(Store, STORE.bytes, with_for_update=True)
            require(tenant is not None and tenant.status == "ACTIVE", "开发商家未启用")
            require(
                store is not None and store.tenant_id == TENANT.bytes and store.status == "ACTIVE",
                "开发店铺未启用或不属于指定商家",
            )
            product = await session.get(CatalogProduct, (TENANT.bytes, STORE.bytes, "keyboard-01"))
            require(product is not None and product.status == "ACTIVE", "键盘商品未发布")
            rule = await session.scalar(select(ExchangeRule).where(
                ExchangeRule.tenant_id == TENANT.bytes, ExchangeRule.store_id == STORE.bytes,
            ).order_by(ExchangeRule.version.desc()).limit(1))
            require(
                rule is not None and not rule.revoked and rule.window_days > 2
                and "keyboard" in rule.product_types,
                "当前规则不支持签收两天的键盘订单，请检查已发布规则",
            )
            manifest["rule"] = {"version": rule.version, "window_days": rule.window_days,
                                "citation": rule.citation}
            manifest["product_name"] = product.name
            manifest["existing_ticket_count"] = await session.scalar(
                select(func.count()).select_from(Ticket).where(
                    Ticket.tenant_id == TENANT.bytes,
                    Ticket.order_id == UUID(manifest["order"]["order_id"]).bytes,
                )
            )
            await ensure_principal(
                session, manifest["customer"], secret["customer_password"],
                "CUSTOMER", CUSTOMER_PERMISSIONS,
            )
            await ensure_principal(
                session, secret["service"], secret["service"]["password"],
                "SERVICE", ("orders:sync",),
            )
            connector = await session.get(Connector, (TENANT.bytes, STORE.bytes))
            if connector is None:
                session.add(Connector(
                    tenant_id=TENANT.bytes, store_id=STORE.bytes, source_system="oms",
                    base_url=CONNECTOR_URL, credential_env=TOKEN_ENV, enabled=True,
                ))
                session.add(BusinessEvent(
                    id=uuid4().bytes, tenant_id=TENANT.bytes,
                    actor_id=UUID(secret["service"]["principal_id"]).bytes,
                    aggregate_id=STORE.bytes, aggregate_version=1,
                    event_type="connector.configured", request_id=str(uuid4()),
                    payload={"credential_env": TOKEN_ENV, "source": "prepare_course_demo"},
                    occurred_at=datetime.now(UTC).replace(tzinfo=None),
                ))
            else:
                require(
                    connector.source_system == "oms" and connector.base_url == CONNECTOR_URL
                    and connector.credential_env == TOKEN_ENV and connector.enabled,
                    "已有 OMS 连接器配置与课程环境不一致",
                )
        oms_sessions = async_sessionmaker(oms, expire_on_commit=False)
        async with oms_sessions.begin() as session:
            for name, can_fulfill in (
                ("oms_platform_token", False), ("oms_fulfillment_token", True),
            ):
                token_hash = hashlib.sha256(secret[name].encode()).digest()
                existing = await session.get(Credential, token_hash, with_for_update=True)
                if existing is None:
                    session.add(Credential(
                        token_hash=token_hash, tenant_id=TENANT.bytes, store_id=STORE.bytes,
                        can_fulfill=can_fulfill, enabled=True,
                    ))
                else:
                    require(
                        existing.tenant_id == TENANT.bytes and existing.store_id == STORE.bytes
                        and existing.can_fulfill == can_fulfill and existing.enabled,
                        "已有 OMS 凭据权限与课程配置不一致",
                    )
    finally:
        await engine.dispose()
        await oms.dispose()


def configure_runtime(secret):
    path = ROOT / ".work" / "runtime.env"
    require(path.is_file(), "完整项目的 .work/runtime.env 不存在")
    existing = dotenv_values(path).get(TOKEN_ENV)
    require(
        existing is None or existing == secret["oms_platform_token"],
        "运行配置已存在不同的课程 OMS 凭据",
    )
    if existing is None:
        set_key(path, TOKEN_ENV, secret["oms_platform_token"], quote_mode="always")


async def request(client, method, url, *, token=None, body=None, key=None, allow_missing=False):
    headers = {}
    if token:
        headers["Authorization"] = f"Bearer {token}"
    if key:
        headers["Idempotency-Key"] = key
    response = await client.request(method, url, headers=headers, json=body)
    if allow_missing and response.status_code == 404:
        return None
    response.raise_for_status()
    return response.json()


async def login(client, manifest, actor, password):
    result = await request(client, "POST", manifest["api_url"] + "/openapi/v1/auth/sessions", body={
        "tenant_id": str(TENANT), "username": actor["username"], "password": password,
    })
    return result["access_token"], time.monotonic() + result["expires_in"] - 60


def verify_snapshot(manifest, snapshot):
    expected = manifest["order"]
    require(
        snapshot["tenant_id"] == str(TENANT) and snapshot["store_id"] == str(STORE)
        and snapshot["subject_id"] == manifest["customer"]["principal_id"]
        and snapshot["order_id"] == expected["order_id"]
        and snapshot["item_id"] == "keyboard-01" and snapshot["product_type"] == "keyboard"
        and datetime.fromisoformat(snapshot["signed_at"]) == datetime.fromisoformat(
            expected["signed_at"]),
        "OMS 订单身份、商品或签收时间与课程记录不一致",
    )
    require(
        snapshot["source_version"] >= 1 and 0 <= snapshot["available_quantity"] <= 1,
        "OMS 订单版本或数量不符合课程订单约束",
    )


async def synchronize(client, manifest, secret, token):
    snapshot = await request(
        client, "GET", manifest["oms_url"] + "/orders/" + manifest["order"]["order_id"],
        token=secret["oms_platform_token"],
    )
    verify_snapshot(manifest, snapshot)
    body = {name: snapshot[name] for name in (
        "store_id", "subject_id", "source_version", "item_id", "product_type",
        "available_quantity", "signed_at",
    )}
    body["source_system"] = "oms"
    # 每次可信观察使用新的幂等键，确保相同来源版本也更新 observed_at。
    await request(
        client, "PUT", manifest["api_url"] + "/internal/v1/orders/" + snapshot["order_id"],
        token=token, body=body, key=f"course-sync-{uuid4().hex}",
    )
    manifest["last_sync_at"] = utc_now()
    manifest["last_order_snapshot"] = snapshot
    return snapshot


async def prepare(settings, args):
    async with asyncio.timeout(90):
        manifest, secret = await asyncio.to_thread(prepare_files, args)
        await bootstrap(settings, manifest, secret)
        await asyncio.to_thread(configure_runtime, secret)
        async with httpx.AsyncClient(timeout=10, trust_env=False) as client:
            order_url = manifest["oms_url"] + "/orders/" + manifest["order"]["order_id"]
            existing = await request(client, "GET", order_url,
                                     token=secret["oms_platform_token"], allow_missing=True)
            if existing is None:
                body = {name: value for name, value in manifest["order"].items()
                        if name != "order_id"}
                body["subject_id"] = manifest["customer"]["principal_id"]
                await request(client, "PUT", order_url,
                              token=secret["oms_fulfillment_token"], body=body)
            else:
                verify_snapshot(manifest, existing)
            service_token, _ = await login(
                client, manifest, secret["service"], secret["service"]["password"],
            )
            snapshot = await synchronize(client, manifest, secret, service_token)
            customer_token, _ = await login(
                client, manifest, manifest["customer"], secret["customer_password"],
            )
            order = await request(
                client, "GET", manifest["api_url"] + "/openapi/v1/orders/" + snapshot["order_id"],
                token=customer_token,
            )
            require(
                order["subject_id"] == manifest["customer"]["principal_id"]
                and order["source_system"] == "oms"
                and order["source_version"] == snapshot["source_version"]
                and order["available_quantity"] == snapshot["available_quantity"],
                "客户读取的订单与 OMS 当前订单不一致",
            )
            conversation = await request(
                client, "POST", manifest["api_url"] + "/openapi/v1/conversations",
                token=customer_token, body={"store_id": str(STORE)},
                key=f"course-conversation-{manifest['customer']['principal_id']}",
            )
            manifest["conversation_id"] = conversation["conversation_id"]
            unused = (
                conversation["sequence"] == 0 and manifest["existing_ticket_count"] == 0
                and snapshot["source_version"] == 1 and snapshot["available_quantity"] == 1
            )
            if manifest["status"] != "OMS_COMPLETED":
                manifest["status"] = "READY" if unused else "IN_USE"
            manifest["verification"] = {
                "verified_at": utc_now(), "customer_order_read": True,
                "conversation_sequence": conversation["sequence"],
                "generation_model_calls": 0,
            }
        await asyncio.to_thread(save_manifest, manifest, activate=True)
        return manifest


async def sync_loop(args):
    manifest = await asyncio.to_thread(load_manifest, args.session)
    secret = await asyncio.to_thread(credentials)
    duration = args.duration_seconds
    deadline = time.monotonic() + duration
    maximum = max(1, math.ceil(duration / args.interval_seconds))
    token, expires = None, 0
    count = 0
    async with (
        asyncio.timeout(duration + 35),
        httpx.AsyncClient(timeout=10, trust_env=False) as client,
    ):
        for index in range(maximum):
            if index and time.monotonic() >= deadline:
                break
            # 锁等待计入本次同步的 30 秒期限，短时间重叠的课程操作允许等待完成。
            lock = FileLock(DATA / "course.lock", timeout=10, thread_local=False)
            async with asyncio.timeout(30):
                await asyncio.to_thread(lock.acquire)
                try:
                    manifest = await asyncio.to_thread(load_manifest, manifest["session"])
                    if token is None or time.monotonic() >= expires:
                        token, expires = await login(
                            client, manifest, secret["service"], secret["service"]["password"],
                        )
                    await synchronize(client, manifest, secret, token)
                    await asyncio.to_thread(save_manifest, manifest)
                finally:
                    await asyncio.to_thread(lock.release)
            count += 1
            print(json.dumps({
                "session": manifest["session"], "sync_count": count,
                "last_sync_at": manifest["last_sync_at"],
                "source_version": manifest["last_order_snapshot"]["source_version"],
                "available_quantity": manifest["last_order_snapshot"]["available_quantity"],
            }, ensure_ascii=False), flush=True)
            if index + 1 < maximum:
                remaining = deadline - time.monotonic()
                if remaining <= 0:
                    break
                await asyncio.sleep(min(args.interval_seconds, remaining))
    return {"session": manifest["session"], "sync_count": count, "generation_model_calls": 0}


async def complete(settings, args):
    async with asyncio.timeout(90):
        manifest = await asyncio.to_thread(load_manifest, args.session)
        secret = await asyncio.to_thread(credentials)
        engine = create_engine(settings)
        try:
            sessions = async_sessionmaker(engine, expire_on_commit=False)
            async with sessions() as session:
                query = select(ExternalOperation).where(
                    ExternalOperation.tenant_id == TENANT.bytes,
                    ExternalOperation.store_id == STORE.bytes,
                    ExternalOperation.order_id == UUID(manifest["order"]["order_id"]).bytes,
                    ExternalOperation.status.in_(("ACCEPTED", "COMPLETED")),
                )
                if args.operation_id:
                    query = query.where(ExternalOperation.id == args.operation_id.bytes)
                operations = (await session.scalars(query)).all()
                require(
                    len(operations) == 1,
                    "需要唯一的本课程已受理履约操作，请检查审批与履约进度",
                )
                operation_id = str(UUID(bytes=operations[0].id))
                ticket_id = str(UUID(bytes=operations[0].ticket_id))
        finally:
            await engine.dispose()
        async with httpx.AsyncClient(timeout=10, trust_env=False) as client:
            url = manifest["oms_url"] + "/exchanges/" + operation_id
            existing = await request(client, "GET", url, token=secret["oms_fulfillment_token"])
            require(
                existing["order_id"] == manifest["order"]["order_id"]
                and existing["status"] in {"ACCEPTED", "COMPLETED"},
                "OMS 操作不属于本课程订单或当前状态不能完成",
            )
            result = await request(client, "POST", url + "/complete",
                                   token=secret["oms_fulfillment_token"])
            require(result["status"] == "COMPLETED", "OMS 未返回履约完成状态")
            manifest["completion"] = {**result, "ticket_id": ticket_id, "completed_at": utc_now()}
            manifest["status"] = "OMS_COMPLETED"
            service_token, _ = await login(
                client, manifest, secret["service"], secret["service"]["password"],
            )
            await synchronize(client, manifest, secret, service_token)
        await asyncio.to_thread(save_manifest, manifest)
        return {
            "session": manifest["session"], **manifest["completion"], "generation_model_calls": 0,
        }


async def locked_command(settings, args):
    # 跨线程共享同一个锁状态；准备与完成的 90 秒期限包含最长 10 秒的锁等待。
    lock = FileLock(DATA / "course.lock", timeout=10, thread_local=False)
    async with asyncio.timeout(90):
        await asyncio.to_thread(lock.acquire)
        try:
            if args.command == "prepare":
                return await prepare(settings, args)
            return await complete(settings, args)
        finally:
            await asyncio.to_thread(lock.release)


def main():
    parser = argparse.ArgumentParser(description="准备、同步和完成完整项目的课程演示订单")
    commands = parser.add_subparsers(dest="command", required=True)
    prepare_parser = commands.add_parser("prepare", help="创建本次课程客户、OMS 订单与空会话")
    prepare_parser.add_argument("--session", type=session_name)
    prepare_parser.add_argument("--api-url", type=validate_url, default="http://127.0.0.1:28000")
    prepare_parser.add_argument("--oms-url", type=validate_url, default="http://127.0.0.1:28100")
    sync_parser = commands.add_parser("sync", help="从 OMS 读取并同步当前课程订单")
    sync_parser.add_argument("--session", type=session_name)
    sync_parser.add_argument("--duration-seconds", type=int, default=0)
    sync_parser.add_argument("--interval-seconds", type=int, default=60)
    complete_parser = commands.add_parser("complete", help="使用仓储身份登记本课程履约完成")
    complete_parser.add_argument("--session", type=session_name)
    complete_parser.add_argument("--operation-id", type=UUID)
    args = parser.parse_args()
    settings = Settings()
    require(settings.environment in {"development", "test"}, "课程工具仅允许 development 或 test")
    require(Path.cwd().resolve() == ROOT, "请在完整项目根目录执行课程工具")
    DATA.mkdir(parents=True, exist_ok=True)
    if args.command == "sync":
        require(0 <= args.duration_seconds <= 43200, "同步时长须为 0 至 43200 秒")
        require(30 <= args.interval_seconds <= 240, "同步间隔须为 30 至 240 秒")
        result = asyncio.run(sync_loop(args))
    else:
        result = asyncio.run(locked_command(settings, args))
    print(json.dumps(result, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()