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

"""实验B：本地 mp4 → base64 → 经 OpenRouter 把整段视频发给 Gemini，按时间戳提炼。

我们没有直连 GEMINI_API_KEY，Gemini 走 OpenRouter。OpenRouter 对 Gemini 视频支持
base64 data URL（Vertex 通道）。本脚本验证这条路能否拿到详细的时间戳分段提炼。

用法：python scripts/smoke_video_openrouter.py <env_file> <video.mp4>
"""
from __future__ import annotations

import base64
import sys

import httpx

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():
    env_file = next((a for a in sys.argv[1:] if a.endswith(".env")), ".env")
    video = next((a for a in sys.argv[1:] if a.endswith((".mp4", ".bin"))), None)
    env = load_env(env_file)
    key = env.get("OPENROUTER_API_KEY")
    base = (env.get("OPENROUTER_BASE_URL") or "https://openrouter.ai/api/v1").rstrip("/")
    model = env.get("OPENROUTER_MODEL") or "google/gemini-3-flash-preview"
    if not key:
        raise SystemExit("缺 OPENROUTER_API_KEY")

    raw = open(video, "rb").read()
    print(f"video {video}: {len(raw)} bytes ({len(raw)//1024//1024} MB) -> base64")
    data_url = "data:video/mp4;base64," + base64.b64encode(raw).decode()

    body = {"model": model, "messages": [{"role": "user", "content": [
        {"type": "text", "text": PROMPT},
        {"type": "video_url", "video_url": {"url": data_url}},
    ]}]}
    print(f"POST {base}/chat/completions  model={model}  payload≈{len(data_url)//1024//1024}MB")
    r = httpx.post(f"{base}/chat/completions",
                   headers={"Authorization": f"Bearer {key}", "Content-Type": "application/json"},
                   json=body, timeout=600)
    print("HTTP", r.status_code)
    if r.status_code != 200:
        print(r.text[:800]); raise SystemExit(1)
    out = r.json()
    print("provider:", out.get("provider"), "| usage:", out.get("usage"))
    print("==== 提炼结果 ====")
    print(out["choices"][0]["message"]["content"])


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