skill-video-lapian/scripts/fetch_video.py

#!/usr/bin/env python3
"""fetch_video.py — 拉片第一步: 拿到完整视频.

用法:
  # 抖音分享链接 / 视频ID → VOC 网关取详情+完整下载
  python3 fetch_video.py --url "https://v.douyin.com/xxxx/" --out-dir /tmp/lapian
  python3 fetch_video.py --aweme-id 7677548065384877346 --platform douyin --out-dir /tmp/lapian
  # 已有本地视频 → 直接登记
  python3 fetch_video.py --local /path/video.mp4 --out-dir /tmp/lapian --platform douyin

产物: <out-dir>/video.mp4 (完整) + meta.json (作者/统计/标签/时长)
铁律: 视频必须完整下载, 用 ffprobe 验证时长, 不许 Range 截断。
"""
import argparse
import json
import os
import re
import subprocess
import sys

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from common import (aweme_summary, download_file, extract_aweme_detail,  # noqa: E402
                    ffprobe_duration, get_api_token, slugify_platform, voc_call)


def resolve_share_url(url: str) -> str:
    """短链接 → 最终 URL(带重定向链)。"""
    import urllib.request
    req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
    with urllib.request.urlopen(req, timeout=30) as r:
        return r.geturl()


def extract_aweme_id(final_url: str) -> str:
    # 抖音: /video/123 或 aweme_id=123
    m = re.search(r"/(?:video|note)/(\d+)", final_url) or re.search(r"aweme_id=(\d+)", final_url)
    if m:
        return m.group(1)
    # 微信视频号: channels/weixin/v/XXX 或 finder.video.qq.com 页面里的 wxv_xxx
    m = re.search(r"wxv_(?:[A-Za-z0-9_\-]{10,})", final_url)
    if m:
        return m.group(0)
    # 兜底: 任意 20+ 位纯数字(部分分享链把视频 id 放 query)
    m = re.search(r"(\d{18,25})", final_url)
    if m:
        return m.group(1)
    raise SystemExit(f"无法从 URL 提取视频ID: {final_url}")


def detect_platform(url: str) -> str:
    """按 URL 形态自动判平台(用户只贴链接不用选)。"""
    u = url.lower()
    if "douyin.com" in u or "iesdouyin.com" in u:
        return "douyin"
    if "channels.weixin" in u or "finder.video.qq.com" in u or "weixin" in u and "wxv" in u:
        return "weixin"
    if "xiaohongshu" in u or "xhslink" in u:
        return "xiaohongshu"
    if "bilibili" in u or "b23.tv" in u:
        return "bilibili"
    return "douyin"


def fetch_meta(aweme_id: str, platform: str) -> dict:
    """VOC 网关取视频详情。douyin=fetch_one_video; weixin=视频号通道。"""
    if platform == "douyin":
        resp = voc_call("douyin/web/fetch_one_video", {"aweme_id": aweme_id})
        if resp.get("error"):
            raise SystemExit(f"[fetch] VOC fetch_one_video 失败: {resp['error']}")
        return aweme_summary(extract_aweme_detail(resp))
    if platform == "weixin":
        # 视频号: 先试专用通道,404/不支持时降级为直连下载(纯拉片不需要 meta 也能跑)
        for proxy_path, params in (
            ("weixin/channels/fetch_one_video", {"video_id": aweme_id, "aweme_id": aweme_id}),
            ("weixin/channels/fetch", {"id": aweme_id}),
        ):
            try:
                resp = voc_call(proxy_path, params)
                if resp and not resp.get("error"):
                    data = resp.get("data") or resp
                    summary = {
                        "video_id": aweme_id,
                        "platform": "weixin",
                        "title": (data.get("title") or data.get("desc") or "")[:120],
                        "author": data.get("nickname") or data.get("author") or "",
                        "play_urls": data.get("play_urls") or data.get("urls") or [],
                        "download_urls": data.get("download_urls") or data.get("video_url") or [],
                        "raw": {k: data.get(k) for k in ("duration", "cover_url") if data.get(k)},
                    }
                    print(f"[fetch] 视频号详情经 {proxy_path} 获取")
                    return summary
            except SystemExit:
                continue
            except Exception as exc:  # noqa: BLE001
                print(f"[fetch] {proxy_path} 不可用: {str(exc)[:80]}")
        print("[fetch] 视频号详情接口未命中, 降级直连模式(仅下载视频做拉片)")
        return {}
    print(f"[fetch] WARN 平台 {platform} 详情接口未接入, 仅下载/登记")
    return {}


def main():
    ap = argparse.ArgumentParser(description="下载/登记待拉片视频")
    ap.add_argument("--url", help="抖音/视频号分享短链或含视频ID的页面URL")
    ap.add_argument("--aweme-id", help="视频ID")
    ap.add_argument("--local", help="已有本地视频文件路径(跳过下载)")
    ap.add_argument("--platform", default="auto",
                    help="平台: auto(按URL自动判)/douyin/weixin/xiaohongshu/bilibili/tiktok/shortdrama")
    ap.add_argument("--out-dir", required=True, help="工作目录(产物放这里)")
    args = ap.parse_args()

    os.makedirs(args.out_dir, exist_ok=True)
    if args.platform == "auto":
        src = args.url or ""
        platform = detect_platform(src) if src else "douyin"
        print(f"[fetch] 自动识别平台: {platform}")
    else:
        platform = slugify_platform(args.platform)
    meta, video_path = {}, os.path.join(args.out_dir, "video.mp4")

    if args.local:
        video_path = os.path.abspath(args.local)
        print(f"[fetch] 使用本地视频: {video_path}")
    else:
        if not (args.url or args.aweme_id):
            ap.error("需要 --url / --aweme-id / --local 之一")
        aweme_id = args.aweme_id
        if args.url:
            final = resolve_share_url(args.url)
            aweme_id = extract_aweme_id(final)
        print(f"[fetch] 视频ID: {aweme_id} (平台: {platform})")
        meta = fetch_meta(aweme_id, platform)
        urls = meta.get("play_urls") or meta.get("download_urls")
        if not urls:
            raise SystemExit("[fetch] 拿不到播放地址(可能视频被删/私密/风控)")
        if os.path.exists(video_path) and ffprobe_duration(video_path) > 1:
            print(f"[fetch] 复用已缓存 {video_path}")
        else:
            size = download_file(urls[0], video_path)
            print(f"[fetch] 下载完成: {size/1e6:.1f} MB")
        meta["aweme_id"] = meta.get("aweme_id") or aweme_id

    dur = ffprobe_duration(video_path)
    if dur <= 1:
        raise SystemExit(f"[fetch] 视频时长异常({dur}s), 文件可能损坏")
    print(f"[fetch] 时长验证通过: {dur:.1f}s")

    meta.update({"video_path": os.path.abspath(video_path),
                 "duration_s": round(dur, 2), "platform": platform})
    meta_path = os.path.join(args.out_dir, "meta.json")
    json.dump(meta, open(meta_path, "w", encoding="utf-8"), ensure_ascii=False, indent=2)
    print(f"[fetch] meta.json 已写入: 作者={meta.get('author')} 统计={meta.get('stats')}")


if __name__ == "__main__":
    main()