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

"""一次性实验：抖音视频 → 直连 Gemini 原生视频 → 按时间戳分段提炼创作知识。

验证"整段视频交给 Gemini 能否详细提炼"，对比抽帧。不接流水线，仅评估。
直连 Gemini File API（GEMINI_API_KEY），不走 OpenRouter（其对 Gemini 视频仅支持 YouTube）。

用法：python scripts/smoke_video_native.py [env_file] [content_id]
"""
from __future__ import annotations

import json
import os
import sys
import time

import httpx

BASE = "https://generativelanguage.googleapis.com"
DETAIL = "http://crawler.aiddit.com/crawler/dou_yin/detail"
IOS_UA = "Mozilla/5.0 (iPhone; CPU iPhone OS 16_0 like Mac OS X) AppleWebKit/605.1.15"

PROMPT = (
    "你在从一条短视频里提炼「创作知识」——能指导别人怎么做内容的方法/原理/清单。\n"
    "视频是口播+画面，请**同时听口播、看画面与字幕**。按视频时间顺序分段，"
    "每段给时间戳和这段讲到的创作知识。原文（含口播）没有的不要编造。只输出 JSON：\n"
    '{"video_title": "", "overall": "全片在教什么，一两句话",\n'
    ' "segments": [{"start": "MM:SS", "end": "MM:SS", "title": "这段小标题",\n'
    '   "knowledge_types": ["what/why/how"], "what": "或null", "why": "或null", "how": "或null"}]}'
)


def load_env(path):
    env = {}
    for line in open(path):
        line = line.strip()
        if line and not line.startswith("#") and "=" in line:
            k, v = line.split("=", 1)
            env[k.strip()] = v.strip().strip('"').strip("'")
    return env


def main():
    args = sys.argv[1:]
    env_file = next((a for a in args if a.endswith(".env")), ".env")
    video_file = 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")
    env = load_env(env_file)
    key = env.get("GEMINI_API_KEY") or os.environ.get("GEMINI_API_KEY")
    if not key:
        raise SystemExit("缺 GEMINI_API_KEY（env_file 或环境变量）")
    model = (env.get("OPENROUTER_MODEL") or "gemini-3-flash-preview").split("/")[-1]

    if video_file:  # 已有本地 mp4（如别处下载后传来），跳过抓取
        data = open(video_file, "rb").read()
        print(f"[1-2] 使用本地视频 {video_file}：{len(data)} bytes ({len(data)//1024//1024} MB)")
    else:
        print(f"[1] 取新鲜视频直链 content_id={cid}")
        inner = httpx.post(DETAIL, json={"content_id": cid}, timeout=40).json()["data"]["data"]
        vurl = inner["video_url_list"][0]["video_url"]
        print("    title:", inner.get("title"), "| dur:",
              inner["video_url_list"][0].get("video_duration"), "s")
        print("[2] 下载 mp4")
        with httpx.stream("GET", vurl, headers={"User-Agent": IOS_UA, "Referer": "https://www.douyin.com/"},
                          timeout=180, follow_redirects=True) as r:
            r.raise_for_status()
            data = b"".join(r.iter_bytes())
        print(f"    {len(data)} bytes ({len(data)//1024//1024} MB)")

    print("[3] File API 上传")
    start = httpx.post(
        f"{BASE}/upload/v1beta/files?key={key}",
        headers={"X-Goog-Upload-Protocol": "resumable", "X-Goog-Upload-Command": "start",
                 "X-Goog-Upload-Header-Content-Length": str(len(data)),
                 "X-Goog-Upload-Header-Content-Type": "video/mp4",
                 "Content-Type": "application/json"},
        json={"file": {"display_name": "douyin_" + cid}}, timeout=60)
    start.raise_for_status()
    up_url = start.headers["X-Goog-Upload-URL"]
    up = httpx.post(up_url, headers={"X-Goog-Upload-Offset": "0",
                                     "X-Goog-Upload-Command": "upload, finalize"},
                    content=data, timeout=600)
    up.raise_for_status()
    fobj = up.json()["file"]
    name, uri = fobj["name"], fobj["uri"]
    print("    file:", name)

    print("[4] 等待处理 ACTIVE")
    for _ in range(80):
        f = httpx.get(f"{BASE}/v1beta/{name}?key={key}", timeout=30).json()
        st = f.get("state")
        if st == "ACTIVE":
            break
        if st == "FAILED":
            raise SystemExit("文件处理失败: " + json.dumps(f, ensure_ascii=False))
        time.sleep(3)
    print("    state:", st)

    print(f"[5] generateContent（model={model}，按时间戳提炼）")
    body = {"contents": [{"parts": [{"file_data": {"mime_type": "video/mp4", "file_uri": uri}},
                                    {"text": PROMPT}]}]}
    for m in [model, "gemini-2.5-flash", "gemini-flash-latest"]:
        g = httpx.post(f"{BASE}/v1beta/models/{m}:generateContent?key={key}", json=body, timeout=300)
        if g.status_code == 200:
            model = m
            break
        print(f"    model {m} -> HTTP {g.status_code}: {g.text[:160]}")
    g.raise_for_status()
    text = g.json()["candidates"][0]["content"]["parts"][0]["text"]
    print("    使用模型:", model)
    print("==== 提炼结果 ====")
    print(text)


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