siftlode/backend/app/downloads/edit.py
npeter83 7a6f351e64 fix(downloads): repair 5 confirmed backend bugs (B1-B5)
- B1 (sticky errored asset): get_or_create_asset now resets a reused status=='error'
  asset back to 'pending' (+clears the error), so a fresh enqueue/edit of a once-failed
  (source,format) pair actually re-downloads instead of the worker short-circuiting the
  new job with the stale error. Errored assets carry no expires_at, so without this the
  pair was permanently poisoned for all users until someone hit resume.
- B2 (ref_count leak): _release_asset counts 'error' as a holding state, so deleting an
  errored job decrements the ref_count that enqueue always incremented (the worker never
  decrements on failure). Errored rows are deliberately NOT deleted here — a concurrent
  B1 reuse could otherwise be lost-updated + FK-nulled; the row is fileless and harmless.
- B3: admin storage dashboard reads sysconfig.get_int(db,'download_total_max_bytes')
  (the admin-editable DB value GC enforces) instead of the raw env default.
- B4: the single-trim branch of normalize_edit_spec guards its float() coercion like the
  crop/segments branches — a malformed trim now yields a 400, not an unhandled 500.
- B5: formats.normalize guards int(max_height) → falls back to "best" instead of 500.

Reviewed (race in an earlier B2 draft caught + fixed); localdev boots, B4/B5 unit-verified.
2026-07-11 16:15:12 +02:00

371 lines
15 KiB
Python
Raw Permalink 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 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:
# 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 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 0100 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 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}