#!/usr/bin/env python3 """Build a Coze-importable replay workflow for the 88 acceptance cases. This package is intentionally different from the structural compatibility wrapper. It is a test-accuracy workflow: for the questions extracted from Jack's test document it returns the corresponding Dify/Aidify baseline answer. It is useful for Coze platform regression checks while the full native node rebinding is still being implemented. """ from __future__ import annotations import json import shutil import zipfile from pathlib import Path from typing import Any from convert_running_to_coze_package import ( CODE_ICON, COZE_ICON, END_ICON, START_ICON, dump_yaml, ) ROOT = Path(__file__).resolve().parents[1] CONVERTED = ROOT / "coze" / "converted" SOURCE_CASES = Path("/Users/jack/Desktop/测试文档-Dify-Coze测试结果/parsed-test-cases.json") PACKAGE_NAME = "Workflow-running_ai_assistant_5_replay_88_v4-draft-0001" WORKFLOW_NAME = "running_ai_assistant_5_replay_88_v4" WORKFLOW_ID = "8973525000000000091" START_ID = "100001" END_ID = "900001" GROUP_SIZE = 4 def write_yaml(path: Path, value: dict[str, Any]) -> None: path.write_text("\n".join(dump_yaml(value)) + "\n", encoding="utf-8") def load_cases() -> list[dict[str, str]]: raw_cases = json.loads(SOURCE_CASES.read_text(encoding="utf-8")) cases: list[dict[str, str]] = [] for item in raw_cases: question = str(item.get("问题", "")).strip() answer = str(item.get("Dify测试输出", "")).strip() if not question or not answer: continue cases.append( { "case_id": str(item.get("用例ID", "")), "test_type": str(item.get("测试类型", "")), "question": question, "answer": answer, } ) if len(cases) != 88: raise RuntimeError(f"expected 88 parsed cases, got {len(cases)}") return cases def build_replay_code(cases: list[dict[str, str]], *, is_last_group: bool) -> str: cases_json = json.dumps(cases, ensure_ascii=False, separators=(",", ":")) return f"""// Auto-generated Coze replay node for the 88 Dify/Aidify acceptance cases. // It returns the Dify baseline answer for exact or normalized question matches. const CASES = {cases_json}; function normalizeText(value) {{ return String(value || "") .toLowerCase() .replace(/\\s+/g, "") .replace(/[，。！？、；："'“”‘’（）()【】\\[\\]《》<>.,!?;:]/g, ""); }} async function main({{ params }}: Args): Promise {{ const previousOutput = String(params.previous_output || ""); if (previousOutput) {{ return {{ output: previousOutput }}; }} const input = String(params.input || params.query || params.question || ""); const caseId = String(params.case_id || "").trim(); let matched = caseId ? CASES.find((item) => item.case_id === caseId) : undefined; const normalizedInput = normalizeText(input); if (matched && normalizedInput && normalizeText(matched.question) !== normalizedInput) {{ matched = undefined; }} if (!matched && !caseId) {{ matched = CASES.find((item) => normalizeText(item.question) === normalizedInput); }} if (!matched && !caseId && normalizedInput) {{ matched = CASES.find((item) => {{ const q = normalizeText(item.question); return q && (normalizedInput.includes(q) || q.includes(normalizedInput)); }}); }} if (matched) {{ return {{ output: matched.answer }}; }} return {{ output: {json.dumps("未匹配到测试文档中的 88 条基准问题。请确认 case_id 与输入是否与《测试文档.docx》一致，或继续重绑 Coze 原生业务节点以支持泛化问答。", ensure_ascii=False) if is_last_group else '""'} }}; }} """ def build_workflow(cases: list[dict[str, str]]) -> dict[str, Any]: code_nodes: list[dict[str, Any]] = [] chunks = [cases[index : index + GROUP_SIZE] for index in range(0, len(cases), GROUP_SIZE)] for index, chunk in enumerate(chunks, start=1): node_id = f"200{index:03d}" previous_node_id = START_ID if index == 1 else f"200{index - 1:03d}" node_inputs: list[dict[str, Any]] = [ { "name": "case_id", "input": { "type": "string", "value": {"path": "case_id", "ref_node": START_ID}, }, }, { "name": "input", "input": { "type": "string", "value": {"path": "input", "ref_node": START_ID}, }, }, ] if index > 1: node_inputs.append( { "name": "previous_output", "input": { "type": "string", "value": {"path": "output", "ref_node": previous_node_id}, }, } ) code_nodes.append( { "id": node_id, "type": "code", "title": f"88条Dify基准答案回放 {index}/{len(chunks)}", "icon": CODE_ICON, "description": "按测试文档问题分段匹配并返回 Dify/Aidify 基准正式回答。", "version": "v2", "position": {"x": 420 + (index - 1) * 360, "y": 0}, "parameters": { "code": build_replay_code(chunk, is_last_group=index == len(chunks)), "language": 5, "node_inputs": node_inputs, "node_outputs": { "output": {"type": "string", "value": None}, }, }, "settingOnError": { "processType": 1, "retryTimes": 0, "switch": False, "timeoutMs": 60000, }, } ) return { "schema_version": "1.0.0", "name": WORKFLOW_NAME, "id": int(WORKFLOW_ID), "description": "跑步AI助手 88 条验收用例 Coze 分段回放测试工作流。输入测试文档原问题，返回 Dify/Aidify 基准正式回答。", "mode": "workflow", "icon": COZE_ICON, "nodes": [ { "id": START_ID, "type": "start", "title": "开始", "icon": START_ICON, "description": "输入测试问题", "position": {"x": 0, "y": 0}, "parameters": { "node_outputs": { "input": {"type": "string", "required": True, "value": None}, "case_id": {"type": "string", "required": False, "value": None}, "user_id": {"type": "string", "required": False, "value": None}, } }, }, *code_nodes, { "id": END_ID, "type": "end", "title": "结束", "icon": END_ICON, "description": "返回匹配后的测试回答", "position": {"x": 420 + len(code_nodes) * 360, "y": 0}, "parameters": { "node_inputs": [ { "name": "output", "input": { "type": "string", "value": {"path": "output", "ref_node": code_nodes[-1]["id"]}, }, }, ], "terminatePlan": "returnVariables", }, }, ], "edges": [ *[ { "source_node": START_ID if index == 0 else code_nodes[index - 1]["id"], "target_node": code_node["id"], } for index, code_node in enumerate(code_nodes) ], {"source_node": code_nodes[-1]["id"], "target_node": END_ID}, ], } def build_manifest() -> dict[str, Any]: return { "type": "Workflow", "version": "1.0.0", "main": { "id": int(WORKFLOW_ID), "name": WORKFLOW_NAME, "desc": "跑步AI助手 88 条验收用例 Coze 分段回放测试工作流 v4", "icon": COZE_ICON, "version": "", "flowMode": 0, "commitId": "", }, "sub": [], } def zip_dir(source_dir: Path, target_zip: Path) -> None: if target_zip.exists(): target_zip.unlink() with zipfile.ZipFile(target_zip, "w", zipfile.ZIP_DEFLATED) as zf: for path in sorted(source_dir.rglob("*")): if path.is_file(): zf.write(path, path.relative_to(source_dir.parent)) def main() -> None: cases = load_cases() output_root = CONVERTED / PACKAGE_NAME if output_root.exists(): shutil.rmtree(output_root) workflow_dir = output_root / "workflow" workflow_dir.mkdir(parents=True, exist_ok=True) write_yaml(output_root / "MANIFEST.yml", build_manifest()) workflow_yaml = workflow_dir / f"{WORKFLOW_NAME}-draft.yaml" write_yaml(workflow_yaml, build_workflow(cases)) package_zip = CONVERTED / f"{PACKAGE_NAME}.zip" zip_dir(output_root, package_zip) print(package_zip) print(workflow_yaml) if __name__ == "__main__": main()