Create-knowledge-find-decode/scripts/run_batch.py

"""M5 真实 e2e：把 5 个样例帖跑完整流水线，落 DB（INGEST_ENABLED=False），再回读校验。

须在能连 DB + OpenRouter 的云端跑（会有 crawler/Gemini/claude 调用费）。
用法：python scripts/run_batch.py [env_file] [content_id ...]
"""
from __future__ import annotations

import json
import sys

from creation_knowledge.config import PgConfig
from creation_knowledge.integrations.db import CkStore
from creation_knowledge.pipeline import run_pipeline

DEFAULT_IDS = [
    "67e4bdf50000000006028a59",  # HOW+WHAT 海狸
    "698481e1000000000a02a7c1",  # HOW+WHAT Irvin
    "67e2e39b0000000003028ff0",  # HOW 拾意
    "699308fa0000000016009697",  # WHY 方圆
    "680659e8000000001a007a11",  # HOW+WHY+WHAT 拾意
]


def main() -> int:
    args = sys.argv[1:]
    env_file = args[0] if args and args[0].endswith(".env") else ".env"
    ids = [a for a in args if not a.endswith(".env")] or DEFAULT_IDS

    print(f"=== 跑流水线：{len(ids)} 个样例（ingest 关闭）===")
    results = run_pipeline(ids, env_file=env_file)
    for r in results:
        print(json.dumps(r, ensure_ascii=False))

    print("\n=== DB 回读校验 ===")
    store = CkStore(PgConfig.from_env(env_file))
    for r in results:
        pid = r.get("post_id")
        if not pid:
            print(f"{r['url']}  -> {r['status']}")
            continue
        post = store.read_post(pid)
        items = store.read_items(pid)
        statuses = [it["ingest_status"] for it in items]
        print(f"{pid}  stage={post['stage']}  items={len(items)}  ingest_status={statuses}")
    return 0


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