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:
    m = re.search(r"/(?:video|note)/(\d+)", final_url) or re.search(r"aweme_id=(\d+)", final_url)
    if not m:
        raise SystemExit(f"无法从 URL 提取视频ID: {final_url}")
    return m.group(1)


def fetch_meta(aweme_id: str, platform: str) -> dict:
    """VOC 网关取视频详情(抖音)。其他平台留 TODO。"""
    if platform != "douyin":
        print(f"[fetch] WARN 平台 {platform} 详情接口未接入, 仅下载/登记")
        return {}
    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))


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="douyin",
                    help="平台英文简称: douyin/xiaohongshu/weixin/bilibili/tiktok/shortdrama")
    ap.add_argument("--out-dir", required=True, help="工作目录(产物放这里)")
    args = ap.parse_args()

    os.makedirs(args.out_dir, exist_ok=True)
    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()