zhifu-platform/scripts/prepare_browser_waiting.py

import argparse
import asyncio
import hashlib
import json
import secrets
from datetime import timedelta
from pathlib import Path
from uuid import UUID, uuid4

from langchain_core.messages import AIMessage, message_to_dict
from sqlalchemy import select, update

from zhifu.app import create_app
from zhifu.core.config import Settings
from zhifu.modules.after_sales.models import OrderProjection
from zhifu.modules.agent_runtime.config import GRAPH_VERSION
from zhifu.modules.agent_runtime.models import AgentRun
from zhifu.modules.agent_runtime.runtime import build_tool_graph
from zhifu.modules.conversations.models import Conversation
from zhifu.modules.identity.access_models import AccessSession, Principal, StoreGrant
from zhifu.modules.identity.auth import now, password_hasher

OUTPUT = Path(".work/frontend-waiting.json")


async def main(cleanup=False, business=False):
    output = Path(".work/frontend-business-waiting.json") if business else OUTPUT
    settings = Settings()
    if settings.environment == "production":
        raise RuntimeError("浏览器验证数据只能在独立开发环境准备")
    app = create_app(settings)
    async with app.router.lifespan_context(app):
        if cleanup:
            data = json.loads(await asyncio.to_thread(output.read_text, encoding="utf-8"))
            async with app.state.sessions.begin() as session:
                row = await session.get(
                    Principal, UUID(data["actor_id"]).bytes, with_for_update=True)
                if row is None or not row.username.startswith("waiting-test-"):
                    raise RuntimeError("待清理账户不属于本专项")
                row.status = "SUSPENDED"
                await session.execute(update(AccessSession).where(
                    AccessSession.principal_id == row.id).values(revoked=True))
            await asyncio.to_thread(output.unlink)
            print("浏览器等待输入专项账户已停用，会话已撤销，临时凭据已删除")
            return
        if await asyncio.to_thread(output.exists):
            raise RuntimeError("请先清理已有浏览器等待输入专项账户")
        data = json.loads(await asyncio.to_thread(
            Path(".work/frontend-accounts.json").read_text, encoding="utf-8"))
        tenant_id, store_id = UUID(data["tenant_id"]), UUID(data["store_id"])
        actor, conversation_id, run_id, order_id = uuid4(), uuid4(), uuid4(), uuid4()
        username = "waiting-test-" + actor.hex[:12]
        password = secrets.token_urlsafe(24)
        token_hash = hashlib.sha256(secrets.token_bytes(32)).digest()
        question = "请补充键盘故障原因与申请数量。"
        tool_name = "create_exchange_proposal" if business else "request_clarification"
        tool_args = {
            "order_id": str(order_id), "item_id": "keyboard-1", "quantity": 1, "reason": "按键失灵",
        } if business else {
            "question": question, "missing_fields": ["quantity", "reason"],
            "candidate_order_ids": [str(order_id)],
        }
        async with app.state.sessions.begin() as session:
            session.add(Principal(id=actor.bytes, tenant_id=tenant_id.bytes, username=username,
                password_hash=password_hasher.hash(password), kind="CUSTOMER"))
            await session.flush()
            for permission in (
                "orders:read", "tickets:read", "after_sales:create", "after_sales:confirm"
            ):
                session.add(StoreGrant(tenant_id=tenant_id.bytes, principal_id=actor.bytes,
                    store_id=store_id.bytes, permission=permission))
            session.add(AccessSession(token_hash=token_hash, tenant_id=tenant_id.bytes,
                principal_id=actor.bytes, expires_at=now() + timedelta(hours=1), revoked=False))
            session.add(OrderProjection(id=order_id.bytes, tenant_id=tenant_id.bytes,
                store_id=store_id.bytes, subject_id=actor.bytes, source_system="browser-acceptance",
                source_version=1, item_id="keyboard-1", product_type="keyboard",
                available_quantity=1,
                signed_at=now(), observed_at=now()))
            conversation = Conversation(id=conversation_id.bytes, tenant_id=tenant_id.bytes,
                store_id=store_id.bytes, subject_id=actor.bytes, request_key=uuid4().hex,
                created_at=now())
            session.add(conversation)
            await session.flush()
            # 直接执行已注册图的工具节点，浏览器验收不生成或替代模型输出。
            run = AgentRun(id=run_id.bytes, tenant_id=tenant_id.bytes, actor_id=actor.bytes,
                store_id=store_id.bytes, conversation_id=conversation_id.bytes, control_version=1,
                token_hash=token_hash, request_key=uuid4().hex,
                request_hash=hashlib.sha256(b"waiting-test").digest(),
                messages=[message_to_dict(AIMessage(content="", tool_calls=[{
                    "name": tool_name, "args": tool_args,
                    "id": uuid4().hex, "type": "tool_call"}]))],
                graph_version=GRAPH_VERSION, status="RUNNING", next_node="tools", model_name="",
                model_url="", model_credential_env="", lease_token=1,
                lease_until=now() + timedelta(seconds=125), created_at=now(),
                deadline_at=now() + timedelta(minutes=10))
            session.add(run)
            await session.flush()
            conversation.active_run_id = run.id
        await build_tool_graph(app.state.sessions).ainvoke({"run_id": run_id.bytes, "token": 1})
        async with app.state.sessions() as session:
            current = await session.scalar(select(AgentRun).where(AgentRun.id == run_id.bytes))
            if current.status != "WAITING_USER" or current.model_calls != 0:
                raise RuntimeError("工具节点未建立预期等待状态")
        await asyncio.to_thread(output.write_text, json.dumps({
            "tenant_id": str(tenant_id), "store_id": str(store_id),
            "actor_id": str(actor), "username": username, "password": password,
            "conversation_id": str(conversation_id), "run_id": str(run_id),
            "question": current.waiting["question"],
            "proposal_id": current.waiting.get("proposal_id")},
            ensure_ascii=False, indent=2), encoding="utf-8")
        print("浏览器等待输入资料已保存，生成调用0次")


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="准备或清理真实浏览器等待输入验收资料")
    parser.add_argument("--cleanup", action="store_true")
    parser.add_argument("--business", action="store_true")
    options = parser.parse_args()
    asyncio.run(main(options.cleanup, options.business))