workbuddy_skills/video-capture/scripts/extract_frames.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
video-capture · 抽帧（双因素 + 首尾 + 补帧）

铁律来源：aigc-idea-impression 的《视频抽帧与字幕规范》（2026-08-30 实测沉淀）。
算法：① 画面 scene 0.3 ② 字幕区 crop 后 scene 0.1（漏这条 = 画面静止但字幕换句的帧全丢）
      ③ 低阈值 scene 0.05 ④ 首帧/尾帧 ⑤ 相邻帧间隔 >5s 补中点；逐点 -ss 抽帧（禁 eq(t)）。

依赖：ffmpeg / ffprobe（本机已就绪）。仅 Python 标准库。
用法：
  python extract_frames.py --video a.mp4 --out frames/ [--max-gap 5] [--crop "720:240:0:720"] [--thresholds 0.3,0.1,0.05]
"""
import argparse
import json
import os
import re
import subprocess
import sys


def run(cmd):
    return subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8", errors="replace")


def ffprobe_meta(video):
    r = run([
        "ffprobe", "-v", "error",
        "-show_entries", "format=duration",
        "-show_entries", "stream=codec_type,width,height",
        "-of", "json", video,
    ])
    info = {"duration": 0.0, "width": 0, "height": 0}
    try:
        j = json.loads(r.stdout or "{}")
        info["duration"] = float((j.get("format") or {}).get("duration") or 0.0)
        for s in (j.get("streams") or []):
            if s.get("codec_type") == "video":
                info["width"] = int(s.get("width") or 0)
                info["height"] = int(s.get("height") or 0)
                break
    except Exception:
        pass
    return info


FPS_MODE_ARGS = ("-fps_mode", "vfr")     # ffmpeg ≥ 5.1 的新写法（9.x 已移除 -vsync）
LEGACY_FPS_ARGS = ("-vsync", "vfr")      # 老 ffmpeg 的写法，兜底用


def _scene_cmd(video, vf, fps_args):
    return ["ffmpeg", "-hide_banner", "-i", video, "-vf", vf, *fps_args, "-f", "null", "-"]


def scene_times(video, vf, label):
    """跑一次 scene 检测，返回时间点列表。

    🔴 2026-10-11 修（实测踩到）：本机 ffmpeg 9.0.1 **已移除 `-vsync`**
      ⇒ 原来的 `-vsync vfr` 写法让三路 scene 检测**全部失败**、整支视频只出 3 帧
      （当时执行会话只能另存一份「逐字同算法的本地副本」才跑通 ⇒ 两套口径，正是要避免的）。
      ⇒ 现在优先用 `-fps_mode vfr`，没跑出 `pts_time` 再退回 `-vsync vfr`，新旧 ffmpeg 都能跑。
    """
    r = run(_scene_cmd(video, vf, FPS_MODE_ARGS))
    out = (r.stderr or "") + (r.stdout or "")
    if "pts_time:" not in out:               # 新写法没跑出来 ⇒ 退回老写法
        r2 = run(_scene_cmd(video, vf, LEGACY_FPS_ARGS))
        out2 = (r2.stderr or "") + (r2.stdout or "")
        if "pts_time:" in out2:
            out = out2
    times = [float(t) for t in re.findall(r"pts_time:([0-9.]+)", out)]
    times = sorted(set(t for t in times if t >= 0))
    return times


def default_subtitle_crop(w, h):
    """字幕区默认取底部 15%。"""
    if w <= 0 or h <= 0:
        return None
    ch = max(1, int(h * 0.15))
    y = h - ch
    return f"crop={w}:{ch}:0:{y}"


def extract(video, out_dir, max_gap, crop, thresholds):
    meta = ffprobe_meta(video)
    dur = meta["duration"]
    if dur <= 0:
        print(json.dumps({"ok": False, "error": "无法读取视频时长（文件损坏或非视频）"}), file=sys.stderr)
        sys.exit(2)

    t_pic, t_sub, t_low = thresholds
    times = set()

    # ① 画面变化
    times.update(scene_times(video, f"select='gt(scene,{t_pic})',showinfo", "pic"))
    # ② 字幕区变化（crop 后检测）—— 漏这条会丢"画面静止但字幕换句"
    sub_crop = crop or default_subtitle_crop(meta["width"], meta["height"])
    if sub_crop:
        times.update(scene_times(video, f"{sub_crop},select='gt(scene,{t_sub})',showinfo", "sub"))
    # ③ 低阈值补漏
    times.update(scene_times(video, f"select='gt(scene,{t_low})',showinfo", "low"))
    # ④ 首尾帧
    times.add(0.0)
    times.add(max(0.0, dur - 0.3))

    ordered = sorted(times)
    # ⑤ 长间隔补中点
    with_gap = []
    prev = None
    for t in ordered:
        if prev is not None and (t - prev) > max_gap:
            with_gap.append((t + prev) / 2.0)
        with_gap.append(t)
        prev = t
    ordered = sorted(set(round(t, 3) for t in (ordered + with_gap)))

    os.makedirs(out_dir, exist_ok=True)
    frames = []
    for i, t in enumerate(ordered, 1):
        t = min(t, max(0.0, dur - 0.05))
        name = f"c_{i:04d}_{t}s.jpg"
        path = os.path.join(out_dir, name)
        run([
            "ffmpeg", "-hide_banner", "-loglevel", "error", "-y",
            "-ss", f"{t}", "-i", video, "-frames:v", "1", "-q:v", "2", path,
        ])
        frames.append({"index": i, "time": t, "file": name})

    manifest = {
        "video": os.path.abspath(video),
        "duration": round(dur, 3),
        "resolution": f'{meta["width"]}x{meta["height"]}',
        "subtitleCrop": sub_crop,
        "thresholds": {"picture": t_pic, "subtitle": t_sub, "low": t_low},
        "maxGap": max_gap,
        "frameCount": len(frames),
        "frames": frames,
    }
    with open(os.path.join(out_dir, "frame_times.json"), "w", encoding="utf-8", newline="\n") as f:
        json.dump(manifest, f, ensure_ascii=False, indent=2)
    return manifest


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--video", required=True)
    ap.add_argument("--out", default="frames")
    ap.add_argument("--max-gap", type=float, default=5.0)
    ap.add_argument("--crop", default="", help='字幕区裁剪，如 "720:240:0:720"；缺省按分辨率底部 15%')
    ap.add_argument("--thresholds", default="0.3,0.1,0.05", help="画面,字幕,低阈值")
    args = ap.parse_args()

    if not os.path.exists(args.video):
        print(json.dumps({"ok": False, "error": f"找不到视频：{args.video}"}), file=sys.stderr)
        sys.exit(2)
    try:
        t = [float(x) for x in args.thresholds.split(",")]
        assert len(t) == 3
    except Exception:
        print("thresholds 需为 3 个数值，如 0.3,0.1,0.05", file=sys.stderr)
        sys.exit(2)

    m = extract(args.video, args.out, args.max_gap, args.crop.strip(), t)
    print(json.dumps({
        "ok": True, "out": os.path.abspath(args.out),
        "frameCount": m["frameCount"], "duration": m["duration"],
        "resolution": m["resolution"], "subtitleCrop": m["subtitleCrop"],
    }, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()