Files
siftlode/backend/app/downloads/edit.py
T
peter c4942eb590 fix(r5): address the fifth /code-review high round — 6 findings
- run_ffmpeg stops the process from ONE exit path (`except BaseException`), so
  an exception out of on_progress (a DB write) can no longer orphan a live
  ffmpeg after a 10s join on its pipes; _stop is now a no-op on a dead process
- both run_ffmpeg callbacks are throttled to ~1s (they were a DB round-trip per
  output line: ~10^5 queries on a long re-encode), mirroring the download hook
- volume arithmetic extracted to lib/playerVolume with unit tests; a corrupt
  level now falls back to the LAST GOOD one instead of full blast, and the
  sanitising happens where the stored value enters
- RSS is back to two except clauses (the shared tail already sits after the try)
- new tests: callback-exception cleanup, the two throttles; the fake Popen grew
  poll() and its wait count is now asserted

Mutation-checked: each new group goes red when its fix is reverted.
2026-07-24 01:50:50 +02:00

550 lines
23 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""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 0100 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 24120 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}