Agent/examples/process_pipeline/script/search_eval/build_workflows.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""把一个 run 目录（如 runs_full/q0000）里**每个帖子**的 workflow.json，
与它对应的 post 信息（含 llm_evaluation）以及 query 词合并成一个 JSON。
以帖子为单位：一个 procedure 输出一个文件，文件落在 search_eval/workflows/ 下。
（例：q0000 有 3 个 procedure -> 写出 3 个 json）

映射逻辑：
  procedure 文件夹名形如  {FORM}_{platform}_{hash前缀}   例: A_gzh_8f5fbfb0
  -> 读 form_{FORM}.json，在 results[] 里找 case_id 以 "{platform}_{hash前缀}" 开头的那条
  -> 该 result 即对应的 post（post / comments / llm_evaluation / source_url ...）
  -> query / original_q 取自 form_{FORM}.json 顶层

输出文件名： {run_id}_{folder}.json    例: q0000_A_gzh_8f5fbfb0.json

本模块既是 build 脚本，也是一个 HTTP 接口：
  * build 函数（build_run / write_run）保留，供外部 import 调用或经 POST /build 触发；
  * 接口本身实时扫描 workflows/ 目录，把里面所有 json 以数组形式返回。

用法（build）：
  python build_workflows.py              # 默认处理 q0000
  python build_workflows.py q0003        # 处理指定 run
  python build_workflows.py --all        # 处理 runs_full 下所有 q* 目录

用法（接口）：
  python build_workflows.py serve [port] # 默认 8771
    GET  /workflows   -> 实时扫描 workflows/*.json，返回数组
    GET  /            -> 同上（方便直接访问）
    POST /build       -> body {"q":"q0003"} 或 {"all":true}，触发 build 后返回结果
"""
import json
import os
import re
import sys
import glob
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

HERE = os.path.dirname(os.path.abspath(__file__))
RUNS_DIR = os.path.join(HERE, "runs_full")
OUT_DIR = os.path.join(HERE, "workflows")
DEFAULT_PORT = 8771

# 文件夹名: 表单字母 _ 平台 _ case_id 哈希前缀
FOLDER_RE = re.compile(r"^([A-Za-z])_([a-z0-9]+)_([0-9a-fA-F]+)$")


def load_json(path):
    with open(path, encoding="utf-8") as f:
        return json.load(f)


def build_run(run_id, runs_dir=None, only_folder=None):
    """为单个 run 目录构建合并结果。

    以帖子为单位：返回一个 list，每个元素是 (folder, merged_dict)，
    merged_dict 即单个帖子的合并 JSON（query + post + llm_evaluation + workflow）。
    找不到 procedures 时返回空 list。
    runs_dir 缺省用模块的 RUNS_DIR；外部脚本（如 batch_extract_procedures.py 用了
    --output-dir）可传入自己的 runs_full，避免两边路径不一致。
    only_folder 非空时只处理那一个 folder（其余直接跳过、不刷 warn）——给 write_one
    用，避免同 q 里别的未完成 folder 每次都被扫一遍刷一堆 warn。"""
    run_dir = os.path.join(runs_dir or RUNS_DIR, run_id)
    proc_root = os.path.join(run_dir, "procedures")
    if not os.path.isdir(proc_root):
        print(f"[skip] {run_id}: 没有 procedures/ 目录")
        return []

    # 缓存已加载的 form_{X}.json，并记录 query（取第一个见到的）
    forms = {}
    query = original_q = None
    platforms = None

    def get_form(letter):
        nonlocal query, original_q, platforms
        if letter not in forms:
            forms[letter] = load_json(os.path.join(run_dir, f"form_{letter}.json"))
            if query is None:
                query = forms[letter].get("query")
                original_q = forms[letter].get("original_q")
                platforms = forms[letter].get("platforms")
        return forms[letter]

    out = []
    for folder in sorted(os.listdir(proc_root)):
        if only_folder and folder != only_folder:
            continue                       # write_one 只要这一个, 其余静默跳过
        folder_path = os.path.join(proc_root, folder)
        if not os.path.isdir(folder_path):
            continue
        m = FOLDER_RE.match(folder)
        if not m:
            print(f"[warn] {run_id}/{folder}: 文件夹名不符合命名规则，跳过")
            continue
        form_letter, platform, hash_prefix = m.groups()

        wf_path = os.path.join(folder_path, "workflow.json")
        if not os.path.isfile(wf_path):
            print(f"[warn] {run_id}/{folder}: 没有 workflow.json，跳过")
            continue
        workflow = load_json(wf_path)

        # 在对应 form 里按 case_id 前缀找匹配的 post
        form_data = get_form(form_letter)
        want_prefix = f"{platform}_{hash_prefix}"
        hits = [r for r in form_data.get("results", [])
                if r.get("case_id", "").startswith(want_prefix)]
        if len(hits) != 1:
            print(f"[warn] {run_id}/{folder}: 匹配到 {len(hits)} 条 result（期望 1），跳过")
            continue
        result = hits[0]

        # 可选：用 _source.json 的 link 校验映射没串台
        src_path = os.path.join(folder_path, "_source.json")
        if os.path.isfile(src_path):
            src = load_json(src_path)
            if src.get("link") and src["link"] != result.get("source_url"):
                print(f"[warn] {run_id}/{folder}: _source.link 与 result.source_url 不一致")

        # 以帖子为单位合并，只保留 5 个字段
        merged = {
            "query_id": run_id,
            "query": query,
            "platform": result.get("platform", platform),
            "post": result.get("post"),
            "llm_evaluation": result.get("llm_evaluation"),
            "workflow": workflow,
        }
        out.append((folder, merged))

    if not out:
        print(f"[skip] {run_id}: 没有可合并的 procedure")
    return out


def _dump_merged(run_id, folder, merged):
    """把单个帖子的 merged 写成 workflows/{run_id}_{folder}.json。"""
    os.makedirs(OUT_DIR, exist_ok=True)
    out_path = os.path.join(OUT_DIR, f"{run_id}_{folder}.json")
    with open(out_path, "w", encoding="utf-8") as f:
        json.dump(merged, f, ensure_ascii=False, indent=2)
    print(f"[ok] {run_id}/{folder} -> {os.path.basename(out_path)}")


def write_run(run_id, runs_dir=None):
    entries = build_run(run_id, runs_dir=runs_dir)
    if not entries:
        return 0
    for folder, merged in entries:
        _dump_merged(run_id, folder, merged)
    return len(entries)


def write_one(run_id, folder, runs_dir=None):
    """只把指定 procedure folder 的合并 json 写出 —— 跑完一个工序就立刻出一个,
    无需等同 q 其他帖子。命中并写出返回 1, 没匹配到 (缺 workflow.json 等) 返回 0。
    复用 build_run 的解析/校验逻辑, 只处理 folder 那一个 (only_folder 让同 q 其余
    未完成目录被静默跳过, 不刷 warn)。"""
    for f, merged in build_run(run_id, runs_dir=runs_dir, only_folder=folder):
        if f == folder:
            _dump_merged(run_id, folder, merged)
            return 1
    return 0


def build_runs(run_ids):
    """对一组 run 执行 write_run，返回写出的帖子 json 总数。"""
    total = 0
    for run_id in run_ids:
        total += write_run(run_id)
    return total


def all_run_ids():
    """runs_full 下所有 q* 目录。"""
    return sorted(d for d in os.listdir(RUNS_DIR)
                  if re.match(r"^q\d+$", d)
                  and os.path.isdir(os.path.join(RUNS_DIR, d)))


# ---------- 接口：实时扫描 workflows/ 并以数组返回 ----------

def scan_workflows():
    """实时扫描 workflows/*.json，把每个文件读成 dict，按文件名排序返回数组。

    每次调用都重新读盘，所以 build 新写入的文件会立刻在接口里出现（无缓存）。"""
    items = []
    for fp in sorted(glob.glob(os.path.join(OUT_DIR, "*.json"))):
        try:
            items.append(load_json(fp))
        except Exception as e:
            print(f"[warn] 读取 {os.path.basename(fp)} 失败：{e}")
    return items


class Handler(BaseHTTPRequestHandler):
    def _send(self, code, obj):
        body = json.dumps(obj, ensure_ascii=False).encode("utf-8")
        self.send_response(code)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Access-Control-Allow-Origin", "*")
        self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
        self.send_header("Access-Control-Allow-Headers", "Content-Type")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_OPTIONS(self):  # CORS 预检
        self._send(204, {})

    def do_GET(self):
        path = self.path.split("?")[0]
        if path in ("/", "/workflows", "/api/workflows"):
            self._send(200, scan_workflows())
        else:
            self._send(404, {"error": "not found"})

    def do_POST(self):
        if self.path.split("?")[0] != "/build":
            self._send(404, {"error": "not found"}); return
        length = int(self.headers.get("Content-Length") or 0)
        raw = self.rfile.read(length).decode("utf-8") if length > 0 else "{}"
        try:
            payload = json.loads(raw)
        except Exception as e:
            self._send(400, {"error": f"bad json: {e}"}); return
        if payload.get("all"):
            run_ids = all_run_ids()
        else:
            q = (payload.get("q") or "").strip()
            if not re.match(r"^q\d+$", q):  # 限定 qNN 形式，避免路径注入
                self._send(400, {"error": f"bad q (expect 'qNN' or all=true): {q!r}"}); return
            run_ids = [q]
        try:
            n = build_runs(run_ids)
            self._send(200, {"status": "ok", "runs": run_ids, "written": n})
        except Exception as e:
            self._send(500, {"error": f"build failed: {e}"})

    def log_message(self, *a):
        pass


def serve(port):
    n = len(scan_workflows())
    print(f"workflows 接口：http://0.0.0.0:{port}/workflows   "
          f"(workflows/ 下当前 {n} 个 json，实时扫描)")
    ThreadingHTTPServer(("0.0.0.0", port), Handler).serve_forever()


def main(argv):
    args = argv[1:]
    if args and args[0] == "serve":
        port = int(args[1]) if len(args) > 1 else DEFAULT_PORT
        serve(port)
        return

    if "--all" in args:
        run_ids = all_run_ids()
    elif args:
        run_ids = args
    else:
        run_ids = ["q0000"]
    total_files = build_runs(run_ids)
    print(f"\n完成：处理 {len(run_ids)} 个 run，共写出 {total_files} 个帖子 json")


if __name__ == "__main__":
    try:  # Windows 控制台默认 cp1252，中文 print 会崩，统一切 utf-8
        sys.stdout.reconfigure(encoding="utf-8")
    except Exception:
        pass
    main(sys.argv)