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

"""M4 真实验收（省 Gemini 费）：用 海狸 帖的富文本 body_text 当提取内容，
真调 claude-sonnet-4.5 跑 screen/split/deconstruct/assemble，打印结果。

只产生文本 LLM 费用，不调 Gemini。须在能联网调 OpenRouter 的云端跑。
用法：python scripts/smoke_stages.py [env_file] [content_id]
"""
from __future__ import annotations

import json
import sys
from pathlib import Path

from creation_knowledge.integrations.crawler import parse_detail_response
from creation_knowledge.integrations.llm import default_chat
from creation_knowledge.models import ExtractedContent
from creation_knowledge.stages import (
    build_ingest_payload,
    deconstruct_item,
    screen_post,
    split_post,
)

FIXTURES = Path(__file__).resolve().parent.parent / "tests" / "fixtures"


def main() -> int:
    args = sys.argv[1:]
    env_file = args[0] if args and args[0].endswith(".env") else ".env"
    cid = next((a for a in args if not a.endswith(".env")), "67e4bdf50000000006028a59")

    resp = json.loads((FIXTURES / f"xhs_case_{cid}.json").read_text("utf-8"))
    post = parse_detail_response(resp, fallback_content_id=cid)
    # 用富文本 body_text 当已提取内容，避免再调 Gemini
    content = ExtractedContent(text=post.body_text, is_empty=not post.body_text.strip())
    chat = default_chat(env_file)

    print(f"==== {post.id}  {post.title} ====\n")
    scr = screen_post(post, content, chat=chat)
    print(f"[screen] passed={scr.passed} score={scr.score} reason={scr.reason}")
    if not scr.passed:
        return 0

    items = split_post(post, content, chat=chat)
    print(f"[split] {len(items)} 个知识片段")
    for i, item in enumerate(items, 1):
        print(f"\n  片段{i}: {item.title}  types={item.knowledge_types}")
        deco = deconstruct_item(item, chat=chat)
        print(f"    stages={deco.stages}  scopes={[(s.scope_type, s.value) for s in deco.scopes]}")
        payload = build_ingest_payload(post, item, deco)
        print(f"    ingest.content={payload.content[:120]}")
    return 0


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