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))