diff --git a/backend/alembic/versions/0041_download_edit.py b/backend/alembic/versions/0041_download_edit.py new file mode 100644 index 0000000..d077a12 --- /dev/null +++ b/backend/alembic/versions/0041_download_edit.py @@ -0,0 +1,66 @@ +"""download editor (phase 2): per-user trim/crop derivatives + +Revision ID: 0041_download_edit +Revises: 0040_reset_profiles +Create Date: 2026-07-04 + +Adds the editor columns to `download_jobs`. An edit job (`job_kind='edit'`) derives a new per-user +clip from an already-downloaded job by running ffmpeg (trim = stream-copy, crop = re-encode). It +still hangs off a `media_assets` row (with `source_kind='edit'` and a per-user cache key) so the +existing file-serve / ref-count / GC / quota machinery is reused unchanged. `source_job_id` / +`source_asset_id` point at the parent download; `edit_spec` is the trim/crop recipe. +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "0041_download_edit" +down_revision: Union[str, None] = "0040_reset_profiles" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.add_column( + "download_jobs", + sa.Column( + "job_kind", sa.String(length=16), server_default="download", nullable=False + ), + ) + op.add_column( + "download_jobs", sa.Column("source_job_id", sa.Integer(), nullable=True) + ) + op.add_column( + "download_jobs", sa.Column("source_asset_id", sa.Integer(), nullable=True) + ) + op.add_column("download_jobs", sa.Column("edit_spec", sa.JSON(), nullable=True)) + op.create_foreign_key( + "fk_download_jobs_source_job", + "download_jobs", + "download_jobs", + ["source_job_id"], + ["id"], + ondelete="SET NULL", + ) + op.create_foreign_key( + "fk_download_jobs_source_asset", + "download_jobs", + "media_assets", + ["source_asset_id"], + ["id"], + ondelete="SET NULL", + ) + + +def downgrade() -> None: + op.drop_constraint( + "fk_download_jobs_source_asset", "download_jobs", type_="foreignkey" + ) + op.drop_constraint( + "fk_download_jobs_source_job", "download_jobs", type_="foreignkey" + ) + op.drop_column("download_jobs", "edit_spec") + op.drop_column("download_jobs", "source_asset_id") + op.drop_column("download_jobs", "source_job_id") + op.drop_column("download_jobs", "job_kind") diff --git a/backend/app/downloads/edit.py b/backend/app/downloads/edit.py new file mode 100644 index 0000000..3ecc80a --- /dev/null +++ b/backend/app/downloads/edit.py @@ -0,0 +1,260 @@ +"""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 = {} + + 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 + + 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 + + # `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.""" + 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 _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 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} diff --git a/backend/app/downloads/service.py b/backend/app/downloads/service.py index ad2f6bd..27204e2 100644 --- a/backend/app/downloads/service.py +++ b/backend/app/downloads/service.py @@ -12,10 +12,19 @@ from sqlalchemy import func, select from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session +from app.downloads import edit as editmod from app.downloads import formats, quota from app.models import Channel, DownloadJob, MediaAsset, Video +class EditError(Exception): + """Bad edit request (empty spec / source not ready); `reason` is an i18n-mappable key.""" + + def __init__(self, reason: str): + self.reason = reason + super().__init__(reason) + + def source_url(source_kind: str, source_ref: str) -> str: """The URL yt-dlp should fetch. YouTube stores the 11-char id; ad-hoc stores the URL.""" if source_kind == "youtube": @@ -121,3 +130,71 @@ def enqueue( db.commit() db.refresh(job) return job + + +def _clip_title(source_title: str | None, display_name: str | None) -> str: + if display_name: + return display_name[:255] + base = source_title or "Clip" + return f"{base} (clip)"[:255] + + +def enqueue_edit( + db: Session, + user_id: int, + source_job: DownloadJob, + edit_spec: dict, + display_name: str | None = None, +) -> DownloadJob: + """Queue a phase-2 editor derivative (trim/crop) of `source_job` for `user_id`. + + The derivative is per-user (never shared-cached): its asset uses `source_kind='edit'` with a + cache key that embeds the user id, so an identical re-edit by the same user is a cache hit but + it never dedups across users. Reuses the same worker/quota/GC path as a download. + + Raises quota.QuotaExceeded (holdings cap) or EditError (empty spec / source not ready).""" + src_asset = db.get(MediaAsset, source_job.asset_id) if source_job.asset_id else None + if src_asset is None or src_asset.status != "ready" or not src_asset.rel_path: + raise EditError("source_not_ready") + + spec = editmod.normalize_edit_spec(edit_spec) + if not spec: + raise EditError("empty_edit") + + quota.check_enqueue(db, user_id) + + sig = editmod.edit_sig(spec) + ref = f"{user_id}:{src_asset.id}"[:512] + asset = get_or_create_asset(db, "edit", ref, sig) + if not asset.title: + asset.title = _clip_title(src_asset.title, display_name) + asset.uploader = src_asset.uploader + asset.thumbnail_url = src_asset.thumbnail_url + asset.upload_date = src_asset.upload_date + asset.duration_s = editmod.clip_duration(spec, src_asset.duration_s) + + job = DownloadJob( + user_id=user_id, + asset_id=asset.id, + job_kind="edit", + source_kind="edit", + source_ref=ref, + source_job_id=source_job.id, + source_asset_id=src_asset.id, + edit_spec=spec, + profile_snapshot={}, + display_name=display_name or asset.title, + status="queued", + queue_pos=_next_queue_pos(db), + ) + db.add(job) + asset.ref_count = (asset.ref_count or 0) + 1 + + if asset.status == "ready": # identical clip already produced for this user → instant + job.status = "done" + job.progress = 100 + asset.last_access_at = datetime.now(timezone.utc) + + db.commit() + db.refresh(job) + return job diff --git a/backend/app/downloads/storage.py b/backend/app/downloads/storage.py index 1c4f166..d15bb3a 100644 --- a/backend/app/downloads/storage.py +++ b/backend/app/downloads/storage.py @@ -87,6 +87,15 @@ def rel_path(meta: MediaMeta, ext: str, layout: str = "plex") -> str: return f"{channel}/Season_{year}/{epname}" +def edit_rel_path(user_id: int, asset_id: int, ext: str) -> str: + """On-disk path (relative to DOWNLOAD_ROOT) for a phase-2 editor derivative. + + Kept in a per-user `.edits/` tree — device-downloadable via the file endpoint, but NOT part of + the Plex library (these are user clips, not canonical episodes; there are no .nfo/poster + sidecars). The name is system-decided and bound to the derivative asset id.""" + return f".edits/{user_id}/clip_{asset_id}.{ext}" + + def abs_path(download_root: str, rel: str) -> Path: return Path(download_root) / rel diff --git a/backend/app/models.py b/backend/app/models.py index bfe6282..c5aa4b5 100644 --- a/backend/app/models.py +++ b/backend/app/models.py @@ -779,7 +779,13 @@ class DownloadJob(Base, TimestampMixin, UpdatedAtMixin): freezes the spec at enqueue so later edits/deletion of the profile don't change this job. `display_name` is the user's own name for the file (display + the Content-Disposition on local download); the on-disk asset is never renamed. `asset_id` links to the shared file - (SET NULL if the asset is GC'd — the job then shows as expired).""" + (SET NULL if the asset is GC'd — the job then shows as expired). + + Phase 2 (editor): `job_kind='edit'` jobs derive a new per-user clip from an already-downloaded + ready job. They still hang off a `MediaAsset` (with `source_kind='edit'`, a per-user cache key) + so all the file-serve/ref-count/GC/quota machinery is reused; the worker branches on the + asset's `source_kind` and runs ffmpeg instead of yt-dlp. `source_job_id`/`source_asset_id` + point at the parent download the clip was cut from; `edit_spec` is the (trim/crop) recipe.""" __tablename__ = "download_jobs" __table_args__ = (Index("ix_download_jobs_claim", "status", "queue_pos"),) @@ -793,6 +799,19 @@ class DownloadJob(Base, TimestampMixin, UpdatedAtMixin): ) source_kind: Mapped[str] = mapped_column(String(16)) # "youtube" | "url" source_ref: Mapped[str] = mapped_column(String(512)) + # "download" (default) or "edit" (a per-user trim/crop derivative of another job). + job_kind: Mapped[str] = mapped_column( + String(16), default="download", server_default="download" + ) + # For edit jobs: the parent download this clip is cut from (SET NULL if the parent is deleted). + source_job_id: Mapped[int | None] = mapped_column( + ForeignKey("download_jobs.id", ondelete="SET NULL") + ) + source_asset_id: Mapped[int | None] = mapped_column( + ForeignKey("media_assets.id", ondelete="SET NULL") + ) + # For edit jobs: the trim/crop recipe {trim?:{start_s,end_s}, crop?:{x,y,w,h}}. + edit_spec: Mapped[dict | None] = mapped_column(JSON) profile_id: Mapped[int | None] = mapped_column( ForeignKey("download_profiles.id", ondelete="SET NULL") ) diff --git a/backend/app/routes/downloads.py b/backend/app/routes/downloads.py index 661f0ae..4c151fd 100644 --- a/backend/app/routes/downloads.py +++ b/backend/app/routes/downloads.py @@ -18,6 +18,7 @@ from sqlalchemy.orm import Session from app.auth import admin_user, require_human from app.config import settings from app.db import get_db +from app.downloads import edit as editmod from app.downloads import quota, service, storage from app.models import ( DownloadJob, @@ -81,6 +82,9 @@ def _serialize(job: DownloadJob, asset: MediaAsset | None) -> dict: "profile_id": job.profile_id, "spec": job.profile_snapshot, "display_name": job.display_name, + "job_kind": job.job_kind, + "source_job_id": job.source_job_id, + "edit_spec": job.edit_spec, "can_download": ready and job.status == "done", "expired": job.asset_id is None and job.status == "done", "asset": None @@ -247,6 +251,36 @@ def enqueue_download( return _serialize(job, asset) +@router.post("/edit") +def enqueue_edit( + payload: dict, user: User = Depends(require_human), db: Session = Depends(get_db) +) -> dict: + """Queue a phase-2 editor derivative (trim/crop) of one of the user's finished downloads.""" + source_job = _own_job(db, user, int(payload.get("source_job_id") or 0)) + if source_job.status != "done" or source_job.asset_id is None: + raise HTTPException(status_code=409, detail="You can only edit a finished download.") + spec = payload.get("edit_spec") + if not isinstance(spec, dict): + raise HTTPException(status_code=400, detail="edit_spec must be an object") + display_name = (payload.get("display_name") or "").strip() or None + try: + job = service.enqueue_edit(db, user.id, source_job, spec, display_name) + except quota.QuotaExceeded as e: + raise HTTPException(status_code=422, detail=_quota_message(e)) + except service.EditError as e: + raise HTTPException(status_code=400, detail=_edit_message(e)) + asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None + return _serialize(job, asset) + + +def _edit_message(e: service.EditError) -> str: + if e.reason == "empty_edit": + return "Choose a trim range or crop area first." + if e.reason == "source_not_ready": + return "The source download isn't ready to edit." + return "That edit couldn't be created." + + def _quota_message(e: quota.QuotaExceeded) -> str: if e.reason == "max_jobs": return f"You've reached your download limit ({e.limit} items). Remove some first." @@ -470,6 +504,49 @@ def download_file( return FileResponse(path, filename=filename) +# --- editor filmstrip (scrub track) -------------------------------------------------------- + +@router.get("/{job_id}/storyboard") +def storyboard_meta( + job_id: int, user: User = Depends(require_human), db: Session = Depends(get_db) +) -> dict: + """Filmstrip geometry for the editor's scrub track, generating the sprite on first use. + + Best-effort: for a very long/high-res source the sprite may not generate in time — then + `available` is False and the editor shows a plain timeline.""" + job = _accessible_job(db, user, job_id) + asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None + if asset is None or asset.status != "ready" or not asset.rel_path: + raise HTTPException(status_code=409, detail="This download isn't ready.") + meta = editmod.ensure_storyboard(asset, settings.download_root) + if meta is None: + return {"available": False} + db.commit() # persist storyboard_path if it was just generated + return { + "available": True, + "cols": meta["cols"], + "rows": meta["rows"], + "count": meta["count"], + "interval_s": meta["interval_s"], + "url": f"/api/downloads/{job_id}/storyboard.jpg", + } + + +@router.get("/{job_id}/storyboard.jpg") +def storyboard_image( + job_id: int, user: User = Depends(require_human), db: Session = Depends(get_db) +): + job = _accessible_job(db, user, job_id) + asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None + if asset is None or not asset.storyboard_path: + raise HTTPException(status_code=404, detail="No filmstrip.") + path = storage.abs_path(settings.download_root, asset.storyboard_path).resolve() + root = Path(settings.download_root).resolve() + if root not in path.parents or not path.exists(): + raise HTTPException(status_code=404, detail="No filmstrip.") + return FileResponse(path, media_type="image/jpeg") + + # --- sharing ------------------------------------------------------------------------------- @router.post("/{job_id}/share") diff --git a/backend/app/worker.py b/backend/app/worker.py index 4633e5e..a38f150 100644 --- a/backend/app/worker.py +++ b/backend/app/worker.py @@ -28,6 +28,7 @@ from sqlalchemy import text from app.config import settings from app.db import SessionLocal +from app.downloads import edit as editmod from app.downloads import formats, quota, service, storage from app.models import DownloadJob, MediaAsset from app.titles import normalize_title @@ -141,6 +142,7 @@ def _process_job(job_id: int) -> None: asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None spec = dict(job.profile_snapshot or {}) source_kind, source_ref = job.source_kind, job.source_ref + job_kind = job.job_kind asset_id = asset.id if asset else None asset_status = asset.status if asset else None @@ -160,13 +162,16 @@ def _process_job(job_id: int) -> None: time.sleep(2) return - # asset_status == "pending": try to win the download. + # asset_status == "pending": try to win the download / edit. if not _claim_asset(asset_id): _set_job(job_id, status="queued", phase="waiting") time.sleep(2) return - _download(job_id, asset_id, source_kind, source_ref, spec) + if job_kind == "edit": + _process_edit(job_id, asset_id) + else: + _download(job_id, asset_id, source_kind, source_ref, spec) def _finish_from_cache(job_id: int, asset_id: int) -> None: @@ -431,6 +436,109 @@ def _reset_asset_pending(asset_id: int) -> None: db.commit() +# --- editor derivatives (phase 2: trim/crop via ffmpeg) ------------------------------------- + +def _edit_staging_dir(asset_id: int) -> Path: + return Path(settings.download_root) / ".staging" / f"edit-{asset_id}" + + +def _fail_edit(job_id: int, asset_id: int, msg: str) -> None: + with SessionLocal() as db: + asset = db.get(MediaAsset, asset_id) + if asset: + asset.status = "error" + asset.error = msg + job = db.get(DownloadJob, job_id) + if job: + job.status = "error" + job.error = msg + job.phase = None + db.commit() + log.warning("edit job %s failed: %s", job_id, msg) + + +def _process_edit(job_id: int, asset_id: int) -> None: + """Produce a per-user trim/crop clip from the source asset's downloaded file via ffmpeg.""" + with SessionLocal() as db: + job = db.get(DownloadJob, job_id) + src = db.get(MediaAsset, job.source_asset_id) if job and job.source_asset_id else None + edit_spec = dict(job.edit_spec or {}) if job else {} + user_id = job.user_id if job else None + src_ready = src is not None and src.status == "ready" and bool(src.rel_path) + src_rel = src.rel_path if src else None + src_w, src_h, src_dur = (src.width, src.height, src.duration_s) if src else (None, None, None) + + if not src_ready or not src_rel: + _fail_edit(job_id, asset_id, "The source download is no longer available.") + return + src_path = storage.abs_path(settings.download_root, src_rel) + if not src_path.exists(): + _fail_edit(job_id, asset_id, "The source file is missing.") + return + + src_ext = src_path.suffix.lstrip(".").lower() or "mp4" + crop = editmod.has_crop(edit_spec) + out_ext = "mp4" if editmod.needs_reencode(edit_spec) else src_ext + total_s = editmod.clip_duration(edit_spec, src_dur) or src_dur or 0 + staging = _edit_staging_dir(asset_id) + + try: + if staging.exists(): + shutil.rmtree(staging, ignore_errors=True) + staging.mkdir(parents=True, exist_ok=True) + dest_staging = staging / f"out.{out_ext}" + + _set_job(job_id, phase="editing", progress=0, speed_bps=None, eta_s=None) + cmd = editmod.build_edit_cmd(src_path, dest_staging, edit_spec, out_ext) + editmod.run_ffmpeg( + cmd, + total_s, + on_progress=lambda pct: _set_job(job_id, progress=int(pct), phase="editing"), + should_cancel=lambda: _job_status(job_id) not in ("running", None), + ) + if not dest_staging.exists(): + raise RuntimeError("ffmpeg produced no output file") + + rel = storage.edit_rel_path(user_id, asset_id, out_ext) + storage.place_file(dest_staging, settings.download_root, rel) + final = storage.abs_path(settings.download_root, rel) + size = final.stat().st_size if final.exists() else None + expires = datetime.now(timezone.utc) + timedelta(days=settings.download_retention_days) + + with SessionLocal() as db: + asset = db.get(MediaAsset, asset_id) + if asset: + asset.status = "ready" + asset.rel_path = rel + asset.size_bytes = size + asset.container = out_ext + asset.duration_s = total_s or asset.duration_s + if crop: + asset.width = edit_spec["crop"]["w"] + asset.height = edit_spec["crop"]["h"] + else: + asset.width, asset.height = src_w, src_h + asset.error = None + asset.last_access_at = datetime.now(timezone.utc) + asset.expires_at = expires + job = db.get(DownloadJob, job_id) + if job: + job.status = "done" + job.progress = 100 + job.phase = None + db.commit() + log.info("edit job %s done: %s", job_id, rel) + + except editmod.EditAborted: + # Paused/canceled mid-encode: free the asset for a later retry (ref_count is the route's job). + _reset_asset_pending(asset_id) + log.info("edit job %s aborted (pause/cancel)", job_id) + except Exception as exc: # noqa: BLE001 — surface the failure, keep the worker alive + _fail_edit(job_id, asset_id, str(exc)[:500]) + finally: + shutil.rmtree(staging, ignore_errors=True) + + # --- loop ----------------------------------------------------------------------------------- def _worker_loop(idx: int) -> None: