aigc-home/calliope-backend/scripts/ui_import_dryrun.py

"""Generate the HEL-681 dry-run report over the repo's UI workflow library.

Walks ``example_ComfyUI_workflows`` (or ``--root``), runs every file through
the production converter + lint + parser pipeline (exactly what
``POST /api/workflows/convert`` does, minus HTTP), and writes a JSON detail
file plus a human-readable Markdown summary. Read-only: nothing is imported,
no DB is touched.

Usage:
    python scripts/ui_import_dryrun.py [--root DIR] [--out PREFIX]
"""

from __future__ import annotations

import argparse
import collections
import json
import sys
from pathlib import Path

BACKEND = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(BACKEND / "src"))

from calliope.comfyui.lint import workflow_warnings  # noqa: E402
from calliope.comfyui.metadata import suggested_metadata  # noqa: E402
from calliope.comfyui.parser import classify_usage  # noqa: E402
from calliope.comfyui.ui_convert import convert_ui_workflow  # noqa: E402

REPO = BACKEND.parent


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--root", default=str(REPO / "example_ComfyUI_workflows"))
    parser.add_argument("--out", default=str(REPO / "dryrun_report"))
    args = parser.parse_args()
    root = Path(args.root)
    out_prefix = Path(args.out)

    entries = []
    for path in sorted(root.rglob("*.json")):
        try:
            data = json.loads(path.read_text(encoding="utf-8"))
        except (json.JSONDecodeError, OSError) as exc:
            entries.append(
                {
                    "source_path": str(path.relative_to(root)),
                    "status": "fail",
                    "issues": [{"code": "unreadable", "message": str(exc)}],
                }
            )
            continue
        if not (isinstance(data, dict) and isinstance(data.get("nodes"), list)):
            continue  # API-format / non-workflow files are out of scope
        rel = str(path.relative_to(root))
        result = convert_ui_workflow(data)
        entry: dict = {
            "source_path": rel,
            "status": result.status,
            "needs_confirmation": result.needs_confirmation,
            "issues": [
                {"node_id": i.node_id, "code": i.code, "level": i.level, "message": i.message}
                for i in result.issues
            ],
        }
        if result.workflow is not None:
            entry["node_count"] = len(result.workflow)
            entry["suggested"] = suggested_metadata(result.workflow, rel, data)
            errors = [w for w in workflow_warnings(result.workflow) if w["level"] == "error"]
            entry["lint_errors"] = [e["code"] for e in errors]
            entry["importable"] = not result.needs_confirmation and not errors
        else:
            entry["importable"] = False
        entries.append(entry)

    counts = collections.Counter(e["status"] for e in entries)
    code_counts = collections.Counter(
        i["code"] for e in entries for i in e["issues"]
    )
    category_counts = collections.Counter(
        (e.get("suggested") or {}).get("category") or "?" for e in entries
    )
    family_counts = collections.Counter(
        (e.get("suggested") or {}).get("model_family") or "?" for e in entries
    )
    importable = sum(1 for e in entries if e.get("importable"))
    needs_confirmation = sum(1 for e in entries if e["needs_confirmation"])
    lint_blocked = [e for e in entries if e.get("lint_errors")]

    detail = {
        "root": str(root),
        "total": len(entries),
        "status_counts": dict(counts),
        "issue_code_counts": dict(code_counts),
        "importable": importable,
        "needs_confirmation": needs_confirmation,
        "lint_blocked": [e["source_path"] for e in lint_blocked],
        "files": entries,
    }
    out_prefix.parent.mkdir(parents=True, exist_ok=True)
    (out_prefix.with_suffix(".json")).write_text(
        json.dumps(detail, ensure_ascii=False, indent=1), encoding="utf-8"
    )

    lines = [
        "# HEL-681 工作流 UI→API 转换 dry-run 报告",
        "",
        f"- 扫描目录：`{root}`（根目录 10 个 API 格式旧文件不在本管线范围）",
        f"- UI 格式文件总数：**{len(entries)}**",
        f"- 转换结果：ok **{counts.get('ok', 0)}** / warn **{counts.get('warn', 0)}** / fail **{counts.get('fail', 0)}**",
        f"- 其中「需人工确认」（控件无法可靠映射，禁止直接导入）：**{needs_confirmation}**",
        f"- 通过转换＋lint 后可直接导入：**{importable}**",
        f"- 转换成功但被现有 lint 拦截（导入时按阻止处理）：**{len(lint_blocked)}**",
        "",
        "## 分类分布（自动提取）",
        "",
        "| 分类 | 数量 |",
        "|---|---|",
    ]
    for cat, n in sorted(category_counts.items()):
        lines.append(f"| {cat} | {n} |")
    lines += ["", "## 模型家族分布（自动提取）", "", "| 家族 | 数量 |", "|---|---|"]
    for fam, n in sorted(family_counts.items(), key=lambda kv: -kv[1]):
        lines.append(f"| {fam} | {n} |")
    lines += ["", "## 问题代码统计", "", "| 代码 | 次数 | 含义 |", "|---|---|---|"]
    meaning = {
        "muted_input_dropped": "上游节点被静音，输入按未连接处理",
        "variable_upstream_dropped": "Set/Get 变量上游被旁路，取值按未连接处理",
        "get_without_set": "GetNode 没有同名 SetNode",
        "widget_unmapped": "控件值无法可靠映射到输入名（需人工确认）",
        "widget_order_mismatch": "控件数量与已知顺序不符（需人工确认）",
        "no_output_node": "图中无已知输出节点，全部活跃节点保留待确认",
        "cyclic_variable": "Set/Get 循环（FOR 循环流），该连线按未连接处理",
        "dangling_link": "连线指向不存在的节点",
    }
    for code, n in code_counts.most_common():
        lines.append(f"| `{code}` | {n} | {meaning.get(code, '')} |")
    lines += ["", "## 需人工确认文件清单", ""]
    for e in entries:
        if e["needs_confirmation"]:
            reasons = sorted({i["code"] for i in e["issues"]})[:3]
            lines.append(f"- `{e['source_path']}`（{', '.join(reasons)}）")
    lines += ["", "## lint 拦截文件清单", ""]
    for e in lint_blocked:
        lines.append(f"- `{e['source_path']}`（{', '.join(e['lint_errors'])}）")
    lines += [
        "",
        "## 逐文件结果",
        "",
        "| 文件 | 状态 | 节点数 | 分类 | 模型家族 | 用途 | 问题 |",
        "|---|---|---|---|---|---|---|",
    ]
    for e in entries:
        sug = e.get("suggested") or {}
        issues = collections.Counter(i["code"] for i in e["issues"])
        issue_text = ", ".join(f"{c}×{n}" for c, n in issues.most_common(2)) or "-"
        usage = sug.get("usage") or ("待定" if sug.get("usage_ambiguous") else "-")
        lines.append(
            f"| `{e['source_path']}` | {e['status']} | {e.get('node_count', '-')} "
            f"| {sug.get('category') or '-'} | {sug.get('model_family') or '-'} "
            f"| {usage} | {issue_text} |"
        )
    (out_prefix.with_suffix(".md")).write_text("\n".join(lines) + "\n", encoding="utf-8")
    print(f"{len(entries)} files: {dict(counts)}, importable={importable}, nc={needs_confirmation}")
    print(f"report → {out_prefix}.md / {out_prefix}.json")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())