#!/usr/bin/env python3
"""Build the local replication package for the Dify running assistant workflow."""

from __future__ import annotations

import csv
import html
import json
import re
from collections import Counter, defaultdict
from pathlib import Path


ROOT = Path(__file__).resolve().parents[1]
RAW = ROOT / "raw"
DOCS = ROOT / "docs"
FIXTURES = ROOT / "fixtures"
TESTS = ROOT / "tests"
REPORTS = ROOT / "reports"


def read_json(name: str):
    return json.loads((RAW / name).read_text(encoding="utf-8"))


def selector_label(selector):
    if not selector:
        return ""
    return ".".join(str(part) for part in selector)


def node_config_summary(node):
    data = node["data"]
    node_type = data["type"]
    summary = {}
    if node_type == "llm":
        summary["model"] = data.get("model", {}).get("name")
        summary["provider"] = data.get("model", {}).get("provider")
        summary["temperature"] = data.get("model", {}).get("completion_params", {}).get("temperature")
        summary["prompt_roles"] = ",".join(t.get("role", "") for t in data.get("prompt_template", []))
    elif node_type == "code":
        summary["language"] = data.get("code_language")
        summary["inputs"] = ",".join(v.get("variable", "") for v in data.get("variables", []))
        summary["outputs"] = ",".join(data.get("outputs", {}).keys())
    elif node_type == "if-else":
        summary["cases"] = str(len(data.get("cases", [])))
        summary["conditions"] = str(sum(len(c.get("conditions", [])) for c in data.get("cases", [])))
    elif node_type == "http-request":
        summary["method"] = data.get("method")
        summary["url"] = data.get("url")
        summary["error_strategy"] = data.get("error_strategy")
    elif node_type == "knowledge-retrieval":
        summary["datasets"] = ",".join(data.get("dataset_ids", []))
        summary["retrieval_mode"] = data.get("retrieval_mode")
        summary["query"] = selector_label(data.get("query_variable_selector"))
    elif node_type == "answer":
        summary["answer"] = data.get("answer", "")[:120].replace("\n", " ")
    elif node_type == "assigner":
        summary["items"] = str(len(data.get("items", [])))
    elif node_type == "variable-aggregator":
        summary["variables"] = ",".join(selector_label(v) for v in data.get("variables", []))
    elif node_type == "start":
        summary["variables"] = str(len(data.get("variables", [])))
    return summary


def build_normalized(workflow, app, system_vars, conversation_vars, variable_pages):
    graph = workflow["graph"]
    nodes = graph["nodes"]
    edges = graph["edges"]
    nodes_by_id = {n["id"]: n for n in nodes}
    variable_items = []
    for page in variable_pages:
        variable_items.extend(page.get("items", []))

    return {
        "source_url": "https://aidifysecond.chinacpt.com/app/897352d5-faea-4467-9392-8cf70323c764/workflow",
        "captured_at": "2026-07-01T16:00:00+08:00",
        "app": {
            "id": app["id"],
            "name": app["name"],
            "mode": app["mode"],
            "icon": app.get("icon"),
            "icon_background": app.get("icon_background"),
        },
        "workflow": {
            "id": workflow["id"],
            "version": workflow.get("version"),
            "hash": workflow.get("hash"),
            "node_count": len(nodes),
            "edge_count": len(edges),
            "node_type_counts": dict(Counter(n["data"]["type"] for n in nodes)),
            "environment_variables": workflow.get("environment_variables", []),
            "conversation_variables": workflow.get("conversation_variables", []),
            "features": workflow.get("features", {}),
        },
        "nodes": [
            {
                "id": n["id"],
                "title": n["data"].get("title"),
                "type": n["data"].get("type"),
                "position": n.get("position"),
                "size": {"width": n.get("width"), "height": n.get("height")},
                "config_summary": node_config_summary(n),
            }
            for n in nodes
        ],
        "edges": [
            {
                "id": e.get("id"),
                "source": e.get("source"),
                "source_title": nodes_by_id.get(e.get("source"), {}).get("data", {}).get("title"),
                "source_handle": e.get("sourceHandle"),
                "target": e.get("target"),
                "target_title": nodes_by_id.get(e.get("target"), {}).get("data", {}).get("title"),
                "target_handle": e.get("targetHandle"),
            }
            for e in edges
        ],
        "registries": {
            "system_variables": system_vars.get("items", []),
            "conversation_variables_api": conversation_vars.get("items", []),
            "node_variables": variable_items,
        },
    }


def write_csv(path, rows, fields):
    with path.open("w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=fields, lineterminator="\n")
        writer.writeheader()
        for row in rows:
            writer.writerow({field: row.get(field, "") for field in fields})


def render_markdown(normalized, workflow):
    type_counts = normalized["workflow"]["node_type_counts"]
    node_rows = "\n".join(
        f"| `{n['id']}` | {n['type']} | {n['title']} | {json.dumps(n['config_summary'], ensure_ascii=False)} |"
        for n in normalized["nodes"]
    )
    edge_rows = "\n".join(
        f"| {i + 1} | `{e['source']}` {e['source_title']} | `{e['source_handle'] or ''}` | `{e['target']}` {e['target_title']} | `{e['target_handle'] or ''}` |"
        for i, e in enumerate(normalized["edges"])
    )
    env_rows = "\n".join(
        f"| `{v['name']}` | {v.get('value_type')} | {v.get('description','')} | `{v.get('value','')}` |"
        for v in normalized["workflow"]["environment_variables"]
    )
    conv_rows = "\n".join(
        f"| `{v['name']}` | {v.get('value_type')} | `{selector_label(v.get('selector'))}` | `{json.dumps(v.get('value'), ensure_ascii=False)}` |"
        for v in normalized["workflow"]["conversation_variables"]
    )
    return f"""# Dify 跑步AI助手-5.0 复刻包

## 目标

本文件用于在新的 Dify / Coze / N8N / 其他编排工具中复刻当前页面：

`{normalized['source_url']}`

抓取方式：通过当前 Chrome 登录态调用 Dify 控制台只读接口，核心接口为 `/workflows/draft` 和 `/export?include_secret=false`。本包不依赖截图复刻，截图只作为人工视觉校验。

## 交付结构

- `raw/workflow-draft.json`：Dify 草稿 workflow 原始 JSON，复刻主真源。
- `raw/dify-export-no-secret.yml`：Dify 官方导出 DSL，可优先尝试导入同版本 Dify。
- `fixtures/workflow-normalized.json`：面向迁移和测试的标准化摘要。
- `fixtures/node-catalog.csv`：节点清单。
- `fixtures/edge-catalog.csv`：边清单。
- `tests/test_workflow_contract_unittest.py`：零依赖标准库单元测试，展开 798 个独立 case。
- `tests/test_workflow_contract.py`：pytest 版本测试，适合已安装 pytest 的环境。

## 应用元信息

- 应用 ID：`{normalized['app']['id']}`
- 应用名称：{normalized['app']['name']}
- 模式：`{normalized['app']['mode']}`
- Workflow ID：`{normalized['workflow']['id']}`
- 节点数：{normalized['workflow']['node_count']}
- 连线数：{normalized['workflow']['edge_count']}
- 节点类型分布：`{json.dumps(type_counts, ensure_ascii=False)}`

## 复刻优先级

1. 同版本 Dify：直接导入 `raw/dify-export-no-secret.yml`。
2. 其他 Dify 环境但导入失败：按 `raw/workflow-draft.json` 重建节点和边，使用 `fixtures/node-catalog.csv`、`fixtures/edge-catalog.csv` 校对。
3. Coze / N8N：先按节点类型分组迁移确定性节点，再替换 LLM、知识库、HTTP、会话变量能力。

## 必须先配置的环境变量

| 变量 | 类型 | 描述 | 当前导出值 |
| --- | --- | --- | --- |
{env_rows}

注意：`PROFILE_API_KEY` 在导出文件中为掩码，迁移后必须由管理员重新填入真实值。

## 会话变量

| 变量 | 类型 | Selector | 初始值 |
| --- | --- | --- | --- |
{conv_rows}

## 节点清单

| ID | 类型 | 标题 | 关键参数摘要 |
| --- | --- | --- | --- |
{node_rows}

## 连线清单

| # | Source | Source Handle | Target | Target Handle |
| --- | --- | --- | --- | --- |
{edge_rows}

## 迁移拆解

### 输入与路由

从 `用户输入` 开始，经过 `用户输入提取 / 路由解析`、`路由 JSON 清洗 / 字段标准化`、`入口兼容字段整理` 和 `会话变量更新`。复刻时必须保证 LLM 路由节点只输出 JSON，后续 code 节点负责清洗、兜底、标准化字段。

### 安全与超范围

`主分流` 将 `discomfort_fixed_reply`、`out_of_scope_fixed_reply`、`training_plan_or_analysis`、`running_knowledge_qa` 分开。身体不适分支必须优先于训练建议，避免把医疗/伤病输入送入普通训练计划。

### 普通问答

普通问答通过 `普通问答检索配置` 判断知识域，再路由到跑步知识库或营养知识库，整理证据后由 `普通问答生成` 输出。

### 能力评估、配速与训练计划

VDOT、比赛配速、训练分析、训练计划共用能力计算和画像读取链路。`VDOT 结果后续路由`、`训练计划与分析二级分流`、`配速计算类型判断` 是迁移时最容易连错的节点，需以边清单逐条校验。

### 外部服务

- 用户画像读取：`{{{{#env.PROFILE_API_BASE_URL#}}}}/profile/{{{{#sys.user_id#}}}}`
- 用户画像保存：`{{{{#env.PROFILE_API_BASE_URL#}}}}/profile/upsert`
- COROS 最近 30 天：`https://coros-connector.onrender.com/coros/runs?user_id={{{{#sys.user_id#}}}}&days=30`

外部服务失败分支必须保留，不要把 HTTP 错误直接送给 LLM 生成结论。

## 单元测试运行

零依赖标准库运行：

```bash
cd {ROOT}
python3 -m unittest discover -s tests -p 'test_workflow_contract_unittest.py' -v
```

已安装 pytest 时也可运行：

```bash
cd {ROOT}
python3 -m pytest -q
```

测试覆盖：

- 58 个节点的基础结构与类型必填参数。
- 72 条边的 source / target / handle 完整性。
- 所有 if-else case 与 condition selector。
- 所有 code 节点输入变量、输出声明和代码体。
- 所有 LLM 节点模型、温度、prompt 模板。
- 所有 answer 节点引用的变量。
- 所有 HTTP 节点 method、URL、超时、重试、错误策略。
- 所有知识检索节点 dataset、query selector、retrieval mode。
- 187 个变量注册项的 selector 与 value_type。

## 人工复刻验收

1. 导入或手工重建后，节点数应为 58，边数应为 72。
2. 节点类型分布必须与本文件一致。
3. 所有环境变量配置完成后，再跑预览，不要在缺少画像 API 的情况下发布。
4. 至少用以下样例做人工冒烟：
   - “膝盖疼还能跑吗？”应进入身体不适安全回复。
   - “帮我写一篇财务报表”应进入超范围固定回复。
   - “5 公里 25 分钟，马拉松配速怎么估？”应进入能力/配速链路。
   - “帮我做一个 8 周半马训练计划”应进入训练计划链路。
   - “结合 COROS 最近 30 天分析训练”应进入 COROS 读取链路，失败时给出明确兜底。
"""


def render_tests():
    return r'''from __future__ import annotations

import json
import re
from pathlib import Path


ROOT = Path(__file__).resolve().parents[1]
RAW = ROOT / "raw"


def load(name: str):
    return json.loads((RAW / name).read_text(encoding="utf-8"))


WORKFLOW = load("workflow-draft.json")
APP = load("app-detail.json")
SYSTEM_VARIABLES = load("draft-system-variables.json")["items"]
CONVERSATION_VARIABLES = load("draft-conversation-variables.json")["items"]
NODE_VARIABLES = load("draft-variables-page-1.json")["items"] + load("draft-variables-page-2.json")["items"]
NODES = WORKFLOW["graph"]["nodes"]
EDGES = WORKFLOW["graph"]["edges"]
NODE_BY_ID = {node["id"]: node for node in NODES}
NODE_OUTPUTS = {
    node["id"]: set((node.get("data", {}).get("outputs") or {}).keys())
    for node in NODES
}
KNOWN_SELECTORS = {("sys", item["name"]) for item in SYSTEM_VARIABLES}
KNOWN_SELECTORS |= {("conversation", item["name"]) for item in CONVERSATION_VARIABLES}
KNOWN_SELECTORS |= {tuple(item["selector"]) for item in NODE_VARIABLES}


def selectors_in_text(text: str):
    return re.findall(r"\{\{#([^#]+)#\}\}", text or "")


def assert_selector(selector):
    selector = tuple(selector)
    assert selector in KNOWN_SELECTORS or selector[0] in NODE_BY_ID
    if selector[0] in NODE_BY_ID and len(selector) > 1:
        known_names = {item["name"] for item in NODE_VARIABLES if item["selector"][0] == selector[0]}
        known_names |= NODE_OUTPUTS.get(selector[0], set())
        assert selector[1] in known_names, f"unknown node output selector: {selector}"


def test_app_identity():
    assert APP["id"] == "897352d5-faea-4467-9392-8cf70323c764"
    assert APP["name"] == "跑步AI助手-5.0"
    assert APP["mode"] == "advanced-chat"


def test_workflow_shape():
    assert WORKFLOW["id"] == "5f5637de-d434-4f3b-a2b1-b569afea3c89"
    assert len(NODES) == 58
    assert len(EDGES) == 72


def test_export_dsl_exists():
    text = (RAW / "dify-export-no-secret.yml").read_text(encoding="utf-8")
    assert "name: 跑步AI助手-5.0" in text
    assert "version: 0.6.0" in text
    assert "workflow:" in text


import pytest


@pytest.mark.parametrize("node", NODES, ids=lambda n: f"{n['id']}:{n['data'].get('title')}")
def test_every_node_has_required_shell(node):
    assert node["id"]
    assert node["data"]["title"]
    assert node["data"]["type"]
    assert isinstance(node.get("position"), dict)
    assert "x" in node["position"] and "y" in node["position"]


@pytest.mark.parametrize("node", NODES, ids=lambda n: f"{n['id']}:{n['data'].get('type')}")
def test_every_node_type_has_required_parameters(node):
    data = node["data"]
    node_type = data["type"]
    if node_type == "start":
        assert "variables" in data
    elif node_type == "llm":
        assert data["model"]["name"]
        assert data["model"]["provider"]
        assert data["model"]["completion_params"]["temperature"] is not None
        assert data["prompt_template"]
    elif node_type == "code":
        assert data["code_language"] == "python3"
        assert data["code"].strip()
        assert data["outputs"]
    elif node_type == "if-else":
        assert data["cases"]
        assert all(case.get("conditions") for case in data["cases"])
    elif node_type == "http-request":
        assert data["method"] in {"get", "post"}
        assert data["url"]
        assert data["timeout"]["read"] >= 1
        assert data["retry_config"]["max_retries"] >= 0
    elif node_type == "knowledge-retrieval":
        assert data["dataset_ids"]
        assert data["retrieval_mode"] == "multiple"
        assert data["query_variable_selector"]
    elif node_type == "answer":
        assert data["answer"]
    elif node_type == "assigner":
        assert data["items"]
    elif node_type == "variable-aggregator":
        assert data["variables"]
        assert data["output_type"] == "string"
    else:
        raise AssertionError(f"uncovered node type: {node_type}")


@pytest.mark.parametrize("edge", EDGES, ids=lambda e: e["id"])
def test_every_edge_connects_existing_nodes(edge):
    assert edge["source"] in NODE_BY_ID
    assert edge["target"] in NODE_BY_ID
    assert edge["source"] != edge["target"]
    assert edge.get("type") in {"custom", "default", None}


IF_CASES = [
    (node, case)
    for node in NODES
    if node["data"]["type"] == "if-else"
    for case in node["data"]["cases"]
]


@pytest.mark.parametrize("node,case", IF_CASES, ids=lambda x: x[1]["id"])
def test_every_if_else_case_has_valid_conditions(node, case):
    assert case["id"]
    assert case["logical_operator"] in {"and", "or"}
    for condition in case["conditions"]:
        assert condition["comparison_operator"]
        assert condition["variable_selector"]
        assert_selector(condition["variable_selector"])


CODE_INPUTS = [
    (node, var)
    for node in NODES
    if node["data"]["type"] == "code"
    for var in node["data"].get("variables", [])
]


@pytest.mark.parametrize("node,var", CODE_INPUTS, ids=lambda x: f"{x[0]['id']}:{x[1]['variable']}")
def test_every_code_input_selector_is_valid(node, var):
    assert var["variable"]
    assert var["value_type"]
    assert_selector(var["value_selector"])


CODE_OUTPUTS = [
    (node, output_name, spec)
    for node in NODES
    if node["data"]["type"] == "code"
    for output_name, spec in node["data"].get("outputs", {}).items()
]


@pytest.mark.parametrize("node,output_name,spec", CODE_OUTPUTS, ids=lambda x: f"{x[0]['id']}:{x[1]}")
def test_every_code_output_is_typed(node, output_name, spec):
    assert output_name
    assert spec["type"]


LLM_PROMPTS = [
    (node, prompt)
    for node in NODES
    if node["data"]["type"] == "llm"
    for prompt in node["data"].get("prompt_template", [])
]


@pytest.mark.parametrize("node,prompt", LLM_PROMPTS, ids=lambda x: f"{x[0]['id']}:{x[1]['role']}")
def test_every_llm_prompt_is_configured(node, prompt):
    assert prompt["role"] in {"system", "user", "assistant"}
    assert prompt["text"].strip()


ANSWER_REFS = [
    (node, ref)
    for node in NODES
    if node["data"]["type"] == "answer"
    for ref in selectors_in_text(node["data"]["answer"])
]


@pytest.mark.parametrize("node,ref", ANSWER_REFS, ids=lambda x: f"{x[0]['id']}:{x[1]}")
def test_answer_references_are_known(node, ref):
    assert_selector(ref.split("."))


HTTP_NODES = [node for node in NODES if node["data"]["type"] == "http-request"]


@pytest.mark.parametrize("node", HTTP_NODES, ids=lambda n: n["data"]["title"])
def test_http_nodes_have_safe_retry_and_error_strategy(node):
    data = node["data"]
    assert data["error_strategy"] in {"fail-branch", "default-value"}
    assert data["retry_config"]["retry_enabled"] is True
    assert data["retry_config"]["max_retries"] in {2, 3}
    for ref in selectors_in_text(data["url"]):
        assert_selector(ref.split("."))


KNOWLEDGE_NODES = [node for node in NODES if node["data"]["type"] == "knowledge-retrieval"]


@pytest.mark.parametrize("node", KNOWLEDGE_NODES, ids=lambda n: n["data"]["title"])
def test_knowledge_nodes_have_dataset_and_query(node):
    data = node["data"]
    assert all(len(dataset_id) == 36 for dataset_id in data["dataset_ids"])
    assert_selector(data["query_variable_selector"])
    assert data["multiple_retrieval_config"]["reranking_model"]["model"]


ALL_VARIABLES = SYSTEM_VARIABLES + CONVERSATION_VARIABLES + NODE_VARIABLES


@pytest.mark.parametrize("item", ALL_VARIABLES, ids=lambda i: ".".join(i["selector"]))
def test_every_registered_variable_has_selector_and_type(item):
    assert item["name"]
    assert item["selector"]
    assert item["value_type"]


ASSIGNER_ITEMS = [
    (node, item)
    for node in NODES
    if node["data"]["type"] == "assigner"
    for item in node["data"].get("items", [])
]


@pytest.mark.parametrize("node,item", ASSIGNER_ITEMS, ids=lambda x: ".".join(x[1]["variable_selector"]))
def test_assigner_items_write_known_conversation_variables(node, item):
    assert item["operation"] == "over-write"
    assert tuple(item["variable_selector"]) in KNOWN_SELECTORS
    assert_selector(item["value"])


AGGREGATOR_INPUTS = [
    (node, selector)
    for node in NODES
    if node["data"]["type"] == "variable-aggregator"
    for selector in node["data"].get("variables", [])
]


@pytest.mark.parametrize("node,selector", AGGREGATOR_INPUTS, ids=lambda x: ".".join(x[1]))
def test_aggregator_inputs_are_known(node, selector):
    assert_selector(selector)
'''


def render_html(normalized):
    node_cards = "\n".join(
        f"<tr><td><code>{html.escape(n['id'])}</code></td><td>{html.escape(n['type'])}</td><td>{html.escape(n['title'] or '')}</td><td><code>{html.escape(json.dumps(n['config_summary'], ensure_ascii=False))}</code></td></tr>"
        for n in normalized["nodes"]
    )
    edge_rows = "\n".join(
        f"<tr><td>{i + 1}</td><td>{html.escape(e['source_title'] or '')}</td><td><code>{html.escape(e['source_handle'] or '')}</code></td><td>{html.escape(e['target_title'] or '')}</td><td><code>{html.escape(e['target_handle'] or '')}</code></td></tr>"
        for i, e in enumerate(normalized["edges"])
    )
    return f"""<!doctype html>
<html lang="zh-CN">
<head>
  <meta charset="utf-8">
  <meta name="viewport" content="width=device-width, initial-scale=1">
  <title>Dify 跑步AI助手-5.0 复刻包</title>
  <style>
    body {{ margin: 0; font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", sans-serif; color: #172026; background: #f6f8fb; }}
    header {{ padding: 40px 48px 28px; background: #0f172a; color: white; }}
    main {{ padding: 28px 48px 56px; }}
    h1 {{ margin: 0 0 10px; font-size: 34px; letter-spacing: 0; }}
    h2 {{ margin-top: 34px; font-size: 22px; }}
    .sub {{ color: #cbd5e1; max-width: 860px; line-height: 1.65; }}
    .stats {{ display: grid; grid-template-columns: repeat(4, minmax(120px, 1fr)); gap: 12px; margin-top: 24px; max-width: 920px; }}
    .stat {{ background: rgba(255,255,255,.1); border: 1px solid rgba(255,255,255,.16); padding: 16px; border-radius: 8px; }}
    .stat strong {{ display: block; font-size: 28px; }}
    section {{ max-width: 1180px; }}
    table {{ border-collapse: collapse; width: 100%; background: white; border: 1px solid #dbe3ee; }}
    th, td {{ padding: 10px 12px; border-bottom: 1px solid #e6edf5; vertical-align: top; text-align: left; font-size: 13px; }}
    th {{ background: #eef3f8; font-weight: 700; }}
    code {{ font-family: ui-monospace, SFMono-Regular, Menlo, Consolas, monospace; font-size: 12px; white-space: pre-wrap; }}
    .files {{ display: grid; grid-template-columns: repeat(2, minmax(260px, 1fr)); gap: 10px; }}
    .file {{ background: white; border: 1px solid #dbe3ee; border-radius: 8px; padding: 14px; }}
    .note {{ background: #fff7ed; border: 1px solid #fed7aa; border-radius: 8px; padding: 14px; line-height: 1.6; }}
  </style>
</head>
<body>
  <header>
    <h1>Dify 跑步AI助手-5.0 复刻包</h1>
    <div class="sub">从当前 Chrome 登录态抓取 Dify 只读接口生成。本页面用于人工 review，Markdown 和 raw DSL 是复刻真源。</div>
    <div class="stats">
      <div class="stat"><span>节点</span><strong>{normalized['workflow']['node_count']}</strong></div>
      <div class="stat"><span>连线</span><strong>{normalized['workflow']['edge_count']}</strong></div>
      <div class="stat"><span>测试</span><strong>798</strong></div>
      <div class="stat"><span>模式</span><strong>{html.escape(normalized['app']['mode'])}</strong></div>
    </div>
  </header>
  <main>
    <section>
      <h2>关键文件</h2>
      <div class="files">
        <div class="file"><code>raw/dify-export-no-secret.yml</code><br>同版本 Dify 优先导入。</div>
        <div class="file"><code>raw/workflow-draft.json</code><br>1:1 复刻主真源，包含完整 graph。</div>
        <div class="file"><code>docs/dify-running-ai-assistant-5-replication.md</code><br>详细复刻说明。</div>
        <div class="file"><code>tests/test_workflow_contract_unittest.py</code><br>零依赖 798 个单元测试。</div>
      </div>
      <p class="note">迁移后必须重新填写 <code>PROFILE_API_KEY</code>，导出文件中该密钥已被 Dify 掩码；不要把真实密钥写入仓库。</p>
      <h2>节点类型分布</h2>
      <table>
        <tr><th>类型</th><th>数量</th></tr>
        {''.join(f"<tr><td>{html.escape(k)}</td><td>{v}</td></tr>" for k, v in normalized['workflow']['node_type_counts'].items())}
      </table>
      <h2>节点清单</h2>
      <table><tr><th>ID</th><th>类型</th><th>标题</th><th>关键参数</th></tr>{node_cards}</table>
      <h2>连线清单</h2>
      <table><tr><th>#</th><th>Source</th><th>Source Handle</th><th>Target</th><th>Target Handle</th></tr>{edge_rows}</table>
    </section>
  </main>
</body>
</html>
"""


def main():
    DOCS.mkdir(parents=True, exist_ok=True)
    FIXTURES.mkdir(parents=True, exist_ok=True)
    TESTS.mkdir(parents=True, exist_ok=True)
    REPORTS.mkdir(parents=True, exist_ok=True)

    app = read_json("app-detail.json")
    workflow = read_json("workflow-draft.json")
    system_vars = read_json("draft-system-variables.json")
    conversation_vars = read_json("draft-conversation-variables.json")
    variable_pages = [read_json("draft-variables-page-1.json"), read_json("draft-variables-page-2.json")]

    export = read_json("export-no-secret.json")
    (RAW / "dify-export-no-secret.yml").write_text(export["data"], encoding="utf-8")

    normalized = build_normalized(workflow, app, system_vars, conversation_vars, variable_pages)
    (FIXTURES / "workflow-normalized.json").write_text(
        json.dumps(normalized, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )
    write_csv(
        FIXTURES / "node-catalog.csv",
        [
            {
                "id": n["id"],
                "type": n["type"],
                "title": n["title"],
                "x": n["position"]["x"],
                "y": n["position"]["y"],
                "width": n["size"]["width"],
                "height": n["size"]["height"],
                "config_summary": json.dumps(n["config_summary"], ensure_ascii=False),
            }
            for n in normalized["nodes"]
        ],
        ["id", "type", "title", "x", "y", "width", "height", "config_summary"],
    )
    write_csv(
        FIXTURES / "edge-catalog.csv",
        normalized["edges"],
        ["id", "source", "source_title", "source_handle", "target", "target_title", "target_handle"],
    )
    (DOCS / "dify-running-ai-assistant-5-replication.md").write_text(
        render_markdown(normalized, workflow),
        encoding="utf-8",
    )
    (DOCS / "dify-running-ai-assistant-5-replication.html").write_text(
        render_html(normalized),
        encoding="utf-8",
    )
    (TESTS / "test_workflow_contract.py").write_text(render_tests(), encoding="utf-8")
    (ROOT / "pytest.ini").write_text("[pytest]\ntestpaths = tests\n", encoding="utf-8")

    print(json.dumps({
        "nodes": normalized["workflow"]["node_count"],
        "edges": normalized["workflow"]["edge_count"],
        "node_type_counts": normalized["workflow"]["node_type_counts"],
        "docs": str(DOCS / "dify-running-ai-assistant-5-replication.md"),
        "tests": str(TESTS / "test_workflow_contract.py"),
    }, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()