siftlode/backend/app/downloads/edit.py

"""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 os
import queue
import shutil
import subprocess
import threading
import time
from pathlib import Path

from app.config import settings

log = logging.getLogger("siftlode.edit")


def encode_thread_cap() -> int:
    """How many encoder threads a browser-safe re-encode may use. `download_ffmpeg_threads` when the
    admin pinned one (>0), else reserve a core for foreground work (max(1, cpu_count - 1)). Shared by
    the ffmpeg `-threads` flag and the child's `OMP_NUM_THREADS` so both honour the same budget."""
    n = settings.download_ffmpeg_threads
    if n > 0:
        return n
    return max(1, (os.cpu_count() or 2) - 1)


class EditAborted(Exception):
    """Raised by run_ffmpeg when the job was paused/canceled mid-encode."""


# --- edit spec ------------------------------------------------------------------------------

# SEC-13: ceiling on a join's cut-list. An accurate join emits two filters per segment into a single
# ffmpeg `filter_complex`, so "however many the caller sent" is an unbounded command line.
_MAX_SEGMENTS = 100


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"])
        # SEC-13: an accurate join builds one ffmpeg `filter_complex` with two filters per segment,
        # so an unbounded list becomes an unbounded command line and an ffmpeg that never finishes.
        # The cap is far above any real cut-list (the UI's own splitter produces a handful) and the
        # excess is DROPPED rather than raising: the surplus segments are the attack, the first
        # _MAX_SEGMENTS are a legitimate edit.
        del norm[_MAX_SEGMENTS:]
        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 escalate_kill(proc: subprocess.Popen, terminate, kill, grace_s: float, label: str) -> None:
    """terminate → wait(grace) → kill → wait(grace) escalation for a process we've given up on, so it
    is never waited on unboundedly. `terminate`/`kill` deliver the stop to whatever scope the caller
    owns — a single process (`proc.terminate`/`proc.kill`) or its whole group (`os.killpg`). No-op if
    the process already exited."""
    if proc.poll() is not None:
        return
    terminate()
    try:
        proc.wait(timeout=grace_s)
        return
    except subprocess.TimeoutExpired:
        pass
    kill()
    try:
        proc.wait(timeout=grace_s)
    except subprocess.TimeoutExpired:
        log.error("%s pid %s survived SIGKILL", label, proc.pid)


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."""
    escalate_kill(proc, proc.terminate, proc.kill, grace_s=5, label="ffmpeg")


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 run_ffprobe_json(
    path: Path,
    *,
    show_streams: bool = False,
    show_format: bool = False,
    entries: str | None = None,
) -> dict | None:
    """One ffprobe call → its parsed JSON dict, or None on any failure (timeout, non-zero exit,
    missing binary, unparseable output). The SINGLE ffprobe-JSON-invocation point in the downloads
    package: the browser-playability codec probe here and the shared media probe (`probe.probe_media`,
    used by both the download worker's finalize and the upload ingest) both go through it, so the
    subprocess flags/timeout/error handling live in one place instead of drifting copies."""
    args = ["ffprobe", "-v", "error"]
    if show_streams:
        args.append("-show_streams")
    if show_format:
        args.append("-show_format")
    if entries:
        args += ["-show_entries", entries]
    args += ["-of", "json", str(path)]
    try:
        out = subprocess.run(args, capture_output=True, text=True, timeout=30, check=True).stdout
        return json.loads(out)
    except (subprocess.TimeoutExpired, subprocess.CalledProcessError, OSError, ValueError):
        return None


def _codecs_from_streams(streams: list) -> tuple[str | None, str | None]:
    """(video_codec, audio_codec) lowercased from an ffprobe JSON stream list — the first of each kind.

    Parsed from the JSON `streams` list, NOT ffprobe's 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. Reading each
    stream dict by its own `codec_type` is immune to that ordering."""
    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)."""
    data = run_ffprobe_json(path, entries="stream=codec_type,codec_name")
    if data is None:
        return None, None
    return _codecs_from_streams(data.get("streams") or [])


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, threads: int | None = None
) -> 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. `threads` caps the encoder so a
    re-encode can't claim every core on the shared box (see the worker-child de-prioritization)."""
    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"]
    if threads and threads > 0:
        cmd += ["-threads", str(threads)]
    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 build_thumbnail_jpg_cmd(src: Path, dest: Path) -> list[str]:
    """Decode a downloaded thumbnail (webp/png/…) to a single baseline JPEG frame."""
    return [
        "ffmpeg",
        "-y",
        "-hide_banner",
        "-loglevel",
        "error",
        "-nostdin",
        "-i",
        str(src),
        "-frames:v",
        "1",
        "-update",
        "1",
        str(dest),
    ]


def _run_ffmpeg_best_effort(cmd: list[str], timeout_s: int) -> Exception | None:
    """Run a one-shot ffmpeg command, bounded by `timeout_s`. Returns None on success, or the
    exception on a timeout/non-zero-exit/OS error — never raises. Shared by the best-effort image
    steps (poster / thumbnail / storyboard) so the run+timeout+swallow policy lives in ONE place;
    each caller logs its own context from the returned exception. (The download/transcode path uses
    the richer `run_ffmpeg` instead, which streams progress and honours cooperative cancel.)"""
    try:
        subprocess.run(cmd, capture_output=True, timeout=timeout_s, check=True)
        return None
    except (subprocess.TimeoutExpired, subprocess.CalledProcessError, OSError) as exc:
        return exc


def _is_jpeg(path: Path) -> bool:
    """True if the file starts with the JPEG SOI marker (FF D8 FF)."""
    try:
        with path.open("rb") as f:
            return f.read(3) == b"\xff\xd8\xff"
    except OSError:
        return False


def thumbnail_to_jpg(src: Path | None, dest: Path, timeout_s: int = 30) -> Path | None:
    """Turn a downloaded thumbnail (webp / png / jpeg — whatever the source shipped) into a real
    baseline JPEG for the `<name>.jpg` poster + og:image share sidecar.

    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). A thumbnail
    that is already JPEG is copied as-is; anything else is decoded via ffmpeg. Best-effort: returns
    None on any failure, so the caller falls back to a cut video frame (`ensure_poster`) and the
    download still succeeds."""
    if src is None or not src.exists():
        return None
    if _is_jpeg(src):  # already JPEG (e.g. Facebook) — no decode/re-encode needed
        try:
            shutil.copyfile(src, dest)
            return dest
        except OSError as exc:
            log.info("thumbnail copy failed (%s); poster falls back to a cut frame", str(exc)[:120])
            return None
    err = _run_ffmpeg_best_effort(build_thumbnail_jpg_cmd(src, dest), timeout_s)
    if err is not None:
        log.info(
            "thumbnail->jpg conversion failed (%s); poster falls back to a cut frame",
            str(err)[: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
    `<base>.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)
    err = _run_ffmpeg_best_effort(build_poster_cmd(src, dest, at), timeout_s)
    if err is not None:
        log.info("poster gen skipped for asset %s: %s", asset.id, str(err)[: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)
    err = _run_ffmpeg_best_effort(cmd, timeout_s)
    if err is not None:
        log.info("storyboard gen skipped for asset %s: %s", asset.id, str(err)[:120])
        return None
    if not dest.exists():
        return None
    asset.storyboard_path = rel  # caller commits
    return {**geom, "rel": rel}