video_utils.py

import os, shutil, subprocess

def has_ffmpeg() -> bool:
    return shutil.which("ffmpeg") is not None and shutil.which("ffprobe") is not None

def probe_duration(video_path: str) -> float | None:
    if not shutil.which("ffprobe"):
        return None
    try:
        out = subprocess.check_output(
            ["ffprobe","-v","error","-select_streams","v:0","-show_entries","stream=duration","-of","default=nw=1:nk=1",video_path],
            stderr=subprocess.STDOUT, text=True
        )
        return float(out.strip())
    except Exception:
        return None

def extract_frames_ffmpeg(video_path: str, fps: float, out_dir: str) -> list[tuple[str, int]]:
    os.makedirs(out_dir, exist_ok=True)
    tpl = os.path.join(out_dir, "frame_%06d.jpg")
    subprocess.check_call(
        ["ffmpeg","-y","-i",video_path,"-vf",f"fps={fps}","-qscale:v","2",tpl],
        stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL
    )
    frames = sorted([os.path.join(out_dir,f) for f in os.listdir(out_dir) if f.lower().endswith(".jpg")])
    return [(p, i) for i, p in enumerate(frames)]

def runs_from_indices(idxs: list[int]) -> list[tuple[int,int]]:
    if not idxs: return []
    idxs = sorted(idxs)
    runs, s, prev = [], idxs[0], idxs[0]
    for x in idxs[1:]:
        if x == prev + 1:
            prev = x
        else:
            runs.append((s, prev)); s = prev = x
    runs.append((s, prev))
    return runs

def merge_seconds_union(all_indices: list[int], fps: float, pad: float = 0.25) -> list[tuple[float,float]]:
    if not all_indices: return []
    runs = runs_from_indices(sorted(all_indices))
    intervals = []
    for a, b in runs:
        start = max(0.0, a / fps - pad)
        end   = (b + 1) / fps + pad
        intervals.append((start, end))
    merged = []
    for s, e in sorted(intervals):
        if not merged or s > merged[-1][1]:
            merged.append((s, e))
        else:
            merged[-1] = (merged[-1][0], max(merged[-1][1], e))
    return merged

def redact_with_ffmpeg(video_path: str, intervals: list[tuple[float,float]], out_path: str):
    if not intervals:
        shutil.copyfile(video_path, out_path); return
    parts = [f"between(t\\,{s:.3f}\\,{e:.3f})" for s, e in intervals]
    expr = f"not({' + '.join(parts)})"
    vf = f"select='{expr}',setpts=N/FRAME_RATE/TB"
    subprocess.check_call(["ffmpeg","-y","-i",video_path,"-vf",vf,"-an",out_path],
                          stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)