#!/usr/bin/env python3 """ 全链路Mock验证脚本 不依赖真实医药城/合同系统，用mock数据跑通：建单→ORDER_FETCH→对码→CONTRACT_CREATE 验证状态机推进、合同编号回填、前端能观测到next_node 用法： export STATE_CENTER_URL="http://127.0.0.1:8000" export AGENT_TOKEN="dev-agent-token" python run_mock_chain.py """ import json import sys import uuid from datetime import datetime, timedelta from urllib.request import Request, urlopen from urllib.error import HTTPError from urllib.parse import quote def encode_job_id(job_id: str) -> str: """job_id含中文，URL路径段需编码""" return quote(job_id, safe="") STATE_CENTER_URL = "http://127.0.0.1:8000" AGENT_TOKEN = "dev-agent-token" def http_post(path: str, payload: dict, token: str | None = None): """POST请求辅助函数""" url = f"{STATE_CENTER_URL}{path}" headers = {"Content-Type": "application/json"} if token: headers["X-Agent-Token"] = token data = json.dumps(payload, ensure_ascii=False).encode("utf-8") request = Request(url, data=data, headers=headers, method="POST") try: with urlopen(request, timeout=15) as response: raw = response.read().decode("utf-8") return json.loads(raw) if raw else {} except HTTPError as exc: detail = exc.read().decode("utf-8", errors="replace") print(f"❌ HTTP {exc.code}: {detail}") raise def http_get(path: str): """GET请求辅助函数""" url = f"{STATE_CENTER_URL}{path}" request = Request(url, method="GET") try: with urlopen(request, timeout=15) as response: raw = response.read().decode("utf-8") return json.loads(raw) if raw else {} except HTTPError as exc: detail = exc.read().decode("utf-8", errors="replace") print(f"❌ HTTP {exc.code}: {detail}") raise def step(num: int, title: str): print(f"\n{'='*60}") print(f"【第{num}步】{title}") print('='*60) def run_chain(): order_date = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d") # supplier加唯一后缀，保证每次全新job（避免幂等复用旧状态） supplier = f"重庆亿昊药业-{uuid.uuid4().hex[:4]}" source_type = "YYC" # 医药城线上单 # ---- 第1步：建单 ---- step(1, "建单 POST /api/jobs") resp = http_post("/api/jobs", { "source_type": source_type, "order_date": order_date, "supplier_name": supplier, "created_by": "mock-chain", }) job_id = resp["job_id"] print(f"✓ job_id: {job_id}") print(f"✓ first_node: {resp.get('first_node')}") print(f"✓ reused: {resp.get('reused')}") # ---- 第2步：加载fixture货品（不依赖Oracle）---- step(2, "加载测试货品 POST /api/dev/match/load-fixture-goods") resp = http_post("/api/dev/match/load-fixture-goods", {}, token=AGENT_TOKEN) print(f"✓ 已加载货品数: {resp.get('loaded')}") # ---- 第3步：ORDER_FETCH 开始回调 ---- step(3, "ORDER_FETCH start 回调") resp = http_post( f"/api/jobs/{encode_job_id(job_id)}/nodes/ORDER_FETCH/start", {"attemptNo": 1, "executorPlatform": "mock-rpa", "operator": "mock-chain"}, token=AGENT_TOKEN, ) print(f"✓ start响应: {resp}") # ---- 第4步：写订单明细（模拟ORDER_FETCH导出的数据）---- step(4, "写订单明细 POST /api/jobs/{job_id}/items") order_no = f"DD-{order_date.replace('-', '')}-001" resp = http_post( f"/api/jobs/{encode_job_id(job_id)}/items", { "replace": True, "rows": [ { "orderNo": order_no, "rawProductName": "阿卡波糖片", "platformProductCode": "YP-001", "spec": "50mg*30片", "manufacturer": "示例制药", "approvalNo": "国药准字H00000001", "qty": 100, "unitPrice": 25.5, "sourceRowNo": 1, }, { "orderNo": order_no, "rawProductName": "阿莫西林胶囊", "platformProductCode": "YP-002", "spec": "0.25g*24粒", "manufacturer": "示例医药", "approvalNo": "国药准字H00000002", "qty": 50, "unitPrice": 18.0, "sourceRowNo": 2, }, ], }, token=AGENT_TOKEN, ) print(f"✓ 写入明细: {resp.get('inserted')} 行") print(f"✓ 订单号: {resp.get('order_nos')}") # ---- 第5步：ORDER_FETCH 完成回调 ---- step(5, "ORDER_FETCH complete 回调") resp = http_post( f"/api/jobs/{encode_job_id(job_id)}/nodes/ORDER_FETCH/complete", { "attemptNo": 1, "executorRunId": f"MOCK-{uuid.uuid4().hex[:8]}", "summary": { "orderCount": 1, "supplierList": [supplier], "orderNoList": [order_no], }, "artifacts": [ { "artifactType": "SOURCE_EXCEL", "storagePath": f"/tmp/mock_rpa_data/orders_{supplier}_{order_date}.xlsx", "fileHash": "sha256:mock-order-excel", } ], }, token=AGENT_TOKEN, ) print(f"✓ next_node: {resp.get('next_node')}") # ---- 第6步：跑对码 ---- step(6, "对码 POST /api/jobs/{job_id}/match/run") resp = http_post(f"/api/jobs/{encode_job_id(job_id)}/match/run", {}, token=AGENT_TOKEN) print(f"✓ 对码结果: {json.dumps(resp, ensure_ascii=False)}") # ---- 第7步：查当前状态，判断是否需要人工对码 ---- step(7, "查单据状态 GET /api/jobs/{job_id}") detail = http_get(f"/api/jobs/{encode_job_id(job_id)}") job = detail.get("job", {}) print(f"✓ 当前节点: {job.get('current_node_code')}") print(f"✓ 当前状态: {job.get('status')}") tasks = detail.get("tasks", []) if tasks: print(f"⚠ 有 {len(tasks)} 条待办需要人工处理") for t in tasks: print(f" - {t.get('title')} [{t.get('reason_type')}]") # ---- 第7.5步：如卡在SOURCE_ADD，模拟人工完成新增货源推进 ---- if job.get("current_node_code") == "SOURCE_ADD": step("7.5", "SOURCE_ADD complete（模拟人工新增货源完成）") resp = http_post( f"/api/jobs/{encode_job_id(job_id)}/nodes/SOURCE_ADD/complete", { "attemptNo": 1, "executorRunId": f"MOCK-SRC-{uuid.uuid4().hex[:8]}", "operator": "mock-chain", "summary": {"sourceAdded": True}, "artifacts": [], }, token=AGENT_TOKEN, ) print(f"✓ next_node: {resp.get('next_node')}") # ---- 第8步：CONTRACT_CREATE mock（直接传假合同编号）---- step(8, "CONTRACT_CREATE mock（传假合同编号，跳过真实页面）") mock_contract_no = f"HT{order_date.replace('-', '')}{uuid.uuid4().hex[:5].upper()}" print(f" 使用mock合同编号: {mock_contract_no}") # 先start http_post( f"/api/jobs/{encode_job_id(job_id)}/nodes/CONTRACT_CREATE/start", {"attemptNo": 1, "executorPlatform": "mock-rpa", "operator": "mock-chain"}, token=AGENT_TOKEN, ) # 再complete，带businessRefs.contractSystemNo resp = http_post( f"/api/jobs/{encode_job_id(job_id)}/nodes/CONTRACT_CREATE/complete", { "attemptNo": 1, "executorRunId": f"MOCK-CONTRACT-{uuid.uuid4().hex[:8]}", "businessRefs": {"contractSystemNo": mock_contract_no}, "summary": { "supplierName": supplier, "contractDate": order_date, "sourceType": source_type, "uploadRule": "merge_pdf_before_upload", }, "artifacts": [], }, token=AGENT_TOKEN, ) print(f"✓ next_node: {resp.get('next_node')}") # ---- 第9步：验证合同编号回填 ---- step(9, "验证合同编号回填 GET /api/jobs/{job_id}") detail = http_get(f"/api/jobs/{encode_job_id(job_id)}") job = detail.get("job", {}) print(f"✓ 当前节点: {job.get('current_node_code')}") print(f"✓ 当前状态: {job.get('status')}") contract_no = job.get("contract_system_no") if contract_no == mock_contract_no: print(f"✅ 合同编号回填成功: {contract_no}") else: print(f"⚠ 合同编号: {contract_no}（期望 {mock_contract_no}）") print(f"\n{'='*60}") print(f"🎉 Mock链跑完！job_id = {job_id}") print(f" 前端观测: http://localhost:5173") print('='*60) return job_id if __name__ == "__main__": try: run_chain() except Exception as e: print(f"\n❌ 链路中断: {e}") sys.exit(1)