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

"""真机验证：抖音视频走完整流水线（原生提取→筛选→拆分→解构→落库），用本地 mp4 绕过下载网络。

用法：python scripts/smoke_video_pipeline.py <env_file> <video.mp4> [content_id]
"""
from __future__ import annotations

import sys

import httpx

from creation_knowledge.config import Settings
from creation_knowledge.integrations.db import CkStore
from creation_knowledge.integrations.llm import default_chat
from creation_knowledge.integrations.video_extract import extract_video
from creation_knowledge.models import Post
from creation_knowledge.stages import (
    build_ingest_payload, deconstruct_item, screen_post, split_post)


def main() -> int:
    args = sys.argv[1:]
    env_file = next((a for a in args if a.endswith(".env")), ".env")
    video = next((a for a in args if a.endswith((".mp4", ".bin"))), None)
    cid = next((a for a in args if not a.endswith((".env", ".mp4", ".bin"))),
               "7612631899479648866")
    settings = Settings.from_env(env_file)
    chat = default_chat(env_file)

    det = httpx.post("http://crawler.aiddit.com/crawler/dou_yin/detail",
                     json={"content_id": cid}, timeout=40).json()["data"]["data"]
    post = Post(id=f"dy_{cid}", platform="douyin",
                url=f"https://www.douyin.com/video/{cid}", content_id=cid,
                title=det.get("title") or "", content_type="video",
                body_text=det.get("body_text") or "",
                video_urls=[det["video_url_list"][0]["video_url"]],
                raw={"data": {"data": det}})

    store = CkStore(settings.pg)
    store.delete_post(post.id)
    store.upsert_post(post)

    print("[extract] 原生整段视频 …")
    content = extract_video(post, settings=settings, video_path=video)
    store.upsert_post(post)
    store.set_extracted(post.id, content.model_dump())
    print(f"  段数={len(post.cards)}  overall={content.text[:50]}")
    print(f"  段卡示例={[(c.index, c.start, c.end) for c in post.cards[:3]]}")

    scr = screen_post(post, content, chat=chat)
    store.set_screening(post.id, scr.model_dump())
    print(f"[screen] passed={scr.passed} score={scr.score}")
    if scr.passed:
        items = split_post(post, content, chat=chat)
        print(f"[split] {len(items)} 条知识")
        for it in items:
            deco = deconstruct_item(it, chat=chat)
            payload = build_ingest_payload(post, it, deco)
            store.save_item(post.id, it.model_dump(), deco.model_dump(), payload.model_dump())
        store.update_stage(post.id, "done")

    p = store.read_post(post.id)
    its = store.read_items(post.id)
    print(f"[DB] stage={p['stage']} cards={len(p.get('cards') or [])} "
          f"kinds={set(c['kind'] for c in (p.get('cards') or []))} items={len(its)}")
    if its:
        print(f"  item0: source_cards={its[0]['item'].get('source_cards')} "
              f"evidence0={(its[0]['item'].get('evidence') or [None])[0]}")
    return 0


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