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

"""M3 真实验收：对样例帖子调真实 Gemini，打印多模态提取结果。

须在能联网调 OpenRouter 的云端跑（会产生 API 费用）。
用 fixture 解析出 Post（含 image_urls），避免重复爬虫调用。

用法：
    python scripts/smoke_extract.py [env_file] [content_id ...]
默认跑「拾意」两条空正文帖（验收点：from_image 非空、is_empty=false）。
"""
from __future__ import annotations

import json
import sys
from pathlib import Path

from acquisition.crawler import parse_detail_response
from creation_knowledge.integrations.extractor import GeminiExtractor

FIXTURES = Path(__file__).resolve().parent.parent / "tests" / "fixtures"
DEFAULT_IDS = ["67e2e39b0000000003028ff0", "680659e8000000001a007a11"]


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

    client = GeminiExtractor.from_env(env_file=env_file)
    print(f"[model] {client.model}  base={client.base_url}\n")

    for cid in ids:
        resp = json.loads((FIXTURES / f"xhs_case_{cid}.json").read_text("utf-8"))
        post = parse_detail_response(resp, fallback_content_id=cid)
        print(f"==== {post.id}  作者={post.author_name}  图片={len(post.image_urls)} ====")
        print(f"标题: {post.title}")
        print(f"body_text(原文): {post.body_text[:50]!r}")
        out = client.extract(post)
        print(f"  is_empty   : {out.is_empty}")
        print(f"  from_image : {out.from_image[:200]}")
        print(f"  text       : {out.text[:200]}")
        print()
    return 0


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