Screenshots via yt-dlp clip download over HTTPS proxy (ffmpeg local); trust proxy CA for yt-dlp

pipeline/frames.py

@@ -1,9 +1,13 @@
-"""Stage 5 + 6: real screenshots via yt-dlp stream URL + ffmpeg, at weighted timestamps.
+"""Stage 5 + 6: real screenshots via short yt-dlp clip downloads + ffmpeg, at weighted
+timestamps.
 
-No full download. We resolve a direct video stream URL with yt-dlp (metadata only), then
-``ffmpeg -ss T -frames:v 1`` grabs a single frame per timestamp over HTTP range requests
-(only a few seconds of data each). For each shot we grab 3 candidates around T and keep
-the sharpest (Laplacian variance) to avoid landing on a blurry/transition frame.
+For each timestamp we download a ~1.5 s clip with yt-dlp's ``download_ranges`` (a couple
+hundred KB) and let ``ffmpeg`` extract frames from that *local* file. We grab 3 candidates
+around the target and keep the sharpest (Laplacian variance) to avoid a blurry/transition
+frame. Downloading (rather than ffmpeg-seeking the remote stream) is required because the
+stream lives on ``googlevideo.com`` and must go through the residential proxy — and when
+``YT_PROXY`` is an HTTPS/TLS-wrapped proxy (needed to hide the target host from egress DPI)
+ffmpeg can't talk to it, but yt-dlp can.
 
 The capture timestamp itself comes from the weighted indicator: a blend of the LLM's
 suggested timestamp and the transcript segment timing of the text the step quotes.
@@ -15,10 +19,7 @@ import os
 import subprocess
 
 from yt_dlp import YoutubeDL
-
-# Format preference: muxed (single A/V url) <=720p first, then video-only, then anything.
-_FORMAT = ("best[height<=720][vcodec!=none][acodec!=none]/"
-           "best[height<=720][vcodec!=none]/bestvideo[height<=720]/best[vcodec!=none]/best")
+from yt_dlp.utils import download_range_func
 
 
 # --------------------------------------------------------------------------- weighting
@@ -69,48 +70,7 @@ def compute_shot_times(steps: list[dict], segs: list[dict], *, w_llm: float = 0.
     }
 
 
-# ----------------------------------------------------------------------- stream + grab
-def get_stream_url(video_id: str, cookiefile: str | None = None,
-                   proxy: str | None = None) -> tuple[str, float]:
-    """Resolve a direct video stream URL (no download); return ``(url, duration_seconds)``."""
-    opts = {"quiet": True, "no_warnings": True, "skip_download": True, "format": _FORMAT}
-    if cookiefile:
-        opts["cookiefile"] = cookiefile
-    if proxy:
-        opts["proxy"] = proxy
-    with YoutubeDL(opts) as ydl:
-        info = ydl.extract_info(f"https://www.youtube.com/watch?v={video_id}", download=False)
-
-    stream = info.get("url")
-    if not stream:
-        for f in info.get("requested_formats") or []:
-            if f.get("vcodec") not in (None, "none") and f.get("url"):
-                stream = f["url"]
-                break
-    if not stream:
-        vids = [f for f in info.get("formats", [])
-                if f.get("vcodec") not in (None, "none") and f.get("url")]
-        if vids:
-            stream = sorted(vids, key=lambda f: f.get("height") or 0)[-1]["url"]
-    if not stream:
-        raise RuntimeError("Could not resolve a video stream URL for frame capture.")
-    return stream, float(info.get("duration") or 0.0)
-
-
-def _grab(stream_url: str, t: float, out_path: str, proxy: str | None) -> bool:
-    """Grab one frame at time ``t`` via ffmpeg input-seek; return True on success."""
-    cmd = ["ffmpeg", "-y", "-loglevel", "error", "-ss", f"{max(0.0, t):.2f}",
-           "-i", stream_url, "-frames:v", "1", "-q:v", "2", out_path]
-    env = dict(os.environ)
-    if proxy:  # stream URL is bound to the extraction IP; route ffmpeg through it too
-        env["http_proxy"] = env["https_proxy"] = proxy
-    try:
-        subprocess.run(cmd, capture_output=True, text=True, timeout=90, env=env)
-    except subprocess.TimeoutExpired:
-        return False
-    return os.path.exists(out_path) and os.path.getsize(out_path) > 0
-
-
+# -------------------------------------------------------------------- download + grab
 def _sharpness(path: str) -> float:
     """Variance of the Laplacian (higher = sharper). Used to reject blurry frames."""
     import numpy as np
@@ -124,13 +84,20 @@ def _sharpness(path: str) -> float:
     return float(lap.var())
 
 
-def grab_best_frame(stream_url: str, t: float, out_dir: str, idx: int,
-                    proxy: str | None = None) -> str | None:
-    """Grab candidates around ``t`` and keep the sharpest; return its path or None."""
+def _best_frame_from_clip(clip: str, out_dir: str, idx: int, target: float) -> str | None:
+    """Extract candidates near ``target`` seconds into a *local* ``clip`` and keep the
+    sharpest. ffmpeg reads a local file here, so no proxy/network is involved."""
     cands = []
-    for k, dt in enumerate((-0.5, 0.0, 0.5)):
+    for k, dt in enumerate((-0.4, 0.0, 0.4)):
         p = os.path.join(out_dir, f"shot_{idx}_{k}.jpg")
-        if _grab(stream_url, t + dt, p, proxy):
+        ss = max(0.0, target + dt)
+        cmd = ["ffmpeg", "-y", "-loglevel", "error", "-ss", f"{ss:.2f}",
+               "-i", clip, "-frames:v", "1", "-q:v", "2", p]
+        try:
+            subprocess.run(cmd, capture_output=True, text=True, timeout=60)
+        except subprocess.TimeoutExpired:
+            continue
+        if os.path.exists(p) and os.path.getsize(p) > 0:
             cands.append(p)
     if not cands:
         return None
@@ -144,16 +111,53 @@ def grab_best_frame(stream_url: str, t: float, out_dir: str, idx: int,
     return best
 
 
-def capture_shots(times: dict[int, float], stream_url: str, out_dir: str,
-                  proxy: str | None = None, progress=None) -> dict[int, dict]:
-    """Grab a frame for each ``{step_index: time}``; return ``{idx: {"time", "path"}}``."""
+def capture_shots(times: dict[int, float], video_id: str, out_dir: str,
+                  cookiefile: str | None = None, proxy: str | None = None,
+                  progress=None, *, half: float = 0.75,
+                  max_height: int = 480) -> dict[int, dict]:
+    """Download a short clip around each ``{step_index: time}`` and keep the sharpest
+    frame; return ``{idx: {"time", "path"}}``. One yt-dlp call fetches every section."""
     os.makedirs(out_dir, exist_ok=True)
-    out: dict[int, dict] = {}
     items = list(times.items())
+    if not items:
+        return {}
+    clips_dir = os.path.join(out_dir, "_clips")
+    os.makedirs(clips_dir, exist_ok=True)
+    ranges = [(max(0.0, t - half), t + half) for _, t in items]
+    opts = {
+        "quiet": True, "no_warnings": True,
+        "format": (f"bv*[height<={max_height}][vcodec!=none]/"
+                   f"best[height<={max_height}]/best[vcodec!=none]/best"),
+        "paths": {"home": clips_dir},
+        "outtmpl": {"default": "sec_%(section_start)0.2f.%(ext)s"},
+        "download_ranges": download_range_func(None, ranges),
+        "force_keyframes_at_cuts": True,
+    }
+    if cookiefile:
+        opts["cookiefile"] = cookiefile
+    if proxy:
+        opts["proxy"] = proxy
+        # yt-dlp trusts only certifi's own bundle by default and ignores SSL_CERT_FILE.
+        # When an HTTPS-proxy CA has been installed into a combined bundle (SSL_CERT_FILE,
+        # set by app._install_proxy_ca), switch yt-dlp to the default cert path so it
+        # honors that file — this is what lets it validate the self-signed proxy cert.
+        if os.environ.get("SSL_CERT_FILE"):
+            opts["compat_opts"] = ["no-certifi"]
+    with YoutubeDL(opts) as ydl:
+        ydl.download([f"https://www.youtube.com/watch?v={video_id}"])
+
+    import glob
+    out: dict[int, dict] = {}
     for n, (idx, t) in enumerate(items):
         if progress:
             progress((n + 1) / max(1, len(items)), desc=f"Screenshot {n + 1}/{len(items)}")
-        path = grab_best_frame(stream_url, t, out_dir, idx, proxy)
+        start = max(0.0, t - half)
+        matches = [m for m in glob.glob(os.path.join(clips_dir, f"sec_{start:.2f}.*"))
+                   if not m.endswith(".part")]
+        if not matches:
+            continue
+        # t sits ``t - start`` seconds into its clip (== half, except when clamped at 0)
+        path = _best_frame_from_clip(matches[0], out_dir, idx, t - start)
         if path:
             out[idx] = {"time": t, "path": path}
     return out