"""Video editor (phase 2): per-user trim/crop derivatives via ffmpeg. An edit derives a NEW per-user clip from an already-downloaded ready asset. It is never shared- cached across users (the cache key embeds the user id), so it fully counts against the owner's quota — but it still rides the same `MediaAsset`/`DownloadJob`/worker machinery as a download, with the worker branching on the asset's `source_kind` to run ffmpeg instead of yt-dlp. * trim, fast → stream copy (`-c copy`), instant, no quality loss, cuts SNAP to keyframes (the in/out point can be off by a second or two) * trim, accurate → re-encode (libx264 + aac), frame-accurate; cost scales with CLIP length (the `-ss` seeks first), so a short clip is cheap even from a long source * crop (± trim) → always re-encode (a filter can't be stream-copied), frame-accurate `edit_spec = {trim?: {start_s, end_s}, crop?: {x, y, w, h}, accurate?: bool}`. `accurate` is the per-edit user choice for a trim-only cut (ignored when a crop forces a re-encode anyway). `split` is a frontend concept: the UI fans a split out into N trim jobs, so there is one output file per job here. Also builds the editor **filmstrip** — a single tiled sprite of the source video used as a visual scrub track. Cached on the source asset's `storyboard_path` (reserved in phase 1) and reusable for a player hover-preview later. """ import hashlib import json import logging import math import queue import subprocess import threading import time from pathlib import Path log = logging.getLogger("siftlode.edit") class EditAborted(Exception): """Raised by run_ffmpeg when the job was paused/canceled mid-encode.""" # --- edit spec ------------------------------------------------------------------------------ def normalize_edit_spec(spec: dict | None) -> dict: """Canonicalize a raw edit spec: clamp/round trim, coerce crop to ints, drop empties. The result is what gets hashed (so two equivalent specs share a cache row) and stored.""" spec = spec or {} out: dict = {} # Crop applies to both a single trim and a multi-segment join. crop = spec.get("crop") or {} if crop: try: c = {k: int(round(float(crop[k]))) for k in ("x", "y", "w", "h")} except (KeyError, TypeError, ValueError): c = None if c and c["w"] > 0 and c["h"] > 0 and c["x"] >= 0 and c["y"] >= 0: out["crop"] = c # Multi-segment JOIN (cut-list → one concatenated file). Takes precedence over `trim`. segs = spec.get("segments") if isinstance(segs, list) and segs: norm: list[dict] = [] for s in segs: try: st = max(0.0, float((s or {}).get("start_s") or 0.0)) end_raw = (s or {}).get("end_s") if end_raw is None: continue en = float(end_raw) except (TypeError, ValueError): continue if en > st: norm.append({"start_s": round(st, 3), "end_s": round(en, 3)}) norm.sort(key=lambda x: x["start_s"]) if norm: out["segments"] = norm # A join always chooses a codec path: accurate=filter-concat re-encode, # fast=demuxer-concat stream-copy (keyframe-snapped). out["accurate"] = bool(spec.get("accurate", True)) return out # Single TRIM (one output file; the frontend fans a "separate" split out into N of these). trim = spec.get("trim") or {} if trim: # Guard the numeric coercion like the crop/segments branches do — a malformed value must # drop the trim (→ empty spec → 400 EditError), not raise an unhandled 500. try: start = max(0.0, float(trim.get("start_s") or 0.0)) end_raw = trim.get("end_s") end = float(end_raw) if end_raw is not None else None except (TypeError, ValueError): start, end = 0.0, None t: dict = {"start_s": round(start, 3)} if end is not None and end > start: t["end_s"] = round(end, 3) # A trim with only a start (open-ended) is valid (cut to the end). if t.get("start_s") or "end_s" in t: out["trim"] = t # `accurate` only changes behaviour for a trim-only cut (a crop always re-encodes). Default to # frame-accurate — an editor should cut where you asked; the user opts into fast/keyframe. if out.get("trim") and "crop" not in out: out["accurate"] = bool(spec.get("accurate", True)) return out def edit_sig(spec: dict) -> str: """Stable 24-char signature of a (normalized) edit spec — the per-user cache identity.""" payload = json.dumps(normalize_edit_spec(spec), sort_keys=True, separators=(",", ":")) return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:24] def has_crop(spec: dict) -> bool: return bool(spec.get("crop")) def needs_reencode(spec: dict) -> bool: """A crop always re-encodes; a trim re-encodes only when the user asked for an accurate cut.""" return bool(spec.get("crop")) or bool(spec.get("accurate")) def clip_duration(spec: dict, source_duration: int | None) -> int | None: """Duration of the derived clip (seconds), for display + progress totals.""" segs = spec.get("segments") if segs: return max(0, round(sum(s["end_s"] - s["start_s"] for s in segs))) trim = spec.get("trim") if not trim: return source_duration start = trim.get("start_s") or 0.0 end = trim.get("end_s") if end is None: end = source_duration if end is None: return None return max(0, round(end - start)) # --- ffmpeg: trim / crop -------------------------------------------------------------------- def build_edit_cmd(src: Path, dest: Path, spec: dict, out_ext: str) -> list[str]: """ffmpeg command for a trim/crop edit. `-progress pipe:1` streams machine-readable progress. Trim uses fast input seek (`-ss` before `-i`) + `-t duration`, which is unambiguous across ffmpeg versions. Trim-only stream-copies; a crop forces a re-encode (a filter can't be stream-copied).""" trim = spec.get("trim") or {} start = trim.get("start_s") or 0.0 end = trim.get("end_s") crop = spec.get("crop") reencode = needs_reencode(spec) cmd = [ "ffmpeg", "-y", "-hide_banner", "-loglevel", "error", "-nostdin", "-progress", "pipe:1", "-nostats", ] if start: cmd += ["-ss", f"{start:.3f}"] cmd += ["-i", str(src)] if end is not None: dur = end - start if dur > 0: cmd += ["-t", f"{dur:.3f}"] if reencode: # Input -ss + re-encode is frame-accurate (decodes from the prior keyframe, discards up to # the exact cut). A crop adds the filter; an accurate trim re-encodes with no filter. if crop: cmd += ["-vf", f"crop={crop['w']}:{crop['h']}:{crop['x']}:{crop['y']}"] cmd += [ "-c:v", "libx264", "-preset", "veryfast", "-crf", "20", "-c:a", "aac", "-b:a", "192k", ] else: cmd += ["-c", "copy", "-avoid_negative_ts", "make_zero"] if out_ext in ("mp4", "m4a", "mov"): cmd += ["-movflags", "+faststart"] cmd += [str(dest)] return cmd def _concat_quote(path: Path) -> str: """Escape a path for a single-quoted `file '…'` line of an ffmpeg concat list. ffmpeg's quoting rule: inside single quotes EVERY character is literal except `'` itself, which is written `'\\''` (close the quote, emit an escaped quote, reopen). Backslashes are literal there — doubling them would corrupt a path that contains one. On-disk names have been ASCII-slugged since 0.51.0, but that change is FORWARD-ONLY: a pre-0.51.0 file still on disk can be named `Don't_….mp4`, and an unescaped apostrophe makes the whole list unparseable.""" return path.as_posix().replace("'", "'\\''") def build_concat_plan(src: Path, dest: Path, spec: dict, out_ext: str, staging: Path): """Plan a multi-segment JOIN (cut-list → one file). Returns (cmd, prep_files) where prep_files maps a path → text content the worker must write before running (a concat list, if any). * accurate/crop → filter_complex trim+concat, frame-accurate re-encode * fast → concat demuxer with per-segment inpoint/outpoint, stream-copy (keyframe-snapped, like the fast single trim)""" segs: list[dict] = spec["segments"] crop = spec.get("crop") prep: dict[Path, str] = {} head = [ "ffmpeg", "-y", "-hide_banner", "-loglevel", "error", "-nostdin", "-progress", "pipe:1", "-nostats", ] if needs_reencode(spec): cropf = f",crop={crop['w']}:{crop['h']}:{crop['x']}:{crop['y']}" if crop else "" parts, labels = [], [] for i, s in enumerate(segs): st, en = s["start_s"], s["end_s"] parts.append(f"[0:v]trim=start={st}:end={en},setpts=PTS-STARTPTS{cropf}[v{i}]") parts.append(f"[0:a]atrim=start={st}:end={en},asetpts=PTS-STARTPTS[a{i}]") labels.append(f"[v{i}][a{i}]") parts.append(f"{''.join(labels)}concat=n={len(segs)}:v=1:a=1[v][a]") cmd = head + [ "-i", str(src), "-filter_complex", ";".join(parts), "-map", "[v]", "-map", "[a]", "-c:v", "libx264", "-preset", "veryfast", "-crf", "20", "-c:a", "aac", "-b:a", "192k", ] else: listp = staging / "concat.txt" lines = [] for s in segs: lines.append(f"file '{_concat_quote(src)}'") lines.append(f"inpoint {s['start_s']}") lines.append(f"outpoint {s['end_s']}") prep[listp] = "\n".join(lines) + "\n" cmd = head + ["-f", "concat", "-safe", "0", "-i", str(listp), "-c", "copy"] if out_ext in ("mp4", "m4a", "mov"): cmd += ["-movflags", "+faststart"] cmd += [str(dest)] return cmd, prep def _parse_out_time(line: str) -> float | None: """Parse a `-progress` line into elapsed output seconds.""" if line.startswith("out_time_us="): try: return int(line.split("=", 1)[1]) / 1_000_000 except ValueError: return None if line.startswith("out_time_ms="): # (some builds mislabel this as microseconds) try: return int(line.split("=", 1)[1]) / 1_000_000 except ValueError: return None return None # No output on `-progress pipe:1` for this long means ffmpeg is wedged (a stuck network source, a # hung filter) — a download job may legitimately be slow, but it is never SILENT for minutes. _FFMPEG_STALL_TIMEOUT_S = 600.0 # Hard cap after we ask the process to go away; past it we stop waiting on it altogether. _FFMPEG_EXIT_GRACE_S = 10.0 def _pump(stream, sink: "queue.Queue[str | None]") -> None: """Read a text pipe to EOF into a queue (None = EOF). Runs in its own thread so the main loop can time out and poll cancel instead of blocking forever on a silent ffmpeg.""" try: for line in stream: sink.put(line) except (ValueError, OSError): # pipe closed under us by the kill path pass finally: sink.put(None) def _drain_tail(stream, sink: list[str]) -> None: """Consume a pipe to EOF, keeping only its tail. ffmpeg's stderr MUST be read *while it runs*: the OS pipe buffer is ~64KB, and once it fills, ffmpeg blocks forever on its next write — the process never exits, so `proc.wait()` never returns and the worker thread is gone for good. Reading it only after the exit (as this used to) is exactly the deadlock.""" try: for line in stream: sink.append(line) if len(sink) > 200: del sink[:100] except (ValueError, OSError): pass def _stop(proc: subprocess.Popen) -> None: """Terminate, then kill — never wait unbounded on a process we've given up on. Safe to call on a process that already exited (the single exit path below may reach it after a clean wait).""" if proc.poll() is not None: return proc.terminate() try: proc.wait(timeout=5) except subprocess.TimeoutExpired: proc.kill() try: proc.wait(timeout=5) except subprocess.TimeoutExpired: log.error("ffmpeg pid %s survived SIGKILL", proc.pid) def run_ffmpeg(cmd, total_s, on_progress, should_cancel, stall_timeout_s=_FFMPEG_STALL_TIMEOUT_S) -> None: """Run an ffmpeg edit, reporting 0–100 progress and honouring cooperative cancel. Both pipes are drained by threads (see `_drain_tail`) and the main loop polls on a 1s tick, so cancel is honoured within a second even while ffmpeg is quiet, and a wedged process is killed by the stall watchdog instead of pinning a worker thread forever. Raises EditAborted if should_cancel() turns True mid-run, RuntimeError on a nonzero exit or a stall.""" proc = subprocess.Popen( cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, bufsize=1 ) err_tail: list[str] = [] lines: queue.Queue[str | None] = queue.Queue() readers = [ threading.Thread(target=_pump, args=(proc.stdout, lines), daemon=True), threading.Thread(target=_drain_tail, args=(proc.stderr, err_tail), daemon=True), ] for t in readers: t.start() try: last_output = time.monotonic() while True: try: line = lines.get(timeout=1.0) except queue.Empty: line = "" if line is None: # stdout hit EOF — ffmpeg is done (or dying) break if line: last_output = time.monotonic() secs = _parse_out_time(line.strip()) if secs is not None and total_s: on_progress(max(0.0, min(99.0, secs * 100.0 / total_s))) if should_cancel(): raise EditAborted if time.monotonic() - last_output > stall_timeout_s: raise RuntimeError(f"ffmpeg stalled: no progress for {int(stall_timeout_s)}s") try: ret = proc.wait(timeout=_FFMPEG_EXIT_GRACE_S) except subprocess.TimeoutExpired: # pipes closed but the process lingers raise RuntimeError("ffmpeg did not exit after closing its output") except BaseException: # ONE exit path stops the process, so the invariant is unconditional: nothing leaves this # function with ffmpeg still running. Not just our own EditAborted/stall — `on_progress` # writes to the database, and an exception from THERE used to unwind past a live process, # orphaning it (still encoding, no one left to reap it) after a 10s join on its pipes. _stop(proc) raise finally: for t in readers: t.join(timeout=_FFMPEG_EXIT_GRACE_S) for pipe in (proc.stdout, proc.stderr): if pipe: pipe.close() if ret != 0: raise RuntimeError(f"ffmpeg failed ({ret}): {''.join(err_tail)[-300:]}") # --- ffmpeg: universal browser playability -------------------------------------------------- # A downloaded file must play in EVERY browser, whatever the source shipped. Universal = H.264 # video + AAC/MP3 audio in an mp4 container: H.264 is the only video codec all browsers decode # (VP9/AV1 fail on Safari/iOS; HEVC fails on Chrome/Firefox), and mp4/AAC is the safe pairing. # Anything else is fixed post-download: re-encode the offending stream(s), or just remux when the # codecs are already fine but the container isn't. _BROWSER_SAFE_VCODEC = ("h264",) _BROWSER_SAFE_ACODEC = ("aac", "mp3") def parse_probe_streams(payload: str) -> tuple[str | None, str | None]: """(video_codec, audio_codec) out of ffprobe's JSON stream list, lowercased. JSON, not the flat `default=nw=1` output: there the fields arrive in ffprobe's own struct order, which puts `codec_name` BEFORE `codec_type`, so a line-by-line "remember the last codec_type" parser attributes every name to the PREVIOUS stream. Measured on real downloads: an ordinary H.264+AAC mp4 probed as `v=aac a=None` — the browser-playability check then believed the video stream was AAC and re-encoded a perfectly fine file on every download.""" try: streams = json.loads(payload or "{}").get("streams") or [] except ValueError: return None, None vcodec = acodec = None for s in streams: name = (s.get("codec_name") or "").lower() or None ctype = (s.get("codec_type") or "").lower() if ctype == "video" and vcodec is None: vcodec = name elif ctype == "audio" and acodec is None: acodec = name return vcodec, acodec def probe_codecs(path: Path) -> tuple[str | None, str | None]: """Return (video_codec, audio_codec) lowercased via one ffprobe call; None for a missing stream or if ffprobe fails (a failed probe is treated as 'leave it alone' upstream).""" try: out = subprocess.run( ["ffprobe", "-v", "error", "-show_entries", "stream=codec_type,codec_name", "-of", "json", str(path)], capture_output=True, text=True, timeout=30, check=True, ).stdout except (subprocess.TimeoutExpired, subprocess.CalledProcessError, OSError): return None, None return parse_probe_streams(out) def probe_duration(path: Path) -> int: """Media duration in whole seconds via ffprobe, or 0 if unknown — a progress denominator for a re-encode when the yt-dlp info dict carried no `duration` (some URL/live extractions).""" try: out = subprocess.run( ["ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "default=nw=1:nk=1", str(path)], capture_output=True, text=True, timeout=30, check=True, ).stdout.strip() return int(float(out)) if out else 0 except (subprocess.TimeoutExpired, subprocess.CalledProcessError, OSError, ValueError): return 0 def browser_transcode_flags(vcodec: str | None, acodec: str | None, is_mp4: bool) -> tuple[bool, bool, bool]: """Decide what (if anything) makes a file universally browser-playable. Returns (needs_change, reencode_video, reencode_audio). needs_change=False => already H.264 + AAC/MP3 in mp4. A None codec (no such stream, or a failed probe) is never re-encoded.""" rv = bool(vcodec) and vcodec not in _BROWSER_SAFE_VCODEC ra = bool(acodec) and acodec not in _BROWSER_SAFE_ACODEC return (rv or ra or not is_mp4), rv, ra def build_browser_safe_cmd(src: Path, dest: Path, reencode_video: bool, reencode_audio: bool) -> list[str]: """ffmpeg command that lands `src` as a browser-safe mp4 (faststart): re-encode only the stream(s) that need it (libx264 / aac), stream-copy the rest, always into mp4. `-progress pipe:1` streams machine-readable progress for run_ffmpeg.""" cmd = [ "ffmpeg", "-y", "-hide_banner", "-loglevel", "error", "-nostdin", "-progress", "pipe:1", "-nostats", "-i", str(src), ] cmd += ["-c:v", "libx264", "-preset", "veryfast", "-crf", "20"] if reencode_video else ["-c:v", "copy"] cmd += ["-c:a", "aac", "-b:a", "192k"] if reencode_audio else ["-c:a", "copy"] cmd += ["-movflags", "+faststart", str(dest)] return cmd # --- filmstrip (editor scrub track) --------------------------------------------------------- _SB_COLS = 12 _SB_TILE_W = 160 def storyboard_geometry(duration_s: int | None) -> dict: """Deterministic sprite geometry for a source duration (so meta is recomputable, unstored). ~1 frame / 5 s, clamped to fill a 12-wide grid of 24–120 tiles.""" dur = max(1, int(duration_s or 1)) count = max(24, min(120, round(dur / 5))) rows = math.ceil(count / _SB_COLS) count = _SB_COLS * rows # fill the grid exactly return { "cols": _SB_COLS, "rows": rows, "count": count, "interval_s": dur / count, "tile_w": _SB_TILE_W, } def build_storyboard_cmd(src: Path, dest: Path, duration_s: int | None, geom: dict) -> list[str]: """One-pass tiled sprite. fps picks `count` frames evenly across the whole clip.""" dur = max(1, int(duration_s or 1)) fps = geom["count"] / dur vf = f"fps={fps:.6f},scale={_SB_TILE_W}:-2,tile={geom['cols']}x{geom['rows']}" return [ "ffmpeg", "-y", "-hide_banner", "-loglevel", "error", "-nostdin", "-i", str(src), "-frames:v", "1", "-q:v", "5", "-vf", vf, str(dest), ] def build_poster_cmd(src: Path, dest: Path, at_s: float) -> list[str]: """Grab one representative frame as a poster JPEG. `-ss` before `-i` seeks fast.""" return [ "ffmpeg", "-y", "-hide_banner", "-loglevel", "error", "-nostdin", "-ss", f"{max(0.0, at_s):.3f}", "-i", str(src), "-frames:v", "1", "-q:v", "3", "-vf", "scale=640:-2", str(dest), ] def thumbnail_to_jpg(src: Path | None, dest: Path, timeout_s: int = 30) -> Path | None: """Convert a downloaded thumbnail (webp / png / jpeg — whatever the source shipped) into a real baseline JPEG for the `.jpg` poster + og:image share sidecar. Uses a plain ffmpeg decode→mjpeg, deliberately NOT yt-dlp's `FFmpegThumbnailsConvertor` (whose `-bsf:v mjpeg2jpeg` path crashes on some already-JPEG thumbnails, e.g. Reddit's, and used to fail the entire download). Best-effort: returns None on any failure, so the caller falls back to a cut video frame via `ensure_poster` and the download still succeeds.""" if src is None or not src.exists(): return None try: subprocess.run( ["ffmpeg", "-y", "-hide_banner", "-loglevel", "error", "-nostdin", "-i", str(src), "-frames:v", "1", "-update", "1", str(dest)], capture_output=True, timeout=timeout_s, check=True, ) except (subprocess.TimeoutExpired, subprocess.CalledProcessError, OSError) as exc: log.info("thumbnail->jpg conversion failed (%s); poster falls back to a cut frame", str(exc)[:120]) return None return dest if dest.exists() else None def ensure_poster(asset, download_root: str, timeout_s: int = 30) -> str | None: """Generate a poster frame for a ready asset that has no thumbnail (e.g. a direct media URL like a reddit HLS manifest, which yt-dlp can't get a thumbnail for). Written as the media's `.jpg` sidecar — so Plex uses it too and `delete_asset_files` cleans it up — and returned as a rel path (or None on failure). Best-effort + time-bounded; a few seconds in (or 10% of a short clip) dodges black intro frames.""" if not asset.rel_path: return None root = Path(download_root) src = root / asset.rel_path if not src.exists(): return None rel = str(Path(asset.rel_path).with_suffix(".jpg")) dest = root / rel if dest.exists(): return rel at = min(3.0, (asset.duration_s or 10) * 0.1) try: subprocess.run(build_poster_cmd(src, dest, at), capture_output=True, timeout=timeout_s, check=True) except (subprocess.TimeoutExpired, subprocess.CalledProcessError, OSError) as exc: log.info("poster gen skipped for asset %s: %s", asset.id, str(exc)[:120]) return None return rel if dest.exists() else None def ensure_storyboard(asset, download_root: str, timeout_s: int = 90) -> dict | None: """Return the filmstrip meta for a ready source asset, generating the sprite on first use. Best-effort + time-bounded: for a very long/high-res source the one-pass decode can be slow, so it's capped by `timeout_s`; on timeout/failure we return None and the editor falls back to a plain timeline (no filmstrip). Caches the sprite path on `asset.storyboard_path`. Returns None (caller should NOT commit) if generation didn't happen; otherwise a meta dict and sets `asset.storyboard_path` (caller commits).""" if not asset.rel_path: return None geom = storyboard_geometry(asset.duration_s) root = Path(download_root) rel = f".storyboards/asset-{asset.id}.jpg" dest = root / rel if asset.storyboard_path and (root / asset.storyboard_path).exists(): return {**geom, "rel": asset.storyboard_path} src = root / asset.rel_path if not src.exists(): return None dest.parent.mkdir(parents=True, exist_ok=True) cmd = build_storyboard_cmd(src, dest, asset.duration_s, geom) try: subprocess.run(cmd, capture_output=True, timeout=timeout_s, check=True) except (subprocess.TimeoutExpired, subprocess.CalledProcessError, OSError) as exc: log.info("storyboard gen skipped for asset %s: %s", asset.id, str(exc)[:120]) return None if not dest.exists(): return None asset.storyboard_path = rel # caller commits return {**geom, "rel": rel}