"""Image + video handling. Stills load directly; videos are sampled into frames (ffmpeg) at the cadence FC sends — so a video becomes a bag of per-frame instances, each with a timestamp.""" import io import logging import os import signal import subprocess import tempfile import time from PIL import Image, ImageFile from .throttle import PidReadMeter log = logging.getLogger("fc_agent.media") # Load slightly-truncated images (a few missing trailing bytes) instead of # raising — matches the server embedder. These are common in scraped libraries # and would otherwise fail the job 3× then error (operator-flagged 2026-06-30). ImageFile.LOAD_TRUNCATED_IMAGES = True # Disable PIL's decompression-bomb guard: this is a TRUSTED local library, not an # untrusted upload surface, so a legitimately huge image (high-res scans/prints, # 90M+ pixels) must load. The default 89M-pixel limit only WARNS, but PIL raises # DecompressionBombError at 2× (~179M px) — which would fail those jobs outright # (operator-flagged 2026-06-30, images of 90–95M px). Image.MAX_IMAGE_PIXELS = None def is_video(mime: str) -> bool: return bool(mime) and (mime.startswith("video/") or mime in {"image/gif"}) def _dhash(img: Image.Image, size: int = 8) -> int: """Difference hash: compare adjacent pixels of a (size+1 × size) grayscale thumbnail → a `size*size`-bit fingerprint. Cheap (64 comparisons on a 72-px thumbnail) and robust to scaling/compression noise — near-identical frames hash within a few bits, a real scene change moves many.""" small = img.convert("L").resize((size + 1, size)) px = list(small.getdata()) bits = 0 for row in range(size): base = row * (size + 1) for col in range(size): bits = (bits << 1) | int(px[base + col] > px[base + col + 1]) return bits def dedupe_frames( frames: list[tuple[float, Image.Image]], min_distance: int ) -> list[tuple[float, Image.Image]]: """Drop visually near-duplicate frames. A near-static video sampled into many frames re-runs the WHOLE detect→CCIP→SigLIP chain on ~identical frames — the dominant video load. Greedy perceptual-hash dedup: keep a frame only if its dHash differs from every already-kept frame by >= min_distance bits (Hamming), so a static run collapses to one frame while genuinely distinct scenes all survive. Order + timestamps preserved. CPU-only (64-bit int XORs), so it runs in the decode stage and spares the GPU the skipped frames entirely. min_distance is the coarseness dial: higher keeps more frames (safer for brief localized changes an 8×8 hash can miss), 0 disables. The first frame is always kept (nothing to compare against).""" if min_distance <= 0 or len(frames) <= 1: return frames kept: list[tuple[float, Image.Image]] = [] hashes: list[int] = [] for t, frame in frames: h = _dhash(frame) if all(bin(h ^ k).count("1") >= min_distance for k in hashes): hashes.append(h) kept.append((t, frame)) return kept def to_rgb(img: Image.Image) -> Image.Image: """RGB, flattening any transparency onto white first. A naive convert('RGB') on a palette-with-transparency image (common for character PNGs on a clear background) lets PIL guess the transparent pixels — usually black artifacts that bleed into the crop + the embedding (and the "should be converted to RGBA" warning). Compositing over white gives a clean, consistent background.""" if img.mode in ("RGBA", "LA", "PA") or ( img.mode == "P" and "transparency" in img.info ): img = img.convert("RGBA") bg = Image.new("RGBA", img.size, (255, 255, 255, 255)) return Image.alpha_composite(bg, img).convert("RGB") return img.convert("RGB") def load_image(data: bytes) -> Image.Image: return to_rgb(Image.open(io.BytesIO(data))) # ffmpeg reconnect flags — resume a dropped HTTP transfer (a slow/contended media # store can cut a long stream) instead of failing the whole job. Relies only on # HTTP + Range, which every FC deployment serves → environment-agnostic. _RECONNECT = [ "-reconnect", "1", "-reconnect_streamed", "1", "-reconnect_on_network_error", "1", "-reconnect_delay_max", "5", ] def _collect_frames( tmp: str, interval: float, cap: int ) -> list[tuple[float, Image.Image]]: out: list[tuple[float, Image.Image]] = [] names = sorted(n for n in os.listdir(tmp) if n.startswith("f_")) for i, name in enumerate(names[:cap]): with Image.open(os.path.join(tmp, name)) as im: out.append((round(i * interval, 2), to_rgb(im))) return out def _terminate(proc: subprocess.Popen) -> None: """Stop an ffmpeg cleanly, then hard-kill if it ignores SIGTERM.""" try: # A bandwidth-paused (SIGSTOPped) process can't receive SIGTERM until it # resumes — always CONT first so termination is prompt, not queued. proc.send_signal(signal.SIGCONT) except OSError: pass proc.terminate() try: proc.wait(timeout=2) except subprocess.TimeoutExpired: proc.kill() try: proc.wait(timeout=2) except subprocess.TimeoutExpired: pass def _pause(proc: subprocess.Popen, seconds: float, should_stop) -> bool: """SIGSTOP ffmpeg for ~`seconds` of bandwidth debt, staying responsive to Stop. While paused, the kernel socket buffer fills and TCP flow control stalls curator's send side — that's the throttle. SIGCONT is ALWAYS sent before returning. False = a Stop arrived mid-pause.""" try: proc.send_signal(signal.SIGSTOP) except OSError: return True # already exited — nothing to pause try: end = time.monotonic() + seconds while (left := end - time.monotonic()) > 0: if should_stop and should_stop(): return False time.sleep(min(0.5, left)) return True finally: try: proc.send_signal(signal.SIGCONT) except OSError: pass def sample_frames_from_url( url: str, interval_seconds: float, max_frames: int, *, headers: str = "", timeout: float = 1200.0, should_stop=None, governor=None, ) -> tuple[list[tuple[float, Image.Image]], str | None]: """Sample frames by pointing ffmpeg STRAIGHT at the media URL — it Range-reads only the video index + up to max_frames worth of content, so the agent never downloads the whole file (VR/4K originals run 800MB+ and would buffer ~1GB in RAM and get cut off mid-download). Reconnect flags resume a dropped transfer; the timeout is the per-video ceiling (a slow/reconnecting stream can otherwise run for minutes). `should_stop` is polled while ffmpeg runs so a Stop KILLS the subprocess at once — otherwise a downloader stuck in a long decode keeps the agent "working" long after Stop. `governor` (the worker's shared TokenBucket) meters ffmpeg's network reads from outside via /proc//io and SIGSTOPs the process into budget, so video streaming honors the same aggregate bandwidth cap as still downloads. Returns (frames, reason): frames is empty on failure/stop/timeout, and `reason` then carries the SPECIFIC cause (ffmpeg's stderr tail / timeout) so the caller can put it in the job's error — a bare "no frames" hid a filter bug as "unprocessable" for weeks. None reason on success.""" interval = max(0.5, float(interval_seconds or 4.0)) cap = max(1, int(max_frames or 64)) hdr = ["-headers", headers] if headers else [] # select (NOT the fps filter): always keep the FIRST frame, then one per # `interval` seconds of timestamp. fps=1/N emits round(duration/N) frames, # which is ZERO for any clip shorter than ~N/2 seconds — a whole class of # short animation loops failed as "unprocessable" that way (operator-flagged # 2026-07-02: 0.5s/1.75s clips). scale=out_range=full converts limited-range # yuv420p to full range so the mjpeg (jpg) encoder accepts it at default # strictness instead of erroring on "non full-range YUV". vf = ( f"select='isnan(prev_selected_t)+gte(t-prev_selected_t\\,{interval})'," "scale=out_range=full" ) with tempfile.TemporaryDirectory() as tmp: pattern = os.path.join(tmp, "f_%05d.jpg") cmd = ["ffmpeg", "-nostdin", "-loglevel", "error", *_RECONNECT, *hdr, "-i", url, "-vf", vf, "-fps_mode", "vfr", "-frames:v", str(cap), "-q:v", "3", pattern] # ffmpeg's stderr goes to a file (not a PIPE, which could fill and # deadlock; not DEVNULL, which is how a filter bug hid as "unprocessable" # for weeks) — on failure its tail is logged so the operator can see WHY. errpath = os.path.join(tmp, "stderr.txt") try: with open(errpath, "wb") as errf: proc = subprocess.Popen( cmd, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=errf, ) meter = PidReadMeter(proc.pid) if governor is not None else None # Poll rather than block, so a Stop (or the per-video timeout) can # kill a slow/wedged ffmpeg promptly instead of waiting it out. start = time.monotonic() while True: try: proc.wait(timeout=0.5) break except subprocess.TimeoutExpired: stopped = should_stop and should_stop() if stopped or (time.monotonic() - start > timeout): _terminate(proc) if stopped: return [], "stopped" log.warning("ffmpeg timed out after %.0fs: %s", timeout, url) return [], f"ffmpeg timed out after {timeout:.0f}s" if meter is not None: read = meter.delta() if read is None: # /proc gone → stop governing meter = None elif (debt := governor.charge(read)) > 0: # Over budget: pause ffmpeg until the bucket # recovers. Pause time counts toward `timeout` # (it stays the wedge backstop either way). if not _pause(proc, debt, should_stop): _terminate(proc) return [], "stopped" except (OSError, ValueError) as exc: return [], f"ffmpeg not runnable: {exc}" frames = _collect_frames(tmp, interval, cap) if not frames: reason = f"ffmpeg exit {proc.returncode}: {_tail(errpath)}" log.warning("ffmpeg produced no frames for %s — %s", url, reason) return [], reason return frames, None def _tail(path: str, limit: int = 300) -> str: """Last `limit` chars of a (stderr) file, flattened — for failure logs.""" try: with open(path, "rb") as f: f.seek(0, os.SEEK_END) f.seek(max(0, f.tell() - limit)) return f.read().decode("utf-8", "replace").replace("\n", " ").strip() except OSError: return "?"