diff --git a/backend/app/config.py b/backend/app/config.py index 8ef87cd..8a41c02 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -205,6 +205,11 @@ class Settings(BaseSettings): # Max concurrent transcodes for the P3 fallback. Low by default — CPU-only transcode is # expensive; direct-serve (browser-compatible files) has no such limit. plex_max_transcodes: int = 1 + # Where on-the-fly HLS remux segments are written (transient, reaped). Empty → a `.plex-hls` + # dir under download_root. On localdev the download_root is a slow Windows bind-mount, so point + # this at a fast container-local path (e.g. /var/tmp/plex-hls); prod's download_root is fast + # local disk, so the default is fine there. + plex_hls_dir: str = "" @property def allowed_email_set(self) -> set[str]: diff --git a/backend/app/plex/stream.py b/backend/app/plex/stream.py new file mode 100644 index 0000000..e45c826 --- /dev/null +++ b/backend/app/plex/stream.py @@ -0,0 +1,163 @@ +"""On-the-fly HLS remux for Plex playback (seek-restart session model). + +Playback is from the LOCAL physical file. Browser-compatible files (playable="direct") are served +raw with HTTP range requests. Everything else that is h264 video (playable="remux") is remuxed on +the fly to HLS with **video stream-copy** (cheap, I/O-bound — no video re-encode) and audio to AAC +when needed. HEVC/VP9 (playable="transcode") needs a full re-encode and is deferred to P3. + +Seek model (Jellyfin-style): one ffmpeg session per item, started at a given offset via `-ss` +(fast keyframe seek). A seek beyond the generated region restarts the session at the new offset, +so seeking is responsive even on long movies without pre-generating the whole file. ffmpeg's own +HLS muxer does the segmentation (the only reliable way to cut a stream-copy at keyframes). + +Sessions live in this (single) API process; a periodic reaper kills idle ones and frees the temp +segments. The remux is CPU-light (video copy + a tiny audio transcode), so it runs fine on the +CPU-only prod host; only P3's full transcode is CPU-heavy. +""" +import logging +import shutil +import subprocess +import threading +import time +from pathlib import Path + +from sqlalchemy.orm import Session as DbSession + +from app.config import settings +from app.models import PlexItem +from app.plex import paths + +log = logging.getLogger("siftlode.plex") + +_HLS_ROOT = Path(settings.plex_hls_dir) if settings.plex_hls_dir else Path(settings.download_root) / ".plex-hls" +_SEG_SECONDS = 6 +_MAX_SESSIONS = 4 # concurrent remux sessions (safety valve for the CPU-only host) +_SESSION_IDLE_S = 600 # reap a session with no access for this long + +_lock = threading.Lock() +_sessions: dict[str, "HlsSession"] = {} + + +class HlsSession: + def __init__(self, key: str, directory: Path, proc: subprocess.Popen, start_s: float): + self.key = key + self.dir = directory + self.proc = proc + self.start_s = start_s + self.last_access = time.time() + + +def _acodec(item: PlexItem) -> str: + # Copy the audio when it's already AAC; otherwise transcode (ac3/dts/… → aac). Video is always + # copied for the remux class. + return "copy" if (item.codec_audio or "").lower() == "aac" else "aac" + + +def _ffmpeg_cmd(src: Path, start_s: float, out_dir: Path, acodec: str) -> list[str]: + args = ["ffmpeg", "-nostdin", "-loglevel", "error"] + if start_s > 0: + args += ["-ss", f"{start_s:.3f}"] # fast keyframe seek before -i + # Map only the first video + first audio; drop subtitles so ffmpeg's HLS muxer doesn't spawn + # WebVTT renditions (subtitle/multi-audio selection is a later refinement served separately). + args += ["-i", str(src), "-map", "0:v:0", "-map", "0:a:0?", "-sn", "-c:v", "copy", "-c:a", acodec] + if acodec == "aac": + args += ["-ac", "2", "-b:a", "192k"] + args += [ + "-f", "hls", + "-hls_time", str(_SEG_SECONDS), + "-hls_list_size", "0", + # EVENT (not VOD): ffmpeg writes/appends the playlist AS segments complete, so playback can + # start immediately from this session's offset. VOD only writes the playlist at the end. The + # full seekbar comes from our known duration + the seek-restart model, not from the playlist. + "-hls_playlist_type", "event", + "-hls_segment_type", "mpegts", + "-hls_flags", "independent_segments+temp_file", + "-hls_segment_filename", str(out_dir / "seg_%d.ts"), + str(out_dir / "index.m3u8"), + ] + return args + + +def _kill(s: HlsSession) -> None: + try: + s.proc.terminate() + try: + s.proc.wait(timeout=3) + except subprocess.TimeoutExpired: + s.proc.kill() + except Exception: + pass + shutil.rmtree(s.dir, ignore_errors=True) + + +def _enforce_cap() -> None: + # Caller holds _lock. Drop the least-recently-accessed sessions over the cap. + if len(_sessions) <= _MAX_SESSIONS: + return + for key, s in sorted(_sessions.items(), key=lambda kv: kv[1].last_access)[: len(_sessions) - _MAX_SESSIONS]: + _kill(s) + _sessions.pop(key, None) + + +def start_session(db: DbSession, item: PlexItem, start_s: float) -> HlsSession | None: + """(Re)start the HLS remux for an item at the given offset. Returns None if the local file + can't be read.""" + src = paths.local_media_path(db, item.file_path) + if src is None: + return None + key = item.rating_key + start_s = max(0.0, float(start_s)) + acodec = _acodec(item) + with _lock: + old = _sessions.pop(key, None) + if old is not None: + _kill(old) + directory = _HLS_ROOT / f"{key}_{int(start_s)}" + shutil.rmtree(directory, ignore_errors=True) + directory.mkdir(parents=True, exist_ok=True) + proc = subprocess.Popen( + _ffmpeg_cmd(src, start_s, directory, acodec), + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + s = HlsSession(key, directory, proc, start_s) + _sessions[key] = s + _enforce_cap() + log.info("plex hls session start key=%s start=%.1f acodec=%s", key, start_s, acodec) + return s + + +def current_session(key: str) -> HlsSession | None: + with _lock: + s = _sessions.get(key) + if s is not None: + s.last_access = time.time() + return s + + +def wait_for(path: Path, timeout: float = 20.0) -> bool: + """Wait until a session file (playlist / segment) exists and is non-empty. Segments are + produced ~faster than realtime, so a segment just ahead of playback appears quickly; a segment + far beyond the generated region won't (the frontend restarts the session at a seek instead).""" + end = time.time() + timeout + while time.time() < end: + try: + if path.exists() and path.stat().st_size > 0: + return True + except OSError: + pass + time.sleep(0.15) + return path.exists() + + +def reap_idle() -> int: + now = time.time() + dropped = 0 + with _lock: + for key, s in list(_sessions.items()): + done = s.proc.poll() is not None + if now - s.last_access > _SESSION_IDLE_S or (done and now - s.last_access > 30): + _kill(s) + _sessions.pop(key, None) + dropped += 1 + return dropped diff --git a/backend/app/routes/plex.py b/backend/app/routes/plex.py index 6b870e7..7accc91 100644 --- a/backend/app/routes/plex.py +++ b/backend/app/routes/plex.py @@ -8,6 +8,7 @@ catalog sync. import logging from fastapi import APIRouter, Depends, HTTPException, Query, Response +from fastapi.responses import FileResponse from sqlalchemy import and_, func, or_ from sqlalchemy.orm import Session, aliased @@ -15,6 +16,8 @@ from app import sysconfig from app.auth import current_user from app.db import get_db from app.models import PlexItem, PlexLibrary, PlexSeason, PlexShow, PlexState, User +from app.plex import paths as plex_paths +from app.plex import stream as plex_stream from app.plex import sync as plex_sync from app.plex.client import PlexClient, PlexError, PlexNotConfigured from app.routes.admin import admin_user @@ -279,3 +282,92 @@ def image( except PlexError as e: raise HTTPException(status_code=502, detail=f"Plex image fetch failed: {e}") return Response(content=data, media_type=ct, headers={"Cache-Control": "public, max-age=86400"}) + + +# --- Playback (local file → direct 206 or on-the-fly HLS remux; P2) ---------------------------- + + +def _item_or_404(db: Session, rating_key: str) -> PlexItem: + it = db.query(PlexItem).filter_by(rating_key=str(rating_key)).first() + if it is None: + raise HTTPException(status_code=404, detail="Unknown Plex item") + return it + + +@router.post("/stream/{rating_key}/session") +def stream_session( + rating_key: str, + start: float = 0.0, + user: User = Depends(current_user), + db: Session = Depends(get_db), +) -> dict: + """Prepare playback from `start` seconds. Direct-playable files are served raw; remux files + (re)start an HLS session at the offset (seek-restart); transcode (HEVC/VP9) is P3.""" + plex_stream.reap_idle() # opportunistic cleanup of idle sessions on each new playback + it = _item_or_404(db, rating_key) + if it.playable == "direct": + return {"mode": "direct", "url": f"/api/plex/stream/{it.rating_key}/file", "start_s": 0.0} + if it.playable == "transcode": + raise HTTPException(status_code=501, detail="This format needs transcoding (a later phase).") + if plex_paths.local_media_path(db, it.file_path) is None: + raise HTTPException(status_code=404, detail="Media file not found on disk") + s = plex_stream.start_session(db, it, start) + if s is None: + raise HTTPException(status_code=404, detail="Media file not found on disk") + return {"mode": "hls", "start_s": s.start_s, "url": f"/api/plex/stream/{it.rating_key}/index.m3u8"} + + +@router.get("/stream/{rating_key}/file") +def stream_file( + rating_key: str, + user: User = Depends(current_user), + db: Session = Depends(get_db), +) -> FileResponse: + """Serve the raw local media file with HTTP range support (direct-playable files).""" + it = _item_or_404(db, rating_key) + src = plex_paths.local_media_path(db, it.file_path) + if src is None: + raise HTTPException(status_code=404, detail="Media file not found on disk") + return FileResponse(str(src)) # Starlette honours the Range header (206) + + +@router.get("/stream/{rating_key}/index.m3u8") +def stream_playlist( + rating_key: str, + user: User = Depends(current_user), + db: Session = Depends(get_db), +) -> Response: + """The current HLS session's playlist (starts a session at 0 if none is running).""" + s = plex_stream.current_session(str(rating_key)) + if s is None: + it = _item_or_404(db, rating_key) + if it.playable not in ("remux",): + raise HTTPException(status_code=400, detail="Not an HLS-remux item") + s = plex_stream.start_session(db, it, 0.0) + if s is None: + raise HTTPException(status_code=404, detail="Media file not found on disk") + if not plex_stream.wait_for(s.dir / "index.m3u8", 30): + raise HTTPException(status_code=503, detail="Playlist not ready") + return Response( + content=(s.dir / "index.m3u8").read_bytes(), + media_type="application/vnd.apple.mpegurl", + headers={"Cache-Control": "no-store"}, + ) + + +@router.get("/stream/{rating_key}/seg_{n}.ts") +def stream_segment( + rating_key: str, + n: int, + user: User = Depends(current_user), +) -> Response: + """Serve one HLS segment from the current session (waits briefly if it's just ahead).""" + s = plex_stream.current_session(str(rating_key)) + if s is None: + raise HTTPException(status_code=404, detail="No active playback session") + seg = s.dir / f"seg_{n}.ts" + if not plex_stream.wait_for(seg, 20): + raise HTTPException(status_code=404, detail="Segment not available") + return Response( + content=seg.read_bytes(), media_type="video/mp2t", headers={"Cache-Control": "no-store"} + ) diff --git a/docker-compose.localdev.yml b/docker-compose.localdev.yml index 0822086..6962167 100644 --- a/docker-compose.localdev.yml +++ b/docker-compose.localdev.yml @@ -46,6 +46,9 @@ services: # Download center: the API serves finished files (it does NOT run the worker loop). DOWNLOAD_ROOT: /downloads WORKER_ENABLED: "false" + # Plex HLS remux scratch on fast container-local storage (the /downloads bind-mount is a slow + # Windows Docker Desktop mount in dev; prod uses fast local disk so it keeps the default). + PLEX_HLS_DIR: /var/tmp/plex-hls depends_on: db: condition: service_healthy