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

"""
三形式 query 批量搜索 + 多模态评估

针对 high_priority_queries.json 的前 N 条高优 query，每条 query 用三种形式搜索 + 评估：

  形式 A（原词组合）：直接用 item["q"]，如 "反推 提示词 教程"
  形式 B（句子填充）：gemini flash 把词组改写成自然搜索短句，**禁止注入具体工具/品牌/示例**
  形式 C（同义替换）：按 synonym_pools 对 动作/类型/知识词 各取同义词重组

输出（按 query 分文件夹）：
  output_dir/
    q00/  form_A.json  form_B.json  form_C.json
    q01/  ...
    ...
  forms_preview.json   # 三形式 query 预览
  summary.json         # 三形式对比汇总

每个 form_X.json = {query 词} ↔ {帖子源信息 + 评估结果}（一对多）。
搜索 / 评估 / 多模态图片逻辑复用 script/search_and_evaluate.py。

用法：
  python batch_3forms.py --count 10 --platforms xhs,gzh,zhihu --max-count 10 \
      --output-dir runs/3forms_001
"""

import argparse
import asyncio
import json
import random
import sys
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional, Tuple

_PROJECT_ROOT = Path(__file__).resolve().parents[4]
if str(_PROJECT_ROOT) not in sys.path:
    sys.path.insert(0, str(_PROJECT_ROOT))

from examples.process_pipeline.script.llm_helper import call_llm_with_retry
from examples.process_pipeline.script.search_and_evaluate import (
    search_all, evaluate_posts, transcribe_video_posts, build_query_overrides,
)
from examples.process_pipeline.script.llm_evaluate_sources import build_eval_llm_call

_EVAL_DIR = _PROJECT_ROOT / "examples" / "process_pipeline" / "test_script" / "evaluation"
_HIGH_PRIORITY = Path(__file__).resolve().parent / "high_priority_queries.json"
_SYNONYM_POOLS = _EVAL_DIR / "synonym_pools.json"


# ── 形式 A：原词组合 ─────────────────────────────────────────────────────────────

def form_a(items: List[Dict[str, Any]]) -> List[str]:
    return [it["q"] for it in items]


# ── 形式 B：gemini 句子化（禁止注入示例）─────────────────────────────────────────

def _validate_sentences(data: Dict[str, Any], n: int) -> Optional[str]:
    qs = data.get("sentences")
    if not isinstance(qs, list):
        return "sentences 必须是数组"
    if len(qs) != n:
        return f"sentences 长度应为 {n}，得到 {len(qs)}"
    if not all(isinstance(x, str) and x.strip() for x in qs):
        return "sentences 每项必须是非空字符串"
    return None


async def form_b(items: List[Dict[str, Any]], llm_call: Callable, model: str) -> Tuple[List[str], float]:
    """把每条词组改写成自然搜索短句（一次批量调用，按序对齐）。"""
    words = [it["q"] for it in items]
    system = (
        "你是中文搜索词改写器。把每个『关键词组』改写成一句自然、口语、适合在内容平台"
        "搜索框输入的短句。严格要求：只表达词组本身的意图，"
        "**绝不添加任何具体工具名 / 品牌 / 产品 / 模型名 / 风格名 / 数字示例**"
        "（如 Midjourney、赛博朋克、SD 等都禁止出现）。只输出 JSON。"
    )
    user = (
        "把下面每个词组改写成一句自然搜索短句，顺序一一对应，输出：\n"
        '{"sentences": ["短句1", "短句2", ...]}\n\n'
        f"词组列表（共 {len(words)} 个）：\n{json.dumps(words, ensure_ascii=False, indent=2)}"
    )
    data, cost = await call_llm_with_retry(
        llm_call=llm_call, messages=[{"role": "system", "content": system},
                                     {"role": "user", "content": user}],
        model=model, temperature=0.4, max_tokens=2000,
        validate_fn=lambda d: _validate_sentences(d, len(words)), task_name="FormB",
    )
    if not data:
        print("   ⚠️ form B 生成失败，回退用原词组")
        return list(words), cost
    return [s.strip() for s in data["sentences"]], cost


# ── 形式 C：同义替换重组 ─────────────────────────────────────────────────────────

class SynonymComposer:
    def __init__(self, pools: Dict[str, Any], rng: random.Random):
        self.action = pools.get("action_leaves", {})
        self.types = pools.get("types", {})
        self.knowledge = pools.get("knowledge", {})
        self.tool_type = pools.get("tool_type", {})
        self.rng = rng

    def _pick(self, pool: Any, fallback: str) -> str:
        pool = [x for x in pool if isinstance(x, str)] if isinstance(pool, list) else []
        return self.rng.choice(pool) if pool else fallback

    def compose(self, item: Dict[str, Any]) -> str:
        """按 synonym_pools._usage：[模态/工具前缀] 动作 类型 知识词。"""
        parts: List[str] = []
        c = item.get("constraint")
        if isinstance(c, dict):
            if c.get("kind") == "模态" and c.get("value"):
                parts.append(str(c["value"]))
            elif c.get("kind") == "工具类型":
                parts.append(self._pick(self.tool_type.get(c.get("value")), str(c.get("限定词") or "")))
        parts.append(self._pick(self.action.get(item.get("action", "")), item.get("action", "")))
        parts.append(self._pick(self.types.get(item.get("type", "")), item.get("type", "")))
        gx = self.knowledge.get("工序", {})
        parts.append(self._pick(gx.get("单步") if isinstance(gx, dict) else None, "教程"))
        return " ".join(p for p in parts if p)


def form_c(items: List[Dict[str, Any]], seed: int) -> List[str]:
    pools = json.loads(_SYNONYM_POOLS.read_text(encoding="utf-8"))
    composer = SynonymComposer(pools, random.Random(seed))
    return [composer.compose(it) for it in items]


# ── 单个 (query, form) 的搜索 + 评估 + 落盘 ──────────────────────────────────────

async def run_one(
    qtext: str, form_key: str, original_q: str,
    args, eval_llm, eval_model_id, out_file: Path,
    query_overrides=None,
) -> Dict[str, Any]:
    platforms = [p.strip() for p in args.platforms.split(",") if p.strip()]
    sources = await search_all(platforms, [qtext], args.max_count, args.max_concurrent,
                               query_overrides=query_overrides)

    try:
        from examples.process_pipeline.script.extract_sources import _convert_timestamps
        _convert_timestamps(sources)
    except Exception:
        pass

    # 视频帖转写：把字幕并入正文再评估（默认开）
    if not args.no_transcribe and sources:
        n = await transcribe_video_posts(sources, concurrency=args.max_concurrent)
        if n:
            print(f"     🎙️  视频转写 {n} 条")

    cost = 0.0
    if not args.no_eval and sources:
        # 评估只看 query 词 + 帖子：把该形式的搜索词 qtext 作为检索锚点
        sources, cost = await evaluate_posts(
            sources, "", eval_llm, eval_model_id, args.max_concurrent,
            include_images=not args.no_images, max_images=args.max_images,
            image_mode=args.image_mode, query=qtext,
        )
    for s in sources:
        imgs = s.pop("_image_data_urls", None)
        if imgs is not None:
            s["images_sent"] = len(imgs)

    rep = sum(1 for s in sources
              if ((s.get("llm_evaluation") or {}).get("制作相关性") or {}).get("得分") in (2, 3, 2.0, 3.0, "2", "3"))
    dis = sum(1 for s in sources
              if ((s.get("llm_evaluation") or {}).get("制作相关性") or {}).get("得分") in (1, 1.0, "1"))
    failed = sum(1 for s in sources if (s.get("llm_evaluation") or {}).get("_error"))

    out_file.parent.mkdir(parents=True, exist_ok=True)
    out_file.write_text(json.dumps({
        "form": form_key,
        "query": qtext,            # 该形式实际搜索用的词（也是评估的检索锚点）
        "original_q": original_q,  # 原词组（形式 A 的基准）
        "platforms": platforms,
        "total": len(sources), "report": rep, "discard": dis, "failed": failed,
        "results": sources,        # 帖子源信息 + llm_evaluation，一对多
    }, ensure_ascii=False, indent=2), encoding="utf-8")
    print(f"     [{form_key}] {qtext!r} → total={len(sources)} report={rep} discard={dis} "
          f"failed={failed} cost=${cost:.4f}")
    return {"form": form_key, "total": len(sources), "report": rep,
            "discard": dis, "failed": failed, "cost": round(cost, 4)}


async def reeval_existing(args, eval_llm, eval_model_id) -> None:
    """只重跑评估、覆盖旧评估，不重新搜索。

    读 output_dir 下已有的 q*/form_*.json，对里面已抓到的 post 重新评估（评估锚点 = 文件里
    记录的该形式 query 词），原地覆盖 llm_evaluation 后写回。适合改了评估 prompt / 模型后复评。

    用 --start / --count 在 q 编号层（自然数序）切片限制范围，与主流程同语义；每个 q 文件夹下
    的所有 form_A/B/C.json 一起复评（三种形式可比性）。
    """
    import re
    output_dir = Path(args.output_dir)

    # 按 q 编号自然数排序：避免 "q10" < "q2" 这种字符串误排（与 server.py _qnum 同口径）
    def _qnum(p):
        m = re.search(r"\d+", p.name)
        return (int(m.group()) if m else 0, p.name)
    q_dirs = sorted([d for d in output_dir.glob("q*") if d.is_dir()], key=_qnum)
    if not q_dirs:
        print(f"❌ {output_dir} 下没有 q*/ 子目录，无可复评内容"); return

    # --reeval-q 优先于 --start/--count：直接按 q 名过滤（接 "q01" 或 "q01,q05,q12" 多选）
    reeval_q = getattr(args, 'reeval_q', None)
    if reeval_q:
        wanted = {x.strip() for x in reeval_q.split(',') if x.strip()}
        sliced = [d for d in q_dirs if d.name in wanted]
        if not sliced:
            print(f"[X] 指定 q ({reeval_q}) 在 {output_dir} 下不存在"); return
        range_label = f"q={','.join(d.name for d in sliced)}"
    else:
        sliced = q_dirs[args.start : args.start + args.count]
        range_label = f"q[{args.start}:{args.start + args.count}]"
    files = [f for qd in sliced for f in sorted(qd.glob("form_*.json"))]
    if not files:
        print(f"❌ {output_dir} 切片 {range_label} 下没有 form_*.json"); return
    print(f"♻️  复评模式：{range_label} → {len(sliced)} 个 query / "
          f"{len(files)} 个文件，模型 {eval_model_id}（不重新搜索）")

    for f in files:
        d = json.loads(f.read_text(encoding="utf-8"))
        results = d.get("results", [])
        if not results:
            print(f"   - {f.relative_to(output_dir)}: 空，跳过"); continue
        # 清掉旧评估痕迹，重新评
        for s in results:
            s.pop("llm_evaluation", None)
            s.pop("images_sent", None)
            s.pop("_image_data_urls", None)
        qtext = d.get("query", "")   # 该形式实际搜索词 = 评估检索锚点

        if not args.no_transcribe and results:
            await transcribe_video_posts(results, concurrency=args.max_concurrent)

        results, cost = await evaluate_posts(
            results, "", eval_llm, eval_model_id, args.max_concurrent,
            include_images=not args.no_images, max_images=args.max_images,
            image_mode=args.image_mode, query=qtext,
        )
        for s in results:
            imgs = s.pop("_image_data_urls", None)
            if imgs is not None:
                s["images_sent"] = len(imgs)

        rep = sum(1 for s in results
              if ((s.get("llm_evaluation") or {}).get("制作相关性") or {}).get("得分") in (2, 3, 2.0, 3.0, "2", "3"))
        dis = sum(1 for s in results
              if ((s.get("llm_evaluation") or {}).get("制作相关性") or {}).get("得分") in (1, 1.0, "1"))
        failed = sum(1 for s in results if (s.get("llm_evaluation") or {}).get("_error"))
        d.update({"results": results,
                  "total": len(results), "report": rep, "discard": dis, "failed": failed})
        d.pop("requirement", None)   # 不再用 requirement
        f.write_text(json.dumps(d, ensure_ascii=False, indent=2), encoding="utf-8")
        print(f"   ✓ {f.relative_to(output_dir)}: total={len(results)} report={rep} "
              f"discard={dis} failed={failed} cost=${cost:.4f}")
    print("♻️  复评完成（已覆盖原文件）")


async def append_existing(args, eval_llm, eval_model_id, gen_llm, gen_model_id) -> None:
    """往已有 q*/form_*.json 追加新渠道结果，不重搜旧渠道。

    用文件里存的 query 词、只搜 --platforms 指定的新渠道，评估后按 (平台, id) 去重合并进
    原 results，旧渠道结果原样保留。适合先跑了中文渠道、再补 youtube/x 等。
    """
    from examples.process_pipeline.script.extract_sources import _convert_timestamps
    output_dir = Path(args.output_dir)
    files = sorted(output_dir.glob("q*/form_*.json"))
    if not files:
        print(f"❌ {output_dir} 下没有 q*/form_*.json，无可追加目标"); return
    new_plats = [p.strip() for p in args.platforms.split(",") if p.strip()]
    print(f"➕ 追加模式：{len(files)} 个文件追加渠道 {new_plats}（不重搜旧渠道）")

    # 英文平台一次性翻译所有 query
    queries = list(dict.fromkeys(json.loads(f.read_text(encoding="utf-8")).get("query", "") for f in files))
    overrides = await build_query_overrides(new_plats, queries, gen_llm, gen_model_id)

    for f in files:
        d = json.loads(f.read_text(encoding="utf-8"))
        qtext = d.get("query", "")
        existing = d.get("results", [])
        existing_keys = {(r.get("platform"), r.get("channel_content_id")) for r in existing}

        new_sources = await search_all(new_plats, [qtext], args.max_count, args.max_concurrent,
                                       query_overrides=overrides)
        new_sources = [s for s in new_sources
                       if (s.get("platform"), s.get("channel_content_id")) not in existing_keys]
        try:
            _convert_timestamps(new_sources)
        except Exception:
            pass
        if not args.no_transcribe and new_sources:
            await transcribe_video_posts(new_sources, concurrency=args.max_concurrent)

        cost = 0.0
        if not args.no_eval and new_sources:
            new_sources, cost = await evaluate_posts(
                new_sources, "", eval_llm, eval_model_id, args.max_concurrent,
                include_images=not args.no_images, max_images=args.max_images,
                image_mode=args.image_mode, query=qtext,
            )
        for s in new_sources:
            imgs = s.pop("_image_data_urls", None)
            if imgs is not None:
                s["images_sent"] = len(imgs)

        merged = existing + new_sources
        plats_union = list(dict.fromkeys((d.get("platforms") or []) + new_plats))
        rep = sum(1 for s in merged
              if ((s.get("llm_evaluation") or {}).get("制作相关性") or {}).get("得分") in (2, 3, 2.0, 3.0, "2", "3"))
        dis = sum(1 for s in merged
              if ((s.get("llm_evaluation") or {}).get("制作相关性") or {}).get("得分") in (1, 1.0, "1"))
        failed = sum(1 for s in merged if (s.get("llm_evaluation") or {}).get("_error"))
        d.update({"platforms": plats_union, "results": merged,
                  "total": len(merged), "report": rep, "discard": dis, "failed": failed})
        f.write_text(json.dumps(d, ensure_ascii=False, indent=2), encoding="utf-8")
        print(f"   ✓ {f.relative_to(output_dir)}: +{len(new_sources)} 新帖 → total={len(merged)} "
              f"report={rep} discard={dis} failed={failed} cost=${cost:.4f}")
    print("➕ 追加完成（已并入原文件）")


async def run(args) -> None:
    eval_llm0, eval_model0 = build_eval_llm_call(args.eval_model)
    if args.reeval:
        await reeval_existing(args, eval_llm0, eval_model0)
        return
    if args.append:
        gen_llm0, gen_model0 = build_eval_llm_call(args.gen_model)
        await append_existing(args, eval_llm0, eval_model0, gen_llm0, gen_model0)
        return

    queries_file = Path(args.queries_file) if getattr(args, "queries_file", None) else _HIGH_PRIORITY
    all_items = json.loads(queries_file.read_text(encoding="utf-8"))["queries"]
    print(f"📂 query 源: {queries_file.name} ({len(all_items)} 条)")
    only_q = getattr(args, "only_q", None)
    if only_q:
        # 支持 "1,5,51" 或 "q01,q05,q51"；优先级高于 --start/--count
        import re as _re
        raw = [t.strip() for t in only_q.split(",") if t.strip()]
        idxs = []
        for t in raw:
            m = _re.match(r"q?(\d+)$", t)
            if not m:
                print(f"⚠️ 忽略无法解析的 q: {t!r}"); continue
            i = int(m.group(1))
            if 0 <= i < len(all_items):
                idxs.append(i)
            else:
                print(f"⚠️ idx {i} 超出范围 [0,{len(all_items)})，忽略")
        idxs = sorted(dict.fromkeys(idxs))   # 去重保序
        if not idxs:
            print("❌ --only-q 没有合法索引可用"); return
        items = [all_items[i] for i in idxs]
        print(f"📋 取 high_priority 指定 {len(idxs)} 条 query (idx={','.join(map(str, idxs))})"
              f" | 渠道 {args.platforms} | 每渠道≤{args.max_count} 帖")
    else:
        start = args.start
        items = all_items[start:start + args.count]
        idxs = list(range(start, start + len(items)))   # 绝对下标，用于文件夹命名
        print(f"📋 取 high_priority 第 {start}~{start+len(items)-1} 条 query（共 {len(items)} 条）"
              f" | 渠道 {args.platforms} | 每渠道≤{args.max_count} 帖")

    eval_llm, eval_model_id = build_eval_llm_call(args.eval_model)
    gen_llm, gen_model_id = build_eval_llm_call(args.gen_model)
    print(f"🧠 评估模型 {args.eval_model}->{eval_model_id} | form B 生成 {args.gen_model}->{gen_model_id}")

    output_dir = Path(args.output_dir)
    output_dir.mkdir(parents=True, exist_ok=True)

    qa = form_a(items)
    qb, b_cost = await form_b(items, gen_llm, gen_model_id)
    qc = form_c(items, args.seed)
    # forms_preview 用绝对下标做 key，多次区间跑不会互相覆盖
    preview_path = output_dir / "forms_preview.json"
    preview = {}
    if preview_path.exists():
        try:
            loaded = json.loads(preview_path.read_text(encoding="utf-8"))
            if isinstance(loaded, dict):
                preview = loaded   # 旧版本写成 list，非 dict 一律重置
        except Exception:
            preview = {}
    for j, absi in enumerate(idxs):
        preview[str(absi)] = {"idx": absi, "A": qa[j], "B": qb[j], "C": qc[j]}
    preview_path.write_text(json.dumps(preview, ensure_ascii=False, indent=2), encoding="utf-8")
    print("📝 三形式预览 → forms_preview.json")
    for j, absi in enumerate(idxs):
        print(f"   [{absi}] A={qa[j]!r}  B={qb[j]!r}  C={qc[j]!r}")

    # 英文平台(youtube/x)：对全部形式的 query 一次性翻成英文
    platforms = [p.strip() for p in args.platforms.split(",") if p.strip()]
    all_q = list(dict.fromkeys(qa + qb + qc))
    overrides = await build_query_overrides(platforms, all_q, gen_llm, gen_model_id)

    summary = []
    for j, absi in enumerate(idxs):
        qdir = output_dir / f"q{absi:02d}"
        print(f"\n▶ q{absi:02d}  原词={qa[j]!r}")
        per_form = {}
        for fk, qtext in (("A", qa[j]), ("B", qb[j]), ("C", qc[j])):
            stat = await run_one(qtext, fk, qa[j], args, eval_llm, eval_model_id,
                                 qdir / f"form_{fk}.json", query_overrides=overrides)
            per_form[fk] = stat
        summary.append({"idx": absi, "q": qa[j], "forms": per_form})

    (output_dir / "summary.json").write_text(json.dumps({
        "count": len(items), "platforms": args.platforms, "eval_model": eval_model_id,
        "form_b_gen_cost": round(b_cost, 4), "per_query": summary,
    }, ensure_ascii=False, indent=2), encoding="utf-8")

    # 形式聚合对比
    print(f"\n{'='*60}\n📊 三形式对比 (各形式 report/total 合计)")
    for fk in ("A", "B", "C"):
        tot = sum(s["forms"][fk]["total"] for s in summary)
        rep = sum(s["forms"][fk]["report"] for s in summary)
        dis = sum(s["forms"][fk]["discard"] for s in summary)
        print(f"   形式 {fk}: report={rep}/{tot} discard={dis}")
    print(f"→ {output_dir/'summary.json'}")


def main() -> None:
    from dotenv import load_dotenv
    load_dotenv()
    from examples.process_pipeline.script.llm_evaluate_sources import EVAL_MODELS

    p = argparse.ArgumentParser(description="三形式 query 批量搜索 + 多模态评估")
    p.add_argument("--start", type=int, default=0, help="起始 query 下标（0-based，默认 0）")
    p.add_argument("--count", type=int, default=10, help="从 --start 起取几条 query（默认 10）")
    p.add_argument("--only-q", default=None,
                   help="离散指定要跑的 q（如 '51,55,331' 或 'q51,q55,q331'），优先于 --start/--count")
    p.add_argument("--queries-file", default=None,
                   help="自定义 query 源 JSON 路径（结构需含 queries[...]），默认读 high_priority_queries.json")
    p.add_argument("--platforms", default="xhs,gzh,zhihu", help="逗号分隔渠道（默认 xhs,gzh,zhihu）")
    p.add_argument("--max-count", type=int, default=10, help="每个 (渠道,query) 取几条帖子（默认 10）")
    p.add_argument("--output-dir", required=True, help="输出目录")
    p.add_argument("--eval-model", default="gemini-flash-lite", choices=list(EVAL_MODELS),
                   help="评估模型（默认 gemini-flash-lite，多模态）")
    p.add_argument("--gen-model", default="gemini-flash-lite", choices=list(EVAL_MODELS),
                   help="form B 句子生成模型（默认 gemini-flash-lite）")
    p.add_argument("--max-concurrent", type=int, default=3, help="搜索 / 评估并发上限")
    p.add_argument("--max-images", type=int, default=4, help="每帖最多发给模型几张配图")
    p.add_argument("--image-mode", choices=["url", "base64"], default="url",
                   help="图片传输：url 直传(快，默认) / base64 下载内嵌(稳)")
    p.add_argument("--no-images", action="store_true", help="不发图（纯文本评估）")
    p.add_argument("--no-transcribe", action="store_true",
                   help="不对视频帖转写（默认会转写并把字幕并入正文再评估）")
    p.add_argument("--no-eval", action="store_true", help="只搜不评估")
    p.add_argument("--reeval", action="store_true",
                   help="只重跑评估、覆盖 output-dir 下已有 q*/form_*.json（不重新搜索）；"
                        "用 --start / --count 在 q 编号层限范围，或 --reeval-q 直接指定")
    p.add_argument("--reeval-q", default=None,
                   help="仅复评指定的 q（如 'q01' 或 'q01,q05,q12'），优先于 --start/--count")
    p.add_argument("--append", action="store_true",
                   help="往已有 q*/form_*.json 追加 --platforms 指定的新渠道结果（不重搜旧渠道）")
    p.add_argument("--seed", type=int, default=42, help="form C 同义替换随机种子")
    args = p.parse_args()
    asyncio.run(run(args))


if __name__ == "__main__":
    main()