Agent/examples/process_pipeline/script/search_and_evaluate.py

"""
独立脚本：query 词多渠道搜索 + qwen/LLM 逐条评估（不经过 agent，也不写 source.json）

定位：一个自包含的「搜 + 评」工具。
  输入  ：一组 query 词（json 文件）
  可选  ：先用 LLM（默认 qwen）改写 / 扩展 query
  搜索  ：直接调各渠道 search_impl 拿 post 详情（绕开 agent）
  评估  ：把每条 post 详情交给 LLM，按 rubric 逐条评估
  输出  ：evaluated.json —— 每条 = post 详情 + llm_evaluation

与 run_pipeline / llm_evaluate_sources 的区别：这里不维护 source.json / filtered_cases，
不做跨轮去重与回写，只产出一份评估结果，方便单独跑、单独看。
评估的 rubric prompt 与单帖评估逻辑仍复用 llm_evaluate_sources（忠实 rubric）。

queries.json 支持两种格式：
  ["query1", "query2", ...]
  {"requirement": "采集目标描述", "queries": ["query1", "query2", ...]}

典型用法：
  # 直接用给定 query 搜索 + qwen 评估
  python search_and_evaluate.py --queries q.json --platforms xhs,zhihu \
      --output-dir scratch/run1 --eval-model qwen
  # 先让 qwen 改写 query 再搜
  python search_and_evaluate.py --queries q.json --platforms xhs \
      --output-dir scratch/run1 --gen-query --gen-model qwen --keep-original
  # 只搜不评估
  python search_and_evaluate.py --queries q.json --platforms xhs --output-dir d --no-eval
"""

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

_PROJECT_ROOT = Path(__file__).resolve().parents[3]
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

logger = logging.getLogger(__name__)


# ── queries 加载 ────────────────────────────────────────────────────────────────

def load_queries(path: Path) -> Tuple[List[str], str]:
    """读 queries.json，返回 (queries, requirement)。requirement 可能为空串。"""
    with open(path, "r", encoding="utf-8") as f:
        data = json.load(f)
    if isinstance(data, list):
        return [str(q).strip() for q in data if str(q).strip()], ""
    if isinstance(data, dict):
        raw = data.get("queries") or []
        queries = [str(q).strip() for q in raw if str(q).strip()]
        return queries, str(data.get("requirement") or "").strip()
    raise ValueError(f"无法识别的 queries 文件格式: {path}（应为数组或 {{queries:[...]}}）")


# ── query 生成 / 改写 ───────────────────────────────────────────────────────────

def _validate_gen(data: Dict[str, Any]) -> Optional[str]:
    qs = data.get("queries")
    if not isinstance(qs, list) or not qs:
        return "queries 必须是非空数组"
    if not all(isinstance(q, str) and q.strip() for q in qs):
        return "queries 每一项必须是非空字符串"
    return None


async def generate_queries(
    base_queries: List[str],
    requirement: str,
    llm_call: Callable,
    model: str,
    target_count: int,
) -> Tuple[List[str], float]:
    """让 LLM 基于采集需求改写 / 扩展已有 query。返回 (新 query 列表, cost)。"""
    system = (
        "你是内容采集的搜索词优化器。基于采集需求和已有 query，产出一组更适合在"
        "社媒/内容平台搜索框直接使用的关键词：覆盖同义表达、具体工具名、典型用法场景，"
        "去掉过于宽泛或重复的词。只输出一个 JSON 对象，不要解释、不要 markdown。"
    )
    user = (
        f"【采集需求 / 目标格子】\n{requirement or '(未提供，参考已有 query 自行归纳)'}\n\n"
        f"【已有 query】\n{json.dumps(base_queries, ensure_ascii=False)}\n\n"
        f"【要求】产出约 {target_count} 个改写 / 扩展后的搜索词，输出格式：\n"
        '{"queries": ["词1", "词2", ...]}\n'
        "词应简短（适合搜索框）、彼此不同、贴合『制作做法』而非泛泛话题。只输出 JSON。"
    )
    data, cost = await call_llm_with_retry(
        llm_call=llm_call,
        messages=[{"role": "system", "content": system},
                  {"role": "user", "content": user}],
        model=model, temperature=0.5, max_tokens=1500,
        validate_fn=_validate_gen, task_name="GenQuery",
    )
    if not data:
        logger.warning("query 生成失败，回退使用原始 query")
        return [], cost
    out, seen = [], set()
    for q in data["queries"]:
        q = q.strip()
        if q and q not in seen:
            seen.add(q)
            out.append(q)
    return out, cost


# ── 渠道搜索 ────────────────────────────────────────────────────────────────────

# 英文为主的平台：搜索前把中文 query 翻成英文（其余平台用中文原词）
_EN_PLATFORMS = {"youtube", "x"}


def _validate_translation(data: Dict[str, Any], n: int) -> Optional[str]:
    t = data.get("translations")
    if not isinstance(t, list) or len(t) != n:
        return f"translations 必须是长度 {n} 的数组"
    if not all(isinstance(x, str) and x.strip() for x in t):
        return "translations 每项必须是非空字符串"
    return None


async def build_query_overrides(
    platforms: List[str], queries: List[str], llm_call: Callable, model: str,
) -> Dict[Tuple[str, str], str]:
    """为英文平台(youtube/x)把 queries 翻成英文搜索词，返回 {(platform, q): en_q}。

    无英文平台 → 返回空 dict（不调用 LLM）。翻译失败 → 回退用原中文（返回空，按原词搜）。
    """
    en_plats = [p for p in platforms if p in _EN_PLATFORMS]
    if not en_plats or not queries:
        return {}
    uniq = list(dict.fromkeys(queries))
    system = (
        "你是搜索词翻译器。把每个中文搜索词组翻译成地道的英文搜索关键词，"
        "面向 YouTube / X(Twitter) 搜索框：用该领域英文母语者真实会搜的说法（含常用工具/术语英文名），"
        "简短、可直接搜，不要整句解释。只输出 JSON。"
    )
    user = (
        "把下面中文搜索词逐个翻译成英文搜索关键词，顺序一一对应，输出：\n"
        '{"translations": ["en1", "en2", ...]}\n\n'
        f"中文词（共 {len(uniq)} 个）：\n{json.dumps(uniq, ensure_ascii=False, indent=2)}"
    )
    data, _ = await call_llm_with_retry(
        llm_call=llm_call, messages=[{"role": "system", "content": system},
                                     {"role": "user", "content": user}],
        model=model, temperature=0.2, max_tokens=1500,
        validate_fn=lambda d: _validate_translation(d, len(uniq)), task_name="TranslateQuery",
    )
    if not data:
        logger.warning("query 翻译失败，英文平台回退用中文原词")
        return {}
    zh2en = {zh: en.strip() for zh, en in zip(uniq, data["translations"])}
    overrides = {(p, q): zh2en[q] for p in en_plats for q in queries if q in zh2en}
    print(f"🌐 英文平台 {en_plats} query 翻译：" +
          "；".join(f"{q}→{zh2en[q]}" for q in uniq[:5]) + (" …" if len(uniq) > 5 else ""))
    return overrides


def _post_cid(post: Dict[str, Any]) -> Optional[str]:
    cid = post.get("channel_content_id") or post.get("video_id")
    if cid:
        return str(cid)
    link = post.get("link") or post.get("url")
    return str(link) if link else None


async def _search_one(pdef, keyword: str, orig_q: str, max_count: int, sem: asyncio.Semaphore):
    """跑一次搜索（实际用 keyword，但回溯仍记原始 orig_q）。返回 (platform, orig_q, posts)。

    keyword 可能是为英文平台翻译后的词；orig_q 是用户的原始 query，用于 found_by_queries
    与跨渠道/跨形式对比的统一锚定。
    """
    async with sem:
        try:
            result = await pdef.search_impl(
                platform_id=pdef.id, keyword=keyword, max_count=max_count,
                cursor="", extras=None,
            )
        except Exception as e:
            logger.warning("search 失败 [%s/%s]: %s", pdef.id, keyword, e)
            return pdef.id, orig_q, []
    if getattr(result, "error", None):
        logger.warning("search 返回错误 [%s/%s]: %s", pdef.id, keyword, result.error)
        return pdef.id, orig_q, []
    posts = (result.metadata or {}).get("posts", []) or []
    return pdef.id, orig_q, posts


async def search_all(
    platforms: List[str], queries: List[str], max_count: int, max_concurrent: int,
    query_overrides: Optional[Dict[Tuple[str, str], str]] = None,
) -> List[Dict[str, Any]]:
    """对所有 (platform × query) 组合并发搜索，按 (platform, cid) 去重。

    query_overrides: {(platform, query): 实际搜索词}。命中则用覆盖词搜（如英文平台用译文），
    未命中用原 query。found_by_queries 始终记原 query。

    返回 source_dict 列表，每条带：case_id / platform / channel_content_id /
    source_url / post / comments / found_by_queries（命中它的 query，用于回溯 query 质量）。
    """
    import agent.tools.builtin.content.tools  # noqa: F401 触发平台自注册
    from agent.tools.builtin.content.registry import get_platform, all_platforms

    pdefs = []
    for p in platforms:
        pdef = get_platform(p)
        if not pdef:
            avail = ", ".join(x.id for x in all_platforms())
            raise ValueError(f"未知平台 '{p}'。可用: {avail}")
        if not pdef.search_impl:
            raise ValueError(f"平台 '{p}' 不支持搜索")
        pdefs.append(pdef)

    ov = query_overrides or {}
    sem = asyncio.Semaphore(max_concurrent)
    tasks = [_search_one(pdef, ov.get((pdef.id, q), q), q, max_count, sem)
             for pdef in pdefs for q in queries]
    print(f"🔎 搜索 {len(pdefs)} 渠道 × {len(queries)} query = {len(tasks)} 次请求 (并发 {max_concurrent})")
    results = await asyncio.gather(*tasks)

    # youtube / sph 的 post 字段与通用 schema 不同（youtube: video_id/description_snippet/
    # thumbnails/url），统一 normalize 成 channel_content_id/body_text/images/link，
    # 否则下游取 cid / 正文 / 图片会漏。其他平台 no-op。
    try:
        from examples.process_pipeline.script.extract_sources import _normalize_post_in_place
    except Exception:
        _normalize_post_in_place = None

    collected: Dict[Tuple[str, str], Dict[str, Any]] = {}
    per_query_counts: Dict[str, int] = {}
    for platform, query, posts in results:
        per_query_counts[f"{platform}/{query}"] = len(posts)
        for post in posts:
            if not isinstance(post, dict):
                continue
            if _normalize_post_in_place:
                try:
                    _normalize_post_in_place(platform, post)
                except Exception:
                    pass
            cid = _post_cid(post)
            if not cid:
                continue
            key = (platform, cid)
            if key in collected:
                collected[key]["found_by_queries"].append(query)
                continue
            link = post.get("link") or post.get("url") or ""
            collected[key] = {
                "case_id": f"{platform}_{cid}",
                "platform": platform,
                "channel_content_id": cid,
                "source_url": link,
                "post": post,
                "comments": post.get("author_comments", []) or [],
                "found_by_queries": [query],
            }

    print("   每个 (渠道/query) 命中数：")
    for k, n in sorted(per_query_counts.items()):
        print(f"     - {k}: {n}")
    print(f"   去重后唯一 post：{len(collected)}")
    return list(collected.values())


# ── 图片获取（下载转 base64 data URL，喂多模态评估）──────────────────────────────

import base64

_MIME_BY_EXT = {".png": "image/png", ".webp": "image/webp", ".gif": "image/gif"}


def _collect_post_image_urls(post: Dict[str, Any], max_images: int) -> List[str]:
    """从 post 收集图片 URL（复用 generate_case 的字段映射），截断到 max_images。"""
    try:
        from examples.process_pipeline.script.generate_case import _extract_raw_images
        urls = _extract_raw_images(post, post.get("channel") or "")
    except Exception:
        urls = post.get("images") or []
    urls = [u for u in urls if isinstance(u, str) and u.startswith("http")]
    return urls[:max_images]


async def _fetch_data_url(url: str, sem: asyncio.Semaphore) -> Optional[str]:
    """下载单张图片转 base64 data URL（用项目 _download_image，带 Referer/UA 绕防盗链）。"""
    from agent.tools.builtin.file.image_cdn import _download_image
    async with sem:
        try:
            data = await _download_image(url)
        except Exception as e:
            logger.warning("图片下载失败 %s: %s", url[:60], e)
            return None
    if not data:
        return None
    ext = next((e for e in _MIME_BY_EXT if url.lower().split("?")[0].endswith(e)), "")
    mime = _MIME_BY_EXT.get(ext, "image/jpeg")
    b64 = base64.b64encode(data).decode("ascii")
    return f"data:{mime};base64,{b64}"


async def _attach_image_refs(
    sources: List[Dict[str, Any]], max_images: int, max_concurrent: int, mode: str,
) -> int:
    """为每条 source 收集配图，挂到 s['_image_data_urls']（评估后会清掉，不写进报告）。

    mode="url"   ：直接用 http 图片 URL，让 OpenRouter/Gemini 服务端去抓（实测 xhs 可直连，最快）。
    mode="base64"：下载图片转 base64 data URL（带 Referer/UA 绕防盗链，最稳，payload 较大）。
    """
    if mode == "url":
        total = 0
        for s in sources:
            urls = _collect_post_image_urls(s.get("post", {}) or {}, max_images)
            s["_image_data_urls"] = urls
            total += len(urls)
        return total

    # base64 模式：并发下载
    sem = asyncio.Semaphore(max_concurrent)
    plan: List[Tuple[Dict[str, Any], List[str]]] = []
    for s in sources:
        urls = _collect_post_image_urls(s.get("post", {}) or {}, max_images)
        plan.append((s, urls))
    flat = [(s, u) for s, urls in plan for u in urls]
    results = await asyncio.gather(*[_fetch_data_url(u, sem) for _, u in flat])
    bucket: Dict[int, List[str]] = {}
    for (s, _), data_url in zip(flat, results):
        if data_url:
            bucket.setdefault(id(s), []).append(data_url)
    total = 0
    for s in sources:
        s["_image_data_urls"] = bucket.get(id(s), [])
        total += len(s["_image_data_urls"])
    return total


# ── 视频字幕：转写并并入正文（与 run_pipeline / extract_sources 同口径）──────────

async def transcribe_video_posts(sources: List[Dict[str, Any]], concurrency: int = 3) -> int:
    """对视频帖跑 Deepgram 转写，把字幕并入 body_text（带 [视频字幕] 标记）再评估。

    视频帖（抖音/视频号/youtube 等）正文常为空或只有 hashtag，真正内容在视频里。
    复用 extract_sources 的逻辑：有视频源且还没 video_transcript 字段的帖子才转写，
    转写结果合并进 body_text —— 这样评估(_format_post_for_eval 读 body_text)就能看到视频内容。
    返回本次获得字幕的条数。
    """
    try:
        from examples.process_pipeline.script.extract_sources import (
            _transcribe_pending_async, _merge_transcript_into_body,
        )
    except Exception as e:
        logger.warning("转写模块导入失败，跳过: %s", e)
        return 0
    try:
        updates = await _transcribe_pending_async(sources, concurrency=concurrency)
    except Exception as e:
        logger.warning("视频转写失败，跳过: %s", e)
        return 0
    for s in sources:
        post = s.get("post")
        if isinstance(post, dict) and post.get("video_transcript"):
            merged = _merge_transcript_into_body(post)
            if merged is not post:
                post["body_text"] = merged.get("body_text", post.get("body_text", ""))
    return len(updates)


# ── 逐条评估（直评，不写 source.json）─────────────────────────────────────────────

async def evaluate_posts(
    sources: List[Dict[str, Any]],
    requirement: str,
    llm_call: Callable,
    model: str,
    max_concurrent: int,
    include_images: bool = True,
    max_images: int = 4,
    image_mode: str = "url",
    query: Optional[str] = None,
) -> Tuple[List[Dict[str, Any]], float]:
    """对每条 post 用 rubric 逐条评估，把 llm_evaluation 挂到 source 上。返回 (sources, total_cost)。

    复用 llm_evaluate_sources 的 rubric 加载与单帖评估逻辑，保证与管线评估口径一致。
    query 非空时作为检索锚点喂给模型（判相关性看『这帖是否回答了这个检索词』）。
    include_images=True 时把帖子配图一并发给模型做多模态评估：
      image_mode="url"   → 直接传图片 URL（最快，实测 xhs 可直连）
      image_mode="base64"→ 下载转 base64（最稳，绕防盗链）
    评估失败（重试耗尽）的条目标 error 标记并保留，不丢。
    """
    from examples.process_pipeline.script.llm_evaluate_sources import _evaluate_one
    # rubric 详解 / 输出 schema 已固化在 eval_prompt_template.md, 不再 load 外部 rubric 文件
    sem = asyncio.Semaphore(max_concurrent)

    if include_images:
        n_img = await _attach_image_refs(sources, max_images, max_concurrent * 2, image_mode)
        verb = "直传 URL" if image_mode == "url" else "下载 base64"
        print(f"🖼️  配图 {n_img} 张（每帖≤{max_images}，{verb}）用于多模态评估")

    print(f"🧠 逐条评估 {len(sources)} 条 (并发 {max_concurrent}) ...")
    results = await asyncio.gather(*[
        _evaluate_one(s, requirement, llm_call, model, sem,
                      image_urls=(s.get("_image_data_urls") if include_images else None),
                      query=query)
        for s in sources
    ])

    # mod.md 中文 schema 不再有 decision/instructive_pass 字段 —— rep/dis 统计基于
    # 制作相关性得分(1=明确无关 → 视作 discard 候选, 2/3 视作 report 候选)。失败仍按 error 标记。
    total_cost = 0.0
    rep = dis = failed = 0
    for s, (llm_eval, cost) in zip(sources, results):
        total_cost += cost
        if llm_eval is None:
            s["llm_evaluation"] = {"_error": True, "_reason": "llm_eval_failed"}
            failed += 1
            pr_score = "?"
        else:
            s["llm_evaluation"] = llm_eval
            pr_obj = llm_eval.get("制作相关性") or {}
            pr_score = pr_obj.get("得分", "?")
            try:
                if int(float(pr_score)) <= 1:
                    dis += 1
                else:
                    rep += 1
            except (TypeError, ValueError):
                rep += 1   # 无法解析得分时按 report 处理(prompt 引导失败,不强淘)
        title = (s.get("post", {}) or {}).get("title", "")[:30]
        print(f"   - [制作相关性={pr_score}] {s['case_id'][:24]} {title}")

    print(f"   汇总: 制作相关>=2 {rep} / 明确无关(1) {dis} / 评估失败 {failed} / cost=${total_cost:.4f}")
    return sources, total_cost


# ── 主流程 ────────────────────────────────────────────────────────────────────

async def run(args: argparse.Namespace) -> None:
    from examples.process_pipeline.script.llm_evaluate_sources import build_eval_llm_call

    queries, req_from_file = load_queries(Path(args.queries))
    if not queries:
        print("❌ queries 为空"); sys.exit(1)

    platforms = [p.strip() for p in args.platforms.split(",") if p.strip()]
    if not platforms:
        print("❌ 未指定 --platforms"); sys.exit(1)

    requirement = args.requirement or req_from_file or ("；".join(queries))[:200]

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

    print(f"📋 需求: {requirement[:80]}")
    print(f"📡 渠道: {platforms} | 原始 query 数: {len(queries)}")

    total_cost = 0.0

    # 1. 可选：LLM 改写 / 扩展 query
    if args.gen_query:
        gen_llm, gen_model_id = build_eval_llm_call(args.gen_model)
        print(f"✍️  query 生成模型: {args.gen_model} -> {gen_model_id}")
        gen_queries, gen_cost = await generate_queries(
            queries, requirement, gen_llm, gen_model_id, args.gen_count,
        )
        total_cost += gen_cost
        if gen_queries:
            search_queries = list(dict.fromkeys(
                (queries + gen_queries) if args.keep_original else gen_queries
            ))
            (output_dir / "generated_queries.json").write_text(
                json.dumps({"requirement": requirement, "original": queries,
                            "generated": gen_queries, "used": search_queries},
                           ensure_ascii=False, indent=2), encoding="utf-8",
            )
            print(f"   生成 {len(gen_queries)} 个 query，实际搜索 {len(search_queries)} 个"
                  f"（{'含' if args.keep_original else '不含'}原始）→ generated_queries.json")
        else:
            search_queries = queries
            print("   生成失败，回退原始 query")
    else:
        search_queries = queries

    # 2. 多渠道搜索（英文平台 youtube/x 先把 query 翻成英文）
    tr_llm, tr_model = build_eval_llm_call("gemini-flash-lite")
    overrides = await build_query_overrides(platforms, search_queries, tr_llm, tr_model)
    sources = await search_all(platforms, search_queries, args.max_count, args.max_concurrent,
                               query_overrides=overrides)

    # 3. 时间戳转可读
    try:
        from examples.process_pipeline.script.extract_sources import _convert_timestamps
        _convert_timestamps(sources)
    except Exception as e:
        logger.warning("时间戳转换跳过: %s", e)

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

    # 5. 评估（除非 --no-eval）
    eval_model_id = None
    include_images = not args.no_images
    if not args.no_eval and sources:
        eval_llm, eval_model_id = build_eval_llm_call(args.eval_model)
        print(f"🧠 评估模型: {args.eval_model} -> {eval_model_id} | 多模态图片: {'开' if include_images else '关'}")
        sources, eval_cost = await evaluate_posts(
            sources, requirement, eval_llm, eval_model_id, args.max_concurrent,
            include_images=include_images, max_images=args.max_images,
            image_mode=args.image_mode,
        )
        total_cost += eval_cost

    # 清掉评估用的临时 base64 图（别写进报告，否则文件爆炸），只留张数留痕
    for s in sources:
        imgs = s.pop("_image_data_urls", None)
        if imgs is not None:
            s["images_sent"] = len(imgs)

    # 6. 写 evaluated.json
    out_file = output_dir / "evaluated.json"
    # mod.md 中文 schema 无 decision 字段; 统计基于 制作相关性.得分
    # (>=2 视作 report 候选, ==1 视作 discard 候选, 失败/无字段不计)
    def _pr_score(s):
        ev = s.get("llm_evaluation") or {}
        if ev.get("_error"):
            return None
        try:
            return int(float((ev.get("制作相关性") or {}).get("得分")))
        except (TypeError, ValueError):
            return None
    rep = sum(1 for s in sources if (_pr_score(s) or 0) >= 2)
    dis = sum(1 for s in sources if _pr_score(s) == 1)
    out = {
        "requirement": requirement,
        "platforms": platforms,
        "queries_used": search_queries,
        "eval_model": eval_model_id,
        "total": len(sources),
        "report": rep,
        "discard": dis,
        "results": sources,
    }
    with open(out_file, "w", encoding="utf-8") as f:
        json.dump(out, f, ensure_ascii=False, indent=2)
    print(f"💾 evaluated.json: {len(sources)} 条 (report={rep} discard={dis}) → {out_file}")
    print(f"💰 累计成本: ${total_cost:.4f}")


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

    parser = argparse.ArgumentParser(description="query 词多渠道直搜 + LLM 逐条评估")
    parser.add_argument("--queries", required=True, help="query 词 json 文件（数组或 {queries:[...]}）")
    parser.add_argument("--platforms", required=True, help="逗号分隔渠道，如 xhs,zhihu,gzh,douyin,sph,youtube")
    parser.add_argument("--output-dir", required=True, help="输出目录（写 evaluated.json）")
    parser.add_argument("--max-count", type=int, default=20, help="每个 (渠道,query) 返回条数上限（部分渠道如 xhs 由后端决定）")
    parser.add_argument("--requirement", default="", help="评估用采集目标描述（缺省取 queries 文件 requirement 或 query 拼接）")
    parser.add_argument("--eval-model", default=DEFAULT_EVAL_MODEL, choices=list(EVAL_MODELS),
                        help=f"评估模型（默认 {DEFAULT_EVAL_MODEL}）")
    parser.add_argument("--max-concurrent", type=int, default=3, help="搜索 / 评估并发上限")
    parser.add_argument("--no-eval", action="store_true", help="只搜索，跳过 LLM 评估")
    parser.add_argument("--no-images", action="store_true", help="不把帖子配图发给模型（默认发，多模态评估；纯文本模型请加此项）")
    parser.add_argument("--max-images", type=int, default=4, help="每帖最多发给模型几张配图（默认 4）")
    parser.add_argument("--image-mode", choices=["url", "base64"], default="url",
                        help="图片传输方式：url 直传(快，默认) / base64 下载内嵌(稳，绕防盗链)")
    parser.add_argument("--no-transcribe", action="store_true",
                        help="不对视频帖跑转写（默认会转写并把字幕并入正文再评估）")
    # query 生成模式
    parser.add_argument("--gen-query", action="store_true", help="搜索前用 LLM 改写 / 扩展 query")
    parser.add_argument("--gen-model", default=DEFAULT_EVAL_MODEL, choices=list(EVAL_MODELS),
                        help=f"query 生成模型（默认 {DEFAULT_EVAL_MODEL}）")
    parser.add_argument("--gen-count", type=int, default=10, help="生成 query 的目标数量")
    parser.add_argument("--keep-original", action="store_true", help="生成的 query 与原始 query 合并搜索（默认只用生成的）")
    args = parser.parse_args()

    asyncio.run(run(args))


if __name__ == "__main__":
    main()