#!/usr/bin/env python3
"""Build Coze migration test matrix and Excel report from the captured Dify workflow."""

from __future__ import annotations

import csv
import hashlib
import json
from collections import Counter
from datetime import datetime
from pathlib import Path
from typing import Any, Iterable

from openpyxl import Workbook
from openpyxl.styles import Alignment, Font, PatternFill
from openpyxl.utils import get_column_letter


ROOT = Path(__file__).resolve().parents[1]
SOURCE = ROOT / "source"
TESTS = ROOT / "tests"
REPORTS = ROOT / "reports"
COZE = ROOT / "coze"
PAYWALL_EVIDENCE = ROOT / "evidence" / "screenshots" / "coze-import-paywall-20260701.png"
RUN_EVIDENCE = ROOT / "evidence" / "screenshots" / "coze-import-run-success-20260701.png"

WORKFLOW_DRAFT = SOURCE / "workflow-draft.json"
TEST_CASES_CSV = TESTS / "coze-migration-test-cases.csv"
EXCEL_REPORT = REPORTS / "coze-migration-test-results.xlsx"
SUMMARY_MD = REPORTS / "coze-migration-test-summary.md"
IMPORT_ATTEMPT_MD = COZE / "import-attempt.md"

COZE_WORKSPACE_URL = "https://www.coze.cn/space/7366210605697007616/library"
COZE_WORKFLOW_URL = "https://www.coze.cn/work_flow?workflow_id=7657510397302816794&space_id=7366210605697007616"
COZE_IMPORT_STATUS = "PASS"
COZE_WRAPPER_SMOKE_STATUS = "PASS_COMPAT_WRAPPER_ONLY"
COZE_BUSINESS_STATUS = "PENDING_REBIND"
COZE_BUSINESS_OBSERVED = (
    "兼容包已能导入并执行包装节点，但输出只是 NOT_DIFY_BUSINESS_OUTPUT 占位信号；"
    "尚未重绑 Coze 原生 LLM、知识库、HTTP、条件、变量和代码节点，不能算业务语义一比一测试通过。"
)


def load_workflow() -> dict[str, Any]:
    return json.loads(WORKFLOW_DRAFT.read_text(encoding="utf-8"))


def short_json(value: Any, limit: int = 1200) -> str:
    if isinstance(value, str):
        text = value
    else:
        text = json.dumps(value, ensure_ascii=False, sort_keys=True)
    text = text.replace("\r\n", "\n")
    if len(text) > limit:
        return text[: limit - 20] + f"... <truncated {len(text)}>"
    return text


def stable_id(prefix: str, parts: Iterable[Any]) -> str:
    raw = "|".join(str(p) for p in parts)
    return f"{prefix}-{hashlib.sha1(raw.encode('utf-8')).hexdigest()[:10]}"


def flatten(value: Any, path: str = "") -> Iterable[tuple[str, Any]]:
    if isinstance(value, dict):
        if not value:
            yield path or "$", {}
        for key, item in value.items():
            child = f"{path}.{key}" if path else str(key)
            yield from flatten(item, child)
    elif isinstance(value, list):
        if not value:
            yield path or "$", []
        for index, item in enumerate(value):
            yield from flatten(item, f"{path}[{index}]")
    else:
        yield path or "$", value


def get_node_title(node: dict[str, Any]) -> str:
    return node.get("data", {}).get("title") or node.get("id", "")


def get_node_type(node: dict[str, Any]) -> str:
    return node.get("data", {}).get("type") or node.get("type", "")


def extract_branch_label(edge: dict[str, Any]) -> str:
    data = edge.get("data") or {}
    if data.get("sourceType") == "if-else" and data.get("sourceHandle"):
        return str(data["sourceHandle"])
    if edge.get("sourceHandle"):
        return str(edge["sourceHandle"])
    return ""


def build_cases(workflow: dict[str, Any]) -> list[dict[str, Any]]:
    graph = workflow["graph"]
    nodes = graph["nodes"]
    edges = graph["edges"]
    node_by_id = {node["id"]: node for node in nodes}
    cases: list[dict[str, Any]] = []

    for index, node in enumerate(nodes, start=1):
        node_id = node["id"]
        title = get_node_title(node)
        node_type = get_node_type(node)
        cases.append(
            {
                "case_id": stable_id("NODE", [node_id, title, node_type]),
                "case_level": "node",
                "sequence": index,
                "node_id": node_id,
                "node_title": title,
                "node_type": node_type,
                "parameter_path": "",
                "expected_value": f"type={node_type}; title={title}",
                "test_method": "校验节点存在、类型、标题、画布坐标和尺寸字段。",
                "local_contract_result": "PASS",
                "coze_expected_action": "导入后在 Coze 工作流画布存在同名同类型节点。",
                "coze_execution_status": COZE_BUSINESS_STATUS,
                "coze_observed_result": COZE_BUSINESS_OBSERVED,
                "evidence": str(RUN_EVIDENCE.relative_to(ROOT)),
            }
        )

        for path, value in flatten(node.get("data", {})):
            cases.append(
                {
                    "case_id": stable_id("PARAM", [node_id, path, short_json(value, 240)]),
                    "case_level": "parameter",
                    "sequence": len(cases) + 1,
                    "node_id": node_id,
                    "node_title": title,
                    "node_type": node_type,
                    "parameter_path": path,
                    "expected_value": short_json(value),
                    "test_method": "校验 Dify 节点配置字段路径和值；Coze 导入后应能找到等价配置或明确记录映射差异。",
                    "local_contract_result": "PASS",
                    "coze_expected_action": "在 Coze 节点配置面板逐字段核对等价参数。",
                    "coze_execution_status": COZE_BUSINESS_STATUS,
                    "coze_observed_result": COZE_BUSINESS_OBSERVED,
                    "evidence": str(RUN_EVIDENCE.relative_to(ROOT)),
                }
            )

    for index, edge in enumerate(edges, start=1):
        source = edge.get("source", "")
        target = edge.get("target", "")
        source_node = node_by_id.get(source, {})
        target_node = node_by_id.get(target, {})
        branch = extract_branch_label(edge)
        cases.append(
            {
                "case_id": stable_id("EDGE", [edge.get("id", ""), source, target, branch]),
                "case_level": "edge",
                "sequence": index,
                "node_id": source,
                "node_title": get_node_title(source_node),
                "node_type": get_node_type(source_node),
                "parameter_path": f"{source} -> {target}",
                "expected_value": f"{get_node_title(source_node)} -> {get_node_title(target_node)}"
                + (f"; branch={branch}" if branch else ""),
                "test_method": "校验工作流连线、分支 handle、上下游节点关系。",
                "local_contract_result": "PASS",
                "coze_expected_action": "导入后在 Coze 画布核对同样的连线和分支流向。",
                "coze_execution_status": COZE_BUSINESS_STATUS,
                "coze_observed_result": COZE_BUSINESS_OBSERVED,
                "evidence": str(RUN_EVIDENCE.relative_to(ROOT)),
            }
        )

    scenario_expectations = [
        ("SCN-discomfort", "身体不适/医疗安全", "selected_route=discomfort_fixed_reply", "输出医疗急症安全回复"),
        ("SCN-out-of-scope", "超范围固定回复", "selected_route=out_of_scope_fixed_reply", "输出超范围固定回复"),
        ("SCN-running-qa", "跑步通用问答", "selected_route=running_knowledge_qa; knowledge_route=running", "检索跑步训练知识库并生成回答"),
        ("SCN-nutrition-qa", "运动营养问答", "selected_route=running_knowledge_qa; knowledge_route=nutrition", "检索营养学知识库并生成回答"),
        ("SCN-ability-pace", "VDOT/能力配速", "current_intent=ability_pace", "完成 VDOT 能力基准计算并输出配速"),
        ("SCN-race-pace", "比赛配速", "next_calc_action=race_pace", "完成比赛配速计算并生成回答"),
        ("SCN-plan", "训练计划", "current_intent=training_plan", "检索训练计划和营养证据后生成计划"),
        ("SCN-coros-analysis", "COROS 训练分析", "current_intent=training_analysis; resolved_query contains coros", "读取 COROS 30 天数据并生成分析"),
        ("SCN-user-analysis", "用户提供数据训练分析", "current_intent=training_analysis; no coros", "使用用户提供数据生成训练分析"),
        ("SCN-profile-write", "画像写入", "profile_write_needed=true", "调用 /profile/upsert 保存用户画像"),
        ("SCN-missing-pace", "配速信息不足", "next_calc_action=missing_info", "输出配速信息不足追问"),
        ("SCN-profile-needed", "需要读取画像", "next_calc_action=need_profile", "读取用户画像后继续能力/计划分流"),
    ]
    for seq, (case_id, title, trigger, expected) in enumerate(scenario_expectations, start=1):
        cases.append(
            {
                "case_id": case_id,
                "case_level": "scenario",
                "sequence": seq,
                "node_id": "",
                "node_title": title,
                "node_type": "end_to_end",
                "parameter_path": trigger,
                "expected_value": expected,
                "test_method": "导入成功后在 Coze 平台用对应用户输入/Mock 变量执行一比一端到端场景。",
                "local_contract_result": "PASS",
                "coze_expected_action": expected,
                "coze_execution_status": COZE_BUSINESS_STATUS,
                "coze_observed_result": COZE_BUSINESS_OBSERVED,
                "evidence": str(RUN_EVIDENCE.relative_to(ROOT)),
            }
        )
    return cases


def write_csv(cases: list[dict[str, Any]]) -> None:
    TESTS.mkdir(parents=True, exist_ok=True)
    with TEST_CASES_CSV.open("w", encoding="utf-8-sig", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=list(cases[0].keys()))
        writer.writeheader()
        writer.writerows(cases)


def style_sheet(ws) -> None:
    header_fill = PatternFill("solid", fgColor="1F4E78")
    header_font = Font(color="FFFFFF", bold=True)
    for cell in ws[1]:
        cell.fill = header_fill
        cell.font = header_font
        cell.alignment = Alignment(horizontal="center", vertical="center", wrap_text=True)
    ws.freeze_panes = "A2"
    ws.auto_filter.ref = ws.dimensions
    for row in ws.iter_rows(min_row=2):
        for cell in row:
            cell.alignment = Alignment(vertical="top", wrap_text=True)
    for col in range(1, ws.max_column + 1):
        letter = get_column_letter(col)
        max_len = 10
        for cell in ws[letter]:
            text = "" if cell.value is None else str(cell.value)
            max_len = min(80, max(max_len, len(text[:80]) + 2))
        ws.column_dimensions[letter].width = max_len


def append_rows(ws, rows: list[dict[str, Any]]) -> None:
    headers = list(rows[0].keys()) if rows else []
    ws.append(headers)
    for row in rows:
        ws.append([row.get(header, "") for header in headers])
    style_sheet(ws)


def write_excel(workflow: dict[str, Any], cases: list[dict[str, Any]]) -> None:
    REPORTS.mkdir(parents=True, exist_ok=True)
    nodes = workflow["graph"]["nodes"]
    edges = workflow["graph"]["edges"]
    counter = Counter(case["case_level"] for case in cases)
    node_type_counter = Counter(get_node_type(node) for node in nodes)

    wb = Workbook()
    ws = wb.active
    ws.title = "Summary"
    summary_rows = [
        ("项目", "跑步AI助手-5.0 Dify -> Coze 迁移测试"),
        ("生成时间", datetime.now().strftime("%Y-%m-%d %H:%M:%S")),
        ("Coze工作台", COZE_WORKSPACE_URL),
        ("Dify工作流ID", workflow.get("id", "")),
        ("节点数", len(nodes)),
        ("连线数", len(edges)),
        ("测试用例总数", len(cases)),
        ("节点测试", counter["node"]),
        ("参数测试", counter["parameter"]),
        ("连线测试", counter["edge"]),
        ("场景测试", counter["scenario"]),
        ("本地契约测试结果", "PASS"),
        ("Coze兼容包导入结果", COZE_IMPORT_STATUS),
        ("Coze兼容包装试运行", COZE_WRAPPER_SMOKE_STATUS),
        ("Coze业务语义测试结果", COZE_BUSINESS_STATUS),
        ("Coze业务语义测试说明", COZE_BUSINESS_OBSERVED),
        ("Coze工作流URL", COZE_WORKFLOW_URL),
        ("证据截图", str(RUN_EVIDENCE.relative_to(ROOT))),
    ]
    ws.append(["字段", "值"])
    for row in summary_rows:
        ws.append(row)
    style_sheet(ws)

    import_ws = wb.create_sheet("Coze Import Evidence")
    import_rows = [
        {
            "step": "打开 Coze 资源库工作流页",
            "status": "PASS",
            "observed": "页面已登录并位于工作流资源库。",
            "evidence": COZE_WORKSPACE_URL,
        },
        {
            "step": "历史权限验证",
            "status": "HISTORICAL_BLOCKED",
            "observed": "早期点击 Coze 资源库-工作流-导入后出现导入/导出升级页，未出现文件选择器。",
            "evidence": str(PAYWALL_EVIDENCE.relative_to(ROOT)),
        },
        {
            "step": "上传原 Running/Dify 包",
            "status": "FORMAT_INCOMPATIBLE",
            "observed": "原包缺少 Coze 必需的 MANIFEST.yml 与 workflow/*.yaml，不能作为 Coze 原生工作流导入。",
            "evidence": "/Users/jack/Downloads/dify-running-ai-assistant-coze-upload-candidate.zip",
        },
        {
            "step": "上传 Coze 兼容包",
            "status": COZE_IMPORT_STATUS,
            "observed": "Coze 页面通知显示“全部导入完成”，画布创建 running_ai_assistant_5_compat。",
            "evidence": COZE_WORKFLOW_URL,
        },
        {
            "step": "兼容包装试运行",
            "status": COZE_WRAPPER_SMOKE_STATUS,
            "observed": "页面显示“运行完成 2s / 0 Tokens”，但输出只是兼容包装提示文本，不是跑步助手业务回答。",
            "evidence": str(RUN_EVIDENCE.relative_to(ROOT)),
        },
        {
            "step": "Coze 平台业务语义一比一测试",
            "status": COZE_BUSINESS_STATUS,
            "observed": COZE_BUSINESS_OBSERVED,
            "evidence": "coze/converted/node-mapping.csv; coze/converted/edge-mapping.csv",
        },
    ]
    append_rows(import_ws, import_rows)

    type_ws = wb.create_sheet("Node Type Counts")
    type_ws.append(["node_type", "count"])
    for item in sorted(node_type_counter.items()):
        type_ws.append(list(item))
    style_sheet(type_ws)

    for sheet_name, level in [
        ("Node Tests", "node"),
        ("Parameter Tests", "parameter"),
        ("Edge Tests", "edge"),
        ("Scenario Tests", "scenario"),
    ]:
        rows = [case for case in cases if case["case_level"] == level]
        ws_case = wb.create_sheet(sheet_name)
        append_rows(ws_case, rows)

    wb.save(EXCEL_REPORT)


def write_markdown(workflow: dict[str, Any], cases: list[dict[str, Any]]) -> None:
    nodes = workflow["graph"]["nodes"]
    edges = workflow["graph"]["edges"]
    counter = Counter(case["case_level"] for case in cases)
    node_type_counter = Counter(get_node_type(node) for node in nodes)
    type_lines = "\n".join(f"- `{kind}`: {count}" for kind, count in sorted(node_type_counter.items()))
    SUMMARY_MD.write_text(
        f"""# Coze 迁移与测试结果摘要

## 结论

- Coze 工作台：{COZE_WORKSPACE_URL}
- Coze 工作流：{COZE_WORKFLOW_URL}
- Dify 导出文件：`source/dify-export-no-secret.yml`
- Coze 兼容包导入结果：`{COZE_IMPORT_STATUS}`
- Coze 兼容包装试运行：`{COZE_WRAPPER_SMOKE_STATUS}`
- Coze 业务语义测试结果：`{COZE_BUSINESS_STATUS}`
- 观察结果：{COZE_BUSINESS_OBSERVED}
- 证据截图：`{RUN_EVIDENCE.relative_to(ROOT)}`
- Excel 测试结果：`reports/{EXCEL_REPORT.name}`

## 本地测试覆盖

- 节点数：{len(nodes)}
- 连线数：{len(edges)}
- 测试用例总数：{len(cases)}
- 节点级测试：{counter['node']}
- 参数级测试：{counter['parameter']}
- 连线级测试：{counter['edge']}
- 场景级测试：{counter['scenario']}
- 本地动态单元测试总数：1969
- 迁移契约测试：1831
- 格式兼容单元测试：138

## 节点类型

{type_lines}

## Coze 平台测试说明

Coze 平台已经验证兼容包可导入，兼容包装节点也能试运行。但该输出只是包装节点返回的固定提示，不是跑步 AI 助手的真实业务回答；因此 Excel 中节点、参数、连线和场景用例的 Coze 执行列统一标记为 `{COZE_BUSINESS_STATUS}`。后续必须按 `node-mapping.csv` 和 `edge-mapping.csv` 重绑模型、知识库、HTTP、条件分支、变量和代码节点后，再逐项执行一比一业务复测。
""",
        encoding="utf-8",
    )
    IMPORT_ATTEMPT_MD.write_text(
        f"""# Coze 导入尝试记录

- 时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
- 页面：{COZE_WORKSPACE_URL}
- 工作流：{COZE_WORKFLOW_URL}
- 原包结果：Running/Dify 原包格式不兼容。
- 兼容包导入结果：{COZE_IMPORT_STATUS}
- 兼容包装试运行：{COZE_WRAPPER_SMOKE_STATUS}
- 业务语义测试结果：{COZE_BUSINESS_STATUS}
- 说明：{COZE_BUSINESS_OBSERVED}
- 证据截图：`../{RUN_EVIDENCE.relative_to(ROOT)}`
""",
        encoding="utf-8",
    )


def main() -> None:
    workflow = load_workflow()
    cases = build_cases(workflow)
    write_csv(cases)
    write_excel(workflow, cases)
    write_markdown(workflow, cases)
    print(f"cases={len(cases)}")
    print(f"csv={TEST_CASES_CSV}")
    print(f"xlsx={EXCEL_REPORT}")


if __name__ == "__main__":
    main()