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

"""平台数据可下载性 + Gemini 直读 URL 能力 一次测全。

各平台 detail → 图片/视频字段 → 实下载探测；最后测：远程视频 URL 直接喂 Gemini（不 base64）。
用法：python scripts/probe_platforms.py <env_file>
"""
from __future__ import annotations

import sys
import time

import httpx

CRAW = "http://crawler.aiddit.com"
IOS = "Mozilla/5.0 (iPhone; CPU iPhone OS 16_0 like Mac OS X) AppleWebKit/605.1.15"
REFERER = {"douyin": "https://www.douyin.com/", "kuaishou": "https://www.kuaishou.com/",
           "bilibili": "https://www.bilibili.com/", "xiaohongshu": "https://www.xiaohongshu.com/"}

PROBES = [
    ("小红书·图文", "/crawler/xiao_hong_shu/detail", "67e4bdf50000000006028a59", "xiaohongshu"),
    ("小红书·视频", "/crawler/xiao_hong_shu/detail", "6976f6fd000000001a01eede", "xiaohongshu"),
    ("抖音·视频", "/crawler/dou_yin/detail", "7612631899479648866", "douyin"),
    ("B站·视频", "/crawler/bilibili/detail", "BV1qd4y1x76w", "bilibili"),
    ("快手·视频", "/crawler/kuai_shou/detail", "3xepqgwddgbickc", "kuaishou"),
]


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


def detail(path, cid):
    return httpx.post(CRAW + path, json={"content_id": cid}, timeout=40).json()


def inner(r):
    d = ((r.get("data") or {}).get("data")) or {}
    if isinstance(d, list):
        return d[0] if d else {}
    return d if isinstance(d, dict) else {}


def vurl_of(d):
    vl = d.get("video_url_list")
    if isinstance(vl, list) and vl:
        return vl[0].get("video_url") if isinstance(vl[0], dict) else vl[0]
    return None


def main():
    env_file = sys.argv[1] if len(sys.argv) > 1 and sys.argv[1].endswith(".env") else ".env"
    env = load_env(env_file)
    or_key = env.get("OPENROUTER_API_KEY")
    or_base = (env.get("OPENROUTER_BASE_URL") or "https://openrouter.ai/api/v1").rstrip("/")
    model = env.get("OPENROUTER_MODEL") or "google/gemini-3-flash-preview"

    for name, path, cid, plat in PROBES:
        try:
            r = detail(path, cid)
            d = inner(r)
            il = d.get("image_url_list") or []
            mb = d.get("multi_bitrate")
            vurl = vurl_of(d)
            print(f"\n[{name}] code={r.get('code')} type={d.get('content_type')} "
                  f"title={(d.get('title') or '')[:26]}")
            print(f"  图片={len(il)}  视频链接={'有' if vurl else '无'}  "
                  f"多码率={list(mb.keys()) if isinstance(mb, dict) else mb}")
            if vurl:
                try:
                    with httpx.stream("GET", vurl, headers={"User-Agent": IOS,
                                      "Referer": REFERER[plat], "Range": "bytes=0-524287"},
                                      timeout=60, follow_redirects=True) as resp:
                        chunk = next(resp.iter_bytes(), b"")
                        print(f"  ↓视频: http={resp.status_code} "
                              f"type={resp.headers.get('content-type')} magic={chunk[4:12]!r}")
                except Exception as e:
                    print("  ↓视频失败:", str(e)[:80])
            elif il:
                u = il[0] if isinstance(il[0], str) else il[0].get("image_url")
                try:
                    resp = httpx.get(u, timeout=30, follow_redirects=True)
                    print(f"  ↓图片: http={resp.status_code} "
                          f"type={resp.headers.get('content-type')} {len(resp.content)}B")
                except Exception as e:
                    print("  ↓图片失败:", str(e)[:80])
            time.sleep(16)
        except Exception as e:
            print(f"[{name}] 失败: {e}")

    print("\n=== 关键测试：远程视频 URL 直接喂 Gemini（不下载、不 base64）===")
    d = inner(detail("/crawler/dou_yin/detail", "7612631899479648866"))
    vurl = vurl_of(d)
    body = {"model": model, "messages": [{"role": "user", "content": [
        {"type": "text", "text": "这视频在讲什么？一句话。"},
        {"type": "video_url", "video_url": {"url": vurl}}]}]}
    try:
        g = httpx.post(f"{or_base}/chat/completions",
                       headers={"Authorization": f"Bearer {or_key}", "Content-Type": "application/json"},
                       json=body, timeout=180)
        print("  抖音远程URL直喂: HTTP", g.status_code)
        if g.status_code == 200:
            j = g.json()
            print("  video_tokens=", j.get("usage", {}).get("video_tokens"),
                  "→ 远程 URL 可直接处理！")
            print("  回答:", j["choices"][0]["message"]["content"][:120])
        else:
            print("  失败:", g.text[:240], "→ 远程 URL 不行，必须下载+base64")
    except Exception as e:
        print("  异常:", str(e)[:150])
    print("\n===PROBEDONE===")
    return 0


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