feat(downloads): ad-hoc clips cut straight from a YouTube video or a Plex item

POST /api/downloads/clip takes a source (a YouTube id or a Plex rating key), a range of at most
ten minutes and, for Plex, the audio track being heard. The clip becomes a per-user asset in the
library like an editor clip, shareable as soon as it is ready. A YouTube clip fetches only its
range (yt-dlp download_ranges: 3-7 s whatever the video's length); a Plex clip reads the library
file in place. Both then go through one re-encode: frame-exact at the marks, H.264 at most 1080p
with HDR tonemapped to SDR, stereo AAC, no chapter track, .mp4.

Job kinds are a JobKind StrEnum now, and Re-download refuses every kind but a download.

backend/app/downloads/pipeline.py

@@ -7,10 +7,13 @@ from datetime import date, datetime, timedelta, timezone
 from pathlib import Path
 
 import yt_dlp
+from sqlalchemy import select
+from yt_dlp.utils import download_range_func
 
 from app import logs, sysconfig
 from app.config import settings
 from app.db import SessionLocal
+from app.downloads import clip as clipmod
 from app.downloads import edit as editmod
 from app.downloads import errors as dlerrors
 from app.downloads import adapters, credentials, formats, mediakind, playable, probe, quota
@@ -26,7 +29,8 @@ from app.downloads.jobdb import (
 )
 from app.downloads.stated import StatedFailure
 from app.downloads.status import JobStatus
-from app.models import DISPLAY_TEXT_MAX, DownloadJob, MediaAsset
+from app.models import DISPLAY_TEXT_MAX, DownloadJob, MediaAsset, PlexItem
+from app.plex import paths as plex_paths
 from app.notifications import create_notification
 from app.titles import normalize_title
 
@@ -333,7 +337,7 @@ def _media_info(info: dict) -> dict:
 
 
 def _extract_with_fallback(
-    job_id, asset_id, url, spec, staging, adapter, cookiefile=None, max_bytes=None
+    job_id, asset_id, url, spec, staging, adapter, cookiefile=None, max_bytes=None, section=None
 ):
     """Download via yt-dlp using the source `adapter`'s client sets + options. For YouTube the adapter
     supplies the primary player-client set then a POT-backed fallback (retried on a bot/DRM/format
@@ -349,7 +353,11 @@ def _extract_with_fallback(
     `cookiefile` is the acting user's per-host cookies.txt, already gated by
     `credentials.resolve_cookiefile`; passed to every attempt so the fallback stays authenticated.
 
-    `max_bytes` (SEC-13) is the owner's remaining storage budget, applied to every attempt."""
+    `max_bytes` (SEC-13) is the owner's remaining storage budget, applied to every attempt.
+
+    `section` = (start_s, end_s) fetches only that range (an ad-hoc clip, see downloads/clip): the
+    range is stream-copied, and its first frame is the marked one (measured), so the clip's own
+    re-encode starts at 0 instead of re-fetching keyframes at the cut."""
     client_sets = adapter.client_sets(authed=cookiefile is not None)
     outtmpl = str(staging / "%(id)s.%(ext)s")
     # Resolving may read a page first — network time the stall watchdog must not see as a stall.
@@ -375,6 +383,9 @@ def _extract_with_fallback(
             max_bytes=max_bytes,
         )
         opts["postprocessor_hooks"] = [_make_pp_hook(job_id)]
+        if section is not None:
+            opts["download_ranges"] = download_range_func([], [section])
+            opts["force_keyframes_at_cuts"] = False
         _touch_job(job_id)  # stall-watchdog heartbeat: extraction writes no progress, so a slow /
         # multi-client-set fallback would otherwise cross the stall timeout with a stale updated_at.
         # (Per-ATTEMPT only; a single extract_info that itself runs past the stall timeout is still
@@ -896,6 +907,159 @@ def _process_edit(job_id: int, asset_id: int) -> None:
         shutil.rmtree(staging, ignore_errors=True)
 
 
+def _clip_source(job_id: int, asset_id: int, spec: dict, staging: Path) -> tuple[Path, float]:
+    """The file a clip is cut from and where in it the clip starts. A YouTube clip fetches ONLY its
+    range into `staging` (so it starts at 0 there); a Plex clip reads the library file in place."""
+    if spec["source"] == "youtube":
+        with SessionLocal() as db:
+            job = db.get(DownloadJob, job_id)
+            remaining = quota.remaining_bytes(db, job.user_id) if job else None
+        if remaining == 0:  # see `_download`: zero is a refusal, not "no limit"
+            raise _OwnerRefused(dlerrors.QUOTA_FULL_MESSAGE)
+        _set_job(job_id, phase="downloading", progress=0)
+        # Signed out, like a download without a cookie: the cookie gate is written for a download's
+        # salted asset identity (`credentials.resolve_cookiefile`), not for a clip's.
+        info = _extract_with_fallback(
+            job_id,
+            asset_id,
+            service.source_url("youtube", spec["ref"]),
+            formats.normalize(clipmod.YOUTUBE_FETCH_SPEC),
+            staging,
+            adapters.resolve_adapter("youtube", spec["ref"]),
+            max_bytes=remaining,
+            section=(spec["start_s"], spec["end_s"]),
+        )
+        return _produced_file(_media_info(dict(info)), staging), 0.0
+    with SessionLocal() as db:
+        item = db.execute(
+            select(PlexItem).where(PlexItem.rating_key == spec["ref"])
+        ).scalar_one_or_none()
+        src = plex_paths.local_media_path(db, item.file_path) if item else None
+    if src is None:
+        raise StatedFailure("The library file is not reachable from here.")
+    return src, spec["start_s"]
+
+
+def _process_clip(job_id: int, asset_id: int) -> None:
+    """Cut an ad-hoc clip (downloads/clip): fetch or open its source, ONE re-encode to the clip
+    shape, then into the user's library beside their edits."""
+    with SessionLocal() as db:
+        job = db.get(DownloadJob, job_id)
+        if job is None:
+            return
+        spec = dict(job.edit_spec or {})
+        user_id = job.user_id
+    staging = _edit_staging_dir(asset_id)
+    try:
+        if staging.exists():
+            shutil.rmtree(staging, ignore_errors=True)
+        staging.mkdir(parents=True, exist_ok=True)
+        src, start = _clip_source(job_id, asset_id, spec, staging)
+        data = probe.probe_json(src) or {}
+        audio_track = spec.get("audio_track")
+        if audio_track is not None and audio_track >= clipmod.audio_track_count(data):
+            raise StatedFailure("The chosen audio track is not in the file.")
+        total_s = clipmod.duration(spec)
+        out = staging / "clip.mp4"
+        _set_job(job_id, phase="editing", progress=0, speed_bps=None, eta_s=None)
+        editmod.run_ffmpeg(
+            clipmod.build_cmd(
+                src,
+                out,
+                start,
+                total_s,
+                audio_track=audio_track,
+                hdr=clipmod.is_hdr(data),
+                threads=editmod.encode_thread_cap(),
+            ),
+            total_s,
+            on_progress=_ffmpeg_progress_writer(job_id, "editing"),
+            should_cancel=_cancel_poll(job_id),
+        )
+        if not out.exists():
+            raise StatedFailure("ffmpeg produced no output file")
+        # The phone-safe shape by construction; still held to the rule every ready file meets.
+        out = _ensure_phone_safe(job_id, out, staging, "editing", round(total_s))
+        ext = out.suffix.lstrip(".").lower() or "mp4"
+        rel = storage.edit_rel_path(user_id, asset_id, ext)
+        storage.place_file(out, settings.download_root, rel)
+        final = storage.abs_path(settings.download_root, rel)
+        vcodec, acodec, width, height, _ = probe.probe_media(final)
+        _, retention_days = _download_settings()
+
+        with SessionLocal() as db:
+            asset = db.get(MediaAsset, asset_id)
+            if asset:
+                asset.status = "ready"
+                asset.rel_path = rel
+                asset.size_bytes = final.stat().st_size if final.exists() else None
+                asset.container = ext
+                asset.vcodec, asset.acodec, asset.width, asset.height = (
+                    vcodec,
+                    acodec,
+                    width,
+                    height,
+                )
+                asset.duration_s = round(total_s)
+                asset.media_kind = mediakind.derive(vcodec, acodec)
+                # The clip's own frame as its poster: the link preview shows the moment, and a Plex
+                # item has no public thumbnail at all.
+                asset.poster_path = editmod.ensure_poster(asset, settings.download_root)
+                asset.error = None
+                asset.last_access_at = datetime.now(timezone.utc)
+                asset.expires_at = datetime.now(timezone.utc) + timedelta(days=retention_days)
+            title = asset.title if asset else None
+            finalized = service.finalize_if_running(
+                db, job_id, status=JobStatus.DONE, progress=100, phase=None
+            )
+            db.commit()
+            if finalized:
+                _notify_download_done(db, user_id, title, job_id, kind="edit")
+                log.info(
+                    "clip job %s done: asset %s",
+                    job_id,
+                    asset_id,
+                    extra=logs.event(
+                        "clip.job.done", job_id=job_id, asset_id=asset_id, source=spec["source"]
+                    ),
+                )
+    except (_Aborted, editmod.EditAborted):
+        # Paused/canceled mid-fetch or mid-encode: free the asset for a later retry.
+        _reset_asset_pending(asset_id)
+        log.info(
+            "clip job %s aborted (pause/cancel)",
+            job_id,
+            extra=logs.event("clip.job.aborted", job_id=job_id),
+        )
+    except _OwnerRefused as exc:
+        # Quota full: the job fails, the asset goes back to pending for a Retry once there is room
+        # (the download's handling; see `_download`).
+        msg = str(exc)[:500]
+        with SessionLocal() as db:
+            service.finalize_if_running(
+                db,
+                job_id,
+                status=JobStatus.ERROR,
+                error=msg,
+                error_code=dlerrors.classify(msg),
+                error_detail=None,
+                phase=None,
+            )
+            db.commit()
+        _reset_asset_pending(asset_id)
+        log.info(
+            "clip job %s refused for its owner: %s",
+            job_id,
+            formats.log_safe(msg),
+            extra=logs.event("clip.job.refused", job_id=job_id),
+        )
+    except Exception as exc:  # noqa: BLE001 — surface the failure, keep the worker alive
+        # silent-ok: not silent — `_fail_edit` is the report, and it takes `exc` for its stack.
+        _fail_edit(job_id, asset_id, str(exc)[:500], exc, what="clip")
+    finally:
+        shutil.rmtree(staging, ignore_errors=True)
+
+
 def _check_converted_fits(owner_id: int | None, old_size: int, new_size: int) -> None:
     """Refuse a converted upload that would push its owner over the storage cap: the upload was
     charged at its original size, and a re-encode can come out larger. The original stays, counted,