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

"""B站取数验证：DASH 分轨（视频 m4s + 音频 m4s）→ 下载 → ffmpeg 合并 → 验证有视频+音频。"""
from __future__ import annotations

import os
import subprocess
import sys

import httpx
import imageio_ffmpeg

CRAW = "http://crawler.aiddit.com"


def dl(url, path):
    # 用 curl，不跟随重定向（避免跟到 B站 mcdn 的 P2P 死节点，httpx 会卡在那）
    subprocess.run(["curl", "-s", "--connect-timeout", "15", "-m", "180",
                    "-A", "Mozilla/5.0", "-e", "https://www.bilibili.com/",
                    "-o", path, url], check=True)
    return os.path.getsize(path)


def main():
    cid = sys.argv[1] if len(sys.argv) > 1 else "BV1qd4y1x76w"
    ff = imageio_ffmpeg.get_ffmpeg_exe()
    d = ((httpx.post(f"{CRAW}/crawler/bilibili/detail", json={"content_id": cid},
                     timeout=40).json().get("data") or {}).get("data")) or {}
    v = d["video_url_list"][0]["video_url"]
    a = (d.get("voice_data") or {}).get("play_url")
    print("title:", (d.get("title") or "")[:30])
    print("视频轨:", v.split("?")[0].split("/")[-1])
    print("音频轨:", a.split("?")[0].split("/")[-1] if a else "无 voice_data.play_url")
    if not a:
        print("→ 没有音频轨，B站这条只有视频"); return 0

    nv = dl(v, "/tmp/bili_v.m4s")
    na = dl(a, "/tmp/bili_a.m4s")
    print(f"下载: 视频 {nv//1024//1024}MB, 音频 {na//1024}KB")

    r = subprocess.run([ff, "-y", "-i", "/tmp/bili_v.m4s", "-i", "/tmp/bili_a.m4s",
                        "-c", "copy", "/tmp/bili_merged.mp4"], capture_output=True, text=True)
    import os
    if not os.path.exists("/tmp/bili_merged.mp4"):
        print("合并失败:", r.stderr[-300:]); return 1
    sz = os.path.getsize("/tmp/bili_merged.mp4")
    err = subprocess.run([ff, "-hide_banner", "-i", "/tmp/bili_merged.mp4"],
                         capture_output=True, text=True).stderr
    streams = [l.strip() for l in err.splitlines() if "Stream #" in l]
    print(f"合并后 mp4: {sz//1024//1024}MB")
    print("流：")
    for s in streams:
        print("  ", s[:90])
    has_v = any("Video" in s for s in streams)
    has_a = any("Audio" in s for s in streams)
    print(f"\n结论：视频={'✅' if has_v else '❌'}  音频={'✅' if has_a else '❌'}  "
          f"→ {'B站可拿到视频+音频，合并即可原生提炼' if has_v and has_a else 'B站缺轨'}")
    print("===BILIDONE===")
    return 0


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