"""Plex playback: image/subtitle proxies, HLS streaming, and watch-state writes.""" import hashlib import os import tempfile import threading from datetime import datetime, timedelta, timezone import httpx from fastapi import ( APIRouter, BackgroundTasks, Depends, HTTPException, Request, Response, ) from fastapi.responses import FileResponse from pydantic import BaseModel from sqlalchemy import select from sqlalchemy.orm import Session from app.auth import current_user, resolved_user_id from app.db import SessionLocal, get_db from app.models import ( PlexItem, PlexSeason, PlexShow, PlexState, User, ) from app.plex import imgcache, subtitles from app.plex import paths as plex_paths from app.plex.lookups import by_rating_key from app.plex import people as plex_people from app.plex import stream as plex_stream from app.plex import watch_sync as plex_watch from app.plex.client import PlexClient, PlexError, PlexNotConfigured from app.routes.plex._shared import item_or_404 router = APIRouter(prefix="/api/plex", tags=["plex"]) def _require_session(request: Request) -> int: """Lightweight auth for the high-fan-out image proxies: validate the signed session cookie WITHOUT a DB user-load, so these endpoints never hold a pool connection for the auth. Poster images are low-sensitivity; a valid session is enough.""" uid, _ = resolved_user_id(request) if not uid: raise HTTPException(status_code=401, detail="Not authenticated") return uid @router.get("/image/{rating_key}") def image(rating_key: str, request: Request, variant: str = "thumb") -> Response: """Proxy a Plex poster/art image (keeps the admin token server-side; no image duplication). A thin on-disk cache serves repeat views without re-hitting Plex. ⚠️ This endpoint is called in BURSTS (a collections page fires 100+ concurrent poster requests), so it must NOT hold a DB connection during the slow Plex fetch — doing so exhausted the pool (QueuePool timeout, broken images on prod). Warm path (cache hit) touches NO DB; cold path opens a brief session only to resolve the image key + read the Plex config, then RELEASES it before the fetch. Auth is the signed-session check (no DB user-load).""" _require_session(request) v = "art" if variant == "art" else "thumb" w, h = imgcache.SIZES[v] # Size is part of the cache key: Plex resizes server-side (see PlexClient.image_bytes), so we cache # the small variant. The width tag also means changing a target size just re-caches, never serves a # stale differently-sized file (old full-size files under the un-tagged key are simply orphaned). cache_key = f"item_{rating_key}_{v}_{w}" hit = imgcache.read(cache_key) if hit: return imgcache.response(*hit) # Cold miss: short-lived session for the key + Plex client config, released before the slow fetch. try: with SessionLocal() as db: key = imgcache.key_for(db, str(rating_key), v) if not key: raise HTTPException(status_code=404, detail="No image") plex = PlexClient(db) # reads base URL + token now; the fetch below needs no DB except PlexNotConfigured as e: raise HTTPException(status_code=400, detail=str(e)) try: with plex: data, ct = plex.image_bytes(key, w, h) except PlexError as e: raise HTTPException(status_code=502, detail=f"Plex image fetch failed: {e}") imgcache.write(cache_key, data, ct) return imgcache.response(data, ct) @router.get("/subtitle/{rating_key}/{sub_ord}") def subtitle(rating_key: str, sub_ord: int, request: Request, offset: float = 0.0) -> Response: """Serve the ord-th subtitle track as WebVTT for the browser's — so choosing a subtitle overlays it directly (no HLS re-mux / session restart). Handles EXTERNAL sidecar subs (fetched from Plex via the stream `key` — the old `-map 0:s` broke on these since they're not in the file) AND embedded text subs (extracted from the local file). Image subs (PGS/VobSub) → 415. Disk-cached.""" _require_session(request) # Cache the ABSOLUTE-time VTT once (per rk+ord); the per-session cue-shift is applied on the way # out, so a resume/seek to any offset reuses the same cached track (no cache-per-offset explosion). cache_key = f"sub_{rating_key}_{sub_ord}" hit = subtitles.cache_read(cache_key) if hit is not None: return Response(subtitles.shift_vtt(hit, offset), media_type="text/vtt") try: with SessionLocal() as db: it = item_or_404(db, rating_key) with PlexClient(db) as plex: part = ((plex.metadata(it.rating_key) or {}).get("Media") or [{}])[0].get( "Part" ) or [{}] subs = [s for s in (part[0].get("Stream") or []) if s.get("streamType") == 3] if sub_ord < 0 or sub_ord >= len(subs): raise HTTPException(status_code=404, detail="No such subtitle") s = subs[sub_ord] codec = (s.get("codec") or s.get("format") or "").lower() if codec in subtitles.BITMAP_SUBS: raise HTTPException( status_code=415, detail="Image-based subtitle can't be shown as text", ) if s.get("key"): # external sidecar → fetch from Plex, convert to VTT raw = plex.raw_get(s["key"]) if codec in ("srt", "subrip", ""): vtt = subtitles.srt_to_vtt(raw.decode("utf-8", "replace")) else: fd, tmp = tempfile.mkstemp(suffix="." + (codec or "sub")) try: os.write(fd, raw) os.close(fd) vtt = subtitles.ffmpeg_to_vtt(tmp, None) finally: os.unlink(tmp) elif s.get("index") is not None: # embedded → extract from the local file 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") emb_ord = sum(1 for x in subs[:sub_ord] if x.get("index") is not None) vtt = subtitles.ffmpeg_to_vtt(str(src), emb_ord) else: raise HTTPException(status_code=404, detail="Subtitle not fetchable") except PlexNotConfigured as e: raise HTTPException(status_code=400, detail=str(e)) except PlexError as e: raise HTTPException(status_code=502, detail=f"Subtitle fetch failed: {e}") subtitles.cache_write(cache_key, vtt) return Response(subtitles.shift_vtt(vtt, offset), media_type="text/vtt") # Resume/watched thresholds. _PROGRESS_MIN_S matches the feed's PROGRESS_MIN_SECONDS (5), but # _FINISH_MARGIN_S is DELIBERATELY larger than the feed's FINISH_MARGIN_SECONDS (30 vs 10): this is # long-form content (movies/episodes with credits) and crossing it MARKS THE ITEM WATCHED, whereas # the feed's margin only clears a short video's resume bar — different domains, intentional divergence, # not drift. Keep them separate on purpose (see routes/feed.py). _PROGRESS_MIN_S = 5 _FINISH_MARGIN_S = 30 # S2 write-back throttle: the resume position is now pushed to Plex on the periodic (non-final) # checkpoint too — not only on the final pause/unmount flush — so Plex (the source of truth) stays # current in near-real-time. To keep a single watch from spraying `/:/timeline` calls, a push is # rate-limited to at most one per item per user every _PUSH_THROTTLE_S seconds; a `final` checkpoint # (pause / seek / stop) always pushes immediately and resets the window. This is an in-memory, single- # process rate-limiter (the deployment is one FastAPI process): after a restart the map is empty, so at # most one extra push happens on the first checkpoint — harmless. It is NOT durable domain state. _PUSH_THROTTLE_S = 10 _last_push_at: dict[tuple[int, int], datetime] = {} _push_throttle_lock = threading.Lock() def _should_push_resume(user_id: int, item_id: int, now: datetime, final: bool) -> bool: """Whether to mirror this resume checkpoint to Plex now. `final` always pushes; a periodic tick pushes only when ≥_PUSH_THROTTLE_S has elapsed since the last push for this (user, item). Records `now` as the last-push time whenever it returns True so the window is measured from the real push (and so `_resume_push_superseded` can drop an older, still-pending push — see there).""" with _push_throttle_lock: last = _last_push_at.get((user_id, item_id)) if final or last is None or (now - last).total_seconds() >= _PUSH_THROTTLE_S: _last_push_at[(user_id, item_id)] = now # Opportunistic eviction so the map can't grow unbounded over a long-lived process (one # entry per item ever checkpointed). Anything older than a few windows is dead — a push # completes in seconds, so a stale entry protects nothing. if len(_last_push_at) > 2048: cutoff = now - timedelta(seconds=_PUSH_THROTTLE_S * 6) for k in [k for k, t in _last_push_at.items() if t < cutoff]: del _last_push_at[k] return True return False def _resume_push_superseded(user_id: int, item_id: int, intent: datetime) -> bool: """True if a newer resume push for this (user, item) was scheduled after `intent`. Called at the HTTP boundary of a background resume push: because periodic and seek/pause pushes are independent background tasks with no ordering guard, an older still-pending periodic push could otherwise land at Plex AFTER a newer backward-seek push and — with the Plex-authoritative pull — durably revive the stale position. Skipping a superseded push keeps the last-scheduled position the one Plex settles on.""" with _push_throttle_lock: return _last_push_at.get((user_id, item_id), intent) > intent @router.get("/person-image") def person_image(u: str, request: Request) -> Response: """Proxy a cast/crew photo from Plex's public metadata CDN (host-whitelisted; no token needed). Signed-session auth (no DB) so it, like /image, never holds a pool connection during the fetch.""" _require_session(request) if not u.startswith(plex_people.PERSON_IMG_HOST): raise HTTPException(status_code=400, detail="Unsupported image host") cache_key = "person_" + hashlib.sha1(u.encode()).hexdigest() hit = imgcache.read(cache_key) if hit: return imgcache.response(*hit) try: with httpx.Client(timeout=10.0, follow_redirects=True) as c: r = c.get(u) r.raise_for_status() except httpx.HTTPError: raise HTTPException(status_code=502, detail="Image fetch failed") ct = r.headers.get("content-type", "image/jpeg") imgcache.write(cache_key, r.content, ct) return imgcache.response(r.content, ct) def _push_resume_guarded( user_id: int, item_id: int, rating_key: str, time_ms: int, duration_ms: int, expect_ts: datetime | None, intent: datetime, ) -> None: """Background entrypoint for a resume push that drops itself if a newer resume push for the same (user, item) was scheduled after it (see `_resume_push_superseded`), then delegates to the real push. This narrows the periodic-vs-seek write-ordering race to the (rare) case where the older push has already left the process before the newer one is scheduled.""" if _resume_push_superseded(user_id, item_id, intent): return plex_watch.push_state_to_plex( user_id, item_id, rating_key, "resume", time_ms, duration_ms, expect_ts ) def _push_watch( background: BackgroundTasks, db: Session, user_id: int, item_id: int, rating_key: str, action: str, time_ms: int = 0, duration_ms: int = 0, intent: datetime | None = None, ) -> None: """Schedule a best-effort Siftlode→Plex watch push, but only for a user whose owner link has push enabled — non-owner users never touch Plex. Guarding here (with the request session) avoids spinning up a throwaway background session on every state change for the common no-link case. A `resume` push with an `intent` timestamp is routed through `_push_resume_guarded` so a superseded (older) push can't land out of order; `watched`/`unwatched` scrobbles go straight through.""" if plex_watch.link_for_push(db, user_id) is None: return # Capture the row's freshest timestamp NOW so the background push can tell if a newer local edit # lands before it settles synced_to_plex (which would otherwise strand that newer value). st = ( db.execute( select(PlexState).where(PlexState.user_id == user_id, PlexState.item_id == item_id) ) .scalars() .first() ) marker = plex_watch._state_marker(st) if action == "resume" and intent is not None: background.add_task( _push_resume_guarded, user_id, item_id, rating_key, time_ms, duration_ms, marker, intent ) return background.add_task( plex_watch.push_state_to_plex, user_id, item_id, rating_key, action, time_ms, duration_ms, marker, ) class ItemProgressIn(BaseModel): # float, truncated in the handler — the player may post a raw currentTime (a float); a plain # `int` field would 422 it on the periodic progress checkpoint (the old `int(...)` truncated). position_seconds: float = 0 duration_seconds: float = 0 final: bool = False @router.post("/item/{rating_key}/progress") def item_progress( rating_key: str, body: ItemProgressIn, background: BackgroundTasks, user: User = Depends(current_user), db: Session = Depends(get_db), ) -> dict: """Checkpoint the resume position while playing (near the end → mark watched + clear it). S2 write-back: the resume position is mirrored to Plex (the source of truth) on the periodic checkpoint too, not only on the final flush — throttled to ≥_PUSH_THROTTLE_S per item (see `_should_push_resume`). `final` (sent on pause / seek / pagehide / unmount) pushes immediately and resets the throttle window, so Plex sees a settled position the instant the user pauses or seeks.""" it = item_or_404(db, rating_key) position = max(0, int(body.position_seconds)) duration = int(body.duration_seconds) final = body.final now = datetime.now(timezone.utc) st = ( db.execute( select(PlexState).where(PlexState.user_id == user.id, PlexState.item_id == it.id) ) .scalars() .first() ) was_watched = st is not None and st.status == "watched" near_end = duration > 0 and position > duration - _FINISH_MARGIN_S if near_end: if st is None: st = PlexState(user_id=user.id, item_id=it.id) db.add(st) st.status = "watched" st.watched_at = now st.position_seconds = 0 st.progress_updated_at = now st.synced_to_plex = False # Terminal state for resume write-back: drop the throttle/ordering entry so the map doesn't # retain a dead key for every finished item. with _push_throttle_lock: _last_push_at.pop((user.id, it.id), None) db.commit() # Immediate on the watched transition; skip a repeat scrobble if it was already watched # (a later checkpoint of the same finished item would otherwise bump Plex's viewCount again). if not was_watched: _push_watch(background, db, user.id, it.id, rating_key, "watched") return {"status": "watched", "position_seconds": 0} if position < _PROGRESS_MIN_S: if st is not None and st.status == "new": db.delete(st) elif st is not None: st.position_seconds = 0 st.progress_updated_at = now db.commit() return {"position_seconds": 0} if st is None: st = PlexState(user_id=user.id, item_id=it.id) db.add(st) st.position_seconds = position st.progress_updated_at = now push_now = _should_push_resume(user.id, it.id, now, final) if push_now: # Mark dirty before the (background) push so the reconcile knows to re-push if it fails, and so # `push_state_to_plex` can settle synced_to_plex once Plex confirms. st.synced_to_plex = False db.commit() if push_now: _push_watch( background, db, user.id, it.id, rating_key, "resume", position * 1000, duration * 1000, intent=now, ) return {"position_seconds": position} class ItemStateIn(BaseModel): status: str = "" @router.post("/item/{rating_key}/state") def item_state( rating_key: str, body: ItemStateIn, background: BackgroundTasks, user: User = Depends(current_user), db: Session = Depends(get_db), ) -> dict: """Set the per-user watch status (new|watched|hidden). watched→scrobble / →new→unscrobble push to a linked Plex account immediately; `hidden` is a Siftlode-only concept with no Plex analogue, so it never touches Plex.""" it = item_or_404(db, rating_key) status = body.status if status not in ("new", "watched", "hidden"): raise HTTPException(status_code=400, detail="Invalid status") st = ( db.execute( select(PlexState).where(PlexState.user_id == user.id, PlexState.item_id == it.id) ) .scalars() .first() ) prev_status = st.status if st is not None else "new" if status == "new": if st is not None: db.delete(st) db.commit() if prev_status != "new": # something → unwatched: mirror the un-mark to Plex _push_watch(background, db, user.id, it.id, rating_key, "unwatched") return {"status": "new"} if st is None: st = PlexState(user_id=user.id, item_id=it.id) db.add(st) st.status = status if status == "watched": st.watched_at = datetime.now(timezone.utc) st.position_seconds = 0 st.synced_to_plex = False db.commit() if status == "watched" and prev_status != "watched": _push_watch(background, db, user.id, it.id, rating_key, "watched") return {"status": status} def _bulk_state( background: BackgroundTasks, db: Session, user_id: int, eps: list[PlexItem], watched: bool, ) -> int: """Mark every episode in `eps` watched (or unwatched) for a user, mirroring each change to a linked Plex account in the background (best-effort). Returns how many rows actually changed.""" now = datetime.now(timezone.utc) existing = { s.item_id: s for s in db.execute( select(PlexState).where( PlexState.user_id == user_id, PlexState.item_id.in_([e.id for e in eps] or [0]), ) ).scalars() } pushable = plex_watch.link_for_push(db, user_id) is not None to_push: list[tuple[int, str, str]] = [] for e in eps: st = existing.get(e.id) prev = st.status if st is not None else "new" if watched: if st is None: st = PlexState(user_id=user_id, item_id=e.id) db.add(st) st.status = "watched" st.watched_at = now st.position_seconds = 0 st.synced_to_plex = False if prev != "watched": to_push.append((e.id, e.rating_key, "watched")) else: # → new: drop any state (but keep the user's private "hidden" marks) if st is not None and prev != "hidden": db.delete(st) if prev != "new": to_push.append((e.id, e.rating_key, "unwatched")) db.commit() if pushable and to_push: # One coalesced background task for the whole batch (reuses a single session + Plex client), # not one task per episode — a 200-ep show mark is 1 task, not 200. background.add_task(plex_watch.push_bulk_state_to_plex, user_id, to_push) return len(to_push) class WatchedIn(BaseModel): watched: bool = False @router.post("/show/{rating_key}/state") def show_state( rating_key: str, body: WatchedIn, background: BackgroundTasks, user: User = Depends(current_user), db: Session = Depends(get_db), ) -> dict: """Mark a whole show watched/unwatched (all its episodes) for the current user + push to Plex.""" sh = by_rating_key(db, PlexShow, rating_key) if sh is None: raise HTTPException(status_code=404, detail="Unknown Plex show") watched = body.watched eps = ( db.execute(select(PlexItem).where(PlexItem.show_id == sh.id, PlexItem.kind == "episode")) .scalars() .all() ) changed = _bulk_state(background, db, user.id, eps, watched) return {"changed": changed, "watched": watched} @router.post("/season/{rating_key}/state") def season_state( rating_key: str, body: WatchedIn, background: BackgroundTasks, user: User = Depends(current_user), db: Session = Depends(get_db), ) -> dict: """Mark a whole season watched/unwatched (all its episodes) for the current user + push to Plex.""" se = by_rating_key(db, PlexSeason, rating_key) if se is None: raise HTTPException(status_code=404, detail="Unknown Plex season") watched = body.watched eps = ( db.execute(select(PlexItem).where(PlexItem.season_id == se.id, PlexItem.kind == "episode")) .scalars() .all() ) changed = _bulk_state(background, db, user.id, eps, watched) return {"changed": changed, "watched": watched} @router.post("/stream/{rating_key}/session") def stream_session( rating_key: str, start: float = 0.0, audio: int | None = None, aoff: float = 0.0, multi: bool = False, 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, video stream-copied); transcode files (non-h264: HEVC/VP9/AV1/…) take the same HLS path but with the video re-encoded to H.264 (VAAPI or libx264). Selecting a non-default audio track (`audio` = stream ordinal) forces the HLS path so the chosen audio can be muxed in. Subtitles are independent WebVTT tracks (GET /subtitle) — no session restart.""" plex_stream.reap_idle() # opportunistic cleanup of idle sessions on each new playback it = item_or_404(db, rating_key) transcode = it.playable == "transcode" # `multi` (item has ≥2 audio tracks) forces the HLS path so ALL audio tracks ship as renditions and # the client can switch audio without a restart — even for otherwise direct-playable files. if it.playable == "direct" and audio is None and not aoff and not multi: return { "mode": "direct", "url": f"/api/plex/stream/{it.rating_key}/file", "start_s": 0.0, } 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, audio, aoff, multi, transcode=transcode) if s is None: raise HTTPException(status_code=404, detail="Media file not found on disk") # Report the REAL media start (the keyframe ffmpeg landed on, which hls.js zero-bases to), not the # requested offset — the client's absolute clock and the subtitle cue-shift both key off this. # `transcode`/`hw`/`tonemap` let the player show an honest cold-start ("Preparing stream…") and # surface the active mode + encoder in the stats overlay (`tonemap` = the HW HDR→SDR path); both # segment streams are otherwise identical. return { "mode": "hls", "start_s": s.media_start_s, "url": f"/api/plex/stream/{it.rating_key}/hls/{s.entry}", "transcode": s.transcode, "hw": s.hw, "tonemap": s.tonemap, } @router.get("/stream/{rating_key}/file") def stream_file( rating_key: str, download: bool = False, user: User = Depends(current_user), db: Session = Depends(get_db), ) -> FileResponse: """Serve the raw local media file with HTTP range support. Used both for direct-play (inline) and for the player's "download original" button (`?download=1` → attachment). The download is the untouched physical file — no re-encode/repackage.""" 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") if download: # A clean, safe download name from the title + the real file extension. ext = src.suffix or (f".{it.container}" if it.container else "") base = ( "".join(c for c in (it.title or "video") if c.isalnum() or c in " -_.").strip() or "video" ) return FileResponse(str(src), filename=f"{base}{ext}") return FileResponse(str(src)) # Starlette honours the Range header (206) _HLS_CT = { ".m3u8": "application/vnd.apple.mpegurl", ".ts": "video/mp2t", ".vtt": "text/vtt", } @router.get("/stream/{rating_key}/hls/{filename}") def stream_hls_file( rating_key: str, filename: str, user: User = Depends(current_user), db: Session = Depends(get_db), ) -> Response: """Serve a file (playlist / segment / subtitle) from the current HLS session directory. The playlist references its segments/subtitles by relative name, all resolving back here.""" # Filename must be a plain hls artifact — no path separators, only the expected extensions. ext = "." + filename.rsplit(".", 1)[-1] if "." in filename else "" if ext not in _HLS_CT or "/" in filename or "\\" in filename or ".." in filename: raise HTTPException(status_code=404, detail="Not found") s = plex_stream.current_session(str(rating_key)) if s is None: # No session yet (e.g. reaped): the ENTRY playlist may auto-start one at 0 — master.m3u8 for a # multi-audio item, index.m3u8 for a single-audio remux item. it = item_or_404(db, rating_key) if filename == "master.m3u8": s = plex_stream.start_session(db, it, 0.0, multi=True) elif filename == "index.m3u8" and it.playable == "remux": s = plex_stream.start_session(db, it, 0.0) else: raise HTTPException(status_code=404, detail="No active playback session") if s is None: raise HTTPException(status_code=404, detail="Media file not found on disk") path = s.dir / filename timeout = 30 if ext == ".m3u8" else 20 if not plex_stream.wait_for(path, timeout): raise HTTPException(status_code=404, detail="Not available yet") return Response( content=path.read_bytes(), media_type=_HLS_CT[ext], headers={"Cache-Control": "no-store"}, )