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 subprocess
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:
        start = max(0.0, float(trim.get("start_s") or 0.0))
        end_raw = trim.get("end_s")
        t: dict = {"start_s": round(start, 3)}
        if end_raw is not None:
            end = float(end_raw)
            if 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 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 '{src.as_posix()}'")
            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


def run_ffmpeg(cmd, total_s, on_progress, should_cancel) -> None:
    """Run an ffmpeg edit, reporting 0–100 progress and honouring cooperative cancel.

    Raises EditAborted if should_cancel() turns True mid-run, RuntimeError on a nonzero exit."""
    proc = subprocess.Popen(
        cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, bufsize=1
    )
    try:
        assert proc.stdout is not None
        for line in proc.stdout:
            line = line.strip()
            if should_cancel():
                proc.terminate()
                try:
                    proc.wait(timeout=5)
                except subprocess.TimeoutExpired:
                    proc.kill()
                raise EditAborted
            secs = _parse_out_time(line)
            if secs is not None and total_s:
                on_progress(max(0.0, min(99.0, secs * 100.0 / total_s)))
    finally:
        if proc.stdout:
            proc.stdout.close()
    ret = proc.wait()
    if ret != 0:
        err = (proc.stderr.read() if proc.stderr else "") or ""
        raise RuntimeError(f"ffmpeg failed ({ret}): {err[:300]}")


# --- 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 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)
    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}