"""Plex integration API. Read endpoints (libraries/browse/show/image) are available to ANY authenticated user (demo + pure-Siftlode included) — the module is an access layer to the Plex library WITHOUT requiring a plex.tv account, and playback is a local file. Admin endpoints test the connection and run the catalog sync. """ import logging from datetime import datetime, timezone from urllib.parse import quote import httpx 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 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 from app.routes.feed import _to_tsquery_str log = logging.getLogger("siftlode.plex") router = APIRouter(prefix="/api/plex", tags=["plex"]) _TS_CONFIG = "public.unaccent_simple" # --- Admin: connectivity + sync --------------------------------------------------------------- @router.post("/test") def test_connection(_: User = Depends(admin_user), db: Session = Depends(get_db)) -> dict: """Admin: verify the configured Plex server URL + token, returning the server name and the movie/show sections (for the config library-picker). Also probes ONE real media file through the local mount, so a missing bind-mount / wrong path-map is caught here rather than only surfacing as a dead player later.""" try: with PlexClient(db) as plex: info = plex.server_info() sections = plex.sections() media_check = _media_check(db, plex, sections) except PlexNotConfigured as e: raise HTTPException(status_code=400, detail=str(e)) except PlexError as e: raise HTTPException(status_code=422, detail=f"Plex connection failed: {e}") return { "ok": True, "server_name": info.get("friendlyName") or info.get("title1") or "Plex", "version": info.get("version"), "sections": [ {"key": str(s.get("key")), "title": s.get("title"), "type": s.get("type"), "count": s.get("count")} for s in sections if s.get("type") in ("movie", "show") ], "media_check": media_check, } def _sample_plex_path(db: Session, plex: PlexClient, sections: list[dict]) -> str | None: """A real Plex-side media file path to validate the mount + path-map against. Prefers the already-mirrored catalog (cheap); before the first sync, pulls one movie leaf from an enabled section straight from Plex.""" row = db.query(PlexItem.file_path).filter(PlexItem.file_path.isnot(None)).first() if row and row[0]: return row[0] wanted = plex_sync._enabled_section_keys(db) for s in sections: if s.get("type") != "movie": continue key = str(s.get("key")) if wanted is not None and key not in wanted: continue items, _ = plex.section_items(key, 1, 0, 1) if items: fp = plex_sync._media_facts(items[0]).get("file_path") if fp: return fp return None def _media_check(db: Session, plex: PlexClient, sections: list[dict]) -> dict: """Resolve a sample Plex file path to its local mount path and confirm it's readable, so the admin sees at Test time whether the physical media is actually reachable (bind-mount + path-map correct). ``checked=False`` when no sample is available (empty library).""" plex_path = _sample_plex_path(db, plex, sections) if not plex_path: return {"checked": False} local_path = plex_paths.map_path(db, plex_path) ok = plex_paths.local_media_path(db, plex_path) is not None return {"checked": True, "ok": ok, "plex_path": plex_path, "local_path": local_path} @router.post("/sync") def trigger_sync(_: User = Depends(admin_user), db: Session = Depends(get_db)) -> dict: """Admin: mirror the enabled Plex sections into the local catalog. Synchronous (can take a while on a large library); a background/scheduled sync also runs periodically.""" return plex_sync.sync(db) # --- Read: libraries / browse / show / image -------------------------------------------------- def _enabled() -> None: """Guard read endpoints when the module is off (avoids leaking a stale mirror).""" # Read endpoints stay usable as long as a mirror exists; the toggle only gates sync + UI. return None @router.get("/libraries") def list_libraries(user: User = Depends(current_user), db: Session = Depends(get_db)) -> dict: """The mirrored libraries the user can browse (drives the Plex scope selector).""" libs = db.query(PlexLibrary).order_by(PlexLibrary.kind.desc(), PlexLibrary.title).all() out = [] for lib in libs: if lib.kind == "movie": count = db.query(func.count(PlexItem.id)).filter_by(library_id=lib.id, kind="movie").scalar() else: count = db.query(func.count(PlexShow.id)).filter_by(library_id=lib.id).scalar() out.append({"key": lib.plex_key, "title": lib.title, "kind": lib.kind, "count": count or 0}) return {"enabled": sysconfig.get_bool(db, "plex_enabled"), "libraries": out} def _movie_card(it: PlexItem, st: PlexState | None) -> dict: return { "id": it.rating_key, "type": "movie", "title": it.title, "year": it.year, "duration_seconds": it.duration_s, "thumb": f"/api/plex/image/{it.rating_key}", "playable": it.playable, "status": st.status if st else "new", "position_seconds": st.position_seconds if st else 0, } def _show_card(sh: PlexShow) -> dict: return { "id": sh.rating_key, "type": "show", "title": sh.title, "year": sh.year, "thumb": f"/api/plex/image/{sh.rating_key}", "season_count": sh.child_count, } def _episode_card(e: PlexItem, st: PlexState | None) -> dict: return { "id": e.rating_key, "type": "episode", "title": e.title, "summary": e.summary, "season_number": e.season_number, "episode_number": e.episode_number, "duration_seconds": e.duration_s, "thumb": f"/api/plex/image/{e.rating_key}", "playable": e.playable, "status": st.status if st else "new", "position_seconds": st.position_seconds if st else 0, } @router.get("/browse") def browse( library: str, q: str | None = None, sort: str = "added", show: str = "all", offset: int = 0, limit: int = Query(default=40, ge=1, le=100), user: User = Depends(current_user), db: Session = Depends(get_db), ) -> dict: """List a library's movies (leaves) or shows, with optional FTS search + sorting + a per-user watch-state filter (`show`: all|unwatched|in_progress|watched — movie libraries only). Movie libraries return playable movie cards; show libraries return show cards (drill in via /show).""" lib = db.query(PlexLibrary).filter_by(plex_key=str(library)).first() if lib is None: raise HTTPException(status_code=404, detail="Unknown Plex library") offset = max(0, offset) model = PlexItem if lib.kind == "movie" else PlexShow query = db.query(model) if lib.kind == "movie": query = query.filter(PlexItem.library_id == lib.id, PlexItem.kind == "movie") # Per-user watch-state filter (movies only; a show isn't a single watch unit). st = aliased(PlexState) query = query.outerjoin(st, and_(st.item_id == PlexItem.id, st.user_id == user.id)) if show == "watched": query = query.filter(st.status == "watched") elif show == "in_progress": query = query.filter(st.position_seconds > 0, or_(st.status.is_(None), st.status != "watched")) elif show == "unwatched": query = query.filter(or_(st.status.is_(None), st.status.notin_(["watched", "hidden"]))) else: # all — hide only the explicitly hidden query = query.filter(or_(st.status.is_(None), st.status != "hidden")) else: query = query.filter(PlexShow.library_id == lib.id) ts = _to_tsquery_str(q) if q else None tsq = None if ts: tsq = func.to_tsquery(_TS_CONFIG, ts) query = query.filter(model.search_vector.op("@@")(tsq)) total = query.count() if tsq is not None: order = [func.ts_rank(model.search_vector, tsq).desc()] elif sort == "title": order = [func.lower(model.title).asc()] else: # newest first order = [model.added_at.desc().nullslast()] order.append(model.id.desc()) rows = query.order_by(*order).offset(offset).limit(limit).all() if lib.kind == "movie": by_item = {r.id: r for r in rows} states = { s.item_id: s for s in db.query(PlexState).filter( PlexState.user_id == user.id, PlexState.item_id.in_(list(by_item.keys()) or [0]) ) } items = [_movie_card(r, states.get(r.id)) for r in rows] else: items = [_show_card(r) for r in rows] return {"kind": lib.kind, "total": total, "offset": offset, "limit": limit, "items": items} @router.get("/show/{rating_key}") def show_detail( rating_key: str, user: User = Depends(current_user), db: Session = Depends(get_db), ) -> dict: """A show's seasons + episodes (the drill-down view) with per-user watch state.""" sh = db.query(PlexShow).filter_by(rating_key=str(rating_key)).first() if sh is None: raise HTTPException(status_code=404, detail="Unknown Plex show") seasons = ( db.query(PlexSeason).filter_by(show_id=sh.id).order_by(PlexSeason.season_number.asc().nullsfirst()).all() ) eps = ( db.query(PlexItem) .filter_by(show_id=sh.id, kind="episode") .order_by(PlexItem.season_number.asc().nullsfirst(), PlexItem.episode_number.asc().nullsfirst()) .all() ) states = { s.item_id: s for s in db.query(PlexState).filter( PlexState.user_id == user.id, PlexState.item_id.in_([e.id for e in eps] or [0]) ) } by_season: dict[int | None, list] = {} for e in eps: by_season.setdefault(e.season_id, []).append(_episode_card(e, states.get(e.id))) return { "show": { "id": sh.rating_key, "title": sh.title, "summary": sh.summary, "year": sh.year, "thumb": f"/api/plex/image/{sh.rating_key}", "art": f"/api/plex/image/{sh.rating_key}?variant=art", }, "seasons": [ { "id": se.rating_key, "season_number": se.season_number, "title": se.title or (f"Season {se.season_number}" if se.season_number else "Season"), "thumb": f"/api/plex/image/{se.rating_key}", "episodes": by_season.get(se.id, []), } for se in seasons ], } def _image_key(db: Session, rating_key: str, variant: str) -> str | None: """The stored Plex image path for a known rating_key (item/show/season). Only mirrored keys are proxyable — this is not an open image proxy.""" it = db.query(PlexItem).filter_by(rating_key=rating_key).first() if it is not None: return it.art_key if variant == "art" else it.thumb_key sh = db.query(PlexShow).filter_by(rating_key=rating_key).first() if sh is not None: return sh.art_key if variant == "art" else sh.thumb_key se = db.query(PlexSeason).filter_by(rating_key=rating_key).first() if se is not None: return se.thumb_key return None @router.get("/image/{rating_key}") def image( rating_key: str, variant: str = "thumb", user: User = Depends(current_user), db: Session = Depends(get_db), ) -> Response: """Proxy a Plex poster/art image (keeps the admin token server-side; no image duplication).""" key = _image_key(db, str(rating_key), "art" if variant == "art" else "thumb") if not key: raise HTTPException(status_code=404, detail="No image") try: with PlexClient(db) as plex: data, ct = plex.image_bytes(key) except PlexNotConfigured as e: raise HTTPException(status_code=400, detail=str(e)) 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 # Progress thresholds (mirror the YouTube player): ignore the first few seconds, and treat the # last stretch as "finished" (mark watched, clear the resume point). _PROGRESS_MIN_S = 5 _FINISH_MARGIN_S = 30 # Cast/crew photos come from Plex's PUBLIC metadata CDN (no token). We proxy them (host-whitelisted) # so the user's browser makes no third-party request (privacy + self-host CSP friendliness). _PERSON_IMG_HOST = "https://metadata-static.plex.tv/" def _rich_meta(meta: dict) -> dict: """Pull the "info page" extras out of a full Plex metadata dict: an IMDb score + link, content rating, genres, director(s), studio, tagline, and cast (name/role/photo). Best-effort — any missing field is just omitted.""" # Rating: prefer the explicit IMDb entry in the Rating array, else the generic audienceRating. imdb_rating = None for r in meta.get("Rating") or []: if str(r.get("image") or "").startswith("imdb://"): try: imdb_rating = round(float(r.get("value")), 1) except (TypeError, ValueError): pass break if imdb_rating is None: try: imdb_rating = round(float(meta.get("audienceRating") or meta.get("rating")), 1) except (TypeError, ValueError): imdb_rating = None # IMDb id (tt…) from the Guid list → a clickable link. imdb_id = None for g in meta.get("Guid") or []: gid = str(g.get("id") or "") if gid.startswith("imdb://"): imdb_id = gid.split("imdb://", 1)[1] break cast = [] for role in (meta.get("Role") or [])[:24]: name = role.get("tag") if not name: continue thumb = str(role.get("thumb") or "") cast.append( { "name": name, "role": role.get("role"), "thumb": f"/api/plex/person-image?u={quote(thumb, safe='')}" if thumb.startswith(_PERSON_IMG_HOST) else None, } ) return { "imdb_rating": imdb_rating, "imdb_id": imdb_id, "imdb_url": f"https://www.imdb.com/title/{imdb_id}/" if imdb_id else None, "content_rating": meta.get("contentRating"), "genres": [g.get("tag") for g in (meta.get("Genre") or []) if g.get("tag")][:8], "directors": [d.get("tag") for d in (meta.get("Director") or []) if d.get("tag")][:4], "studio": meta.get("studio"), "tagline": meta.get("tagline"), "cast": cast, } @router.get("/person-image") def person_image(u: str, _: User = Depends(current_user)) -> Response: """Proxy a cast/crew photo from Plex's public metadata CDN (host-whitelisted; no token needed).""" if not u.startswith(_PERSON_IMG_HOST): raise HTTPException(status_code=400, detail="Unsupported image host") 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") return Response( content=r.content, media_type=r.headers.get("content-type", "image/jpeg"), headers={"Cache-Control": "public, max-age=604800"}, ) def _episode_nav(db: Session, it: PlexItem) -> tuple[str | None, str | None]: if it.kind != "episode" or it.show_id is None: return None, None keys = [ r[0] for r in db.query(PlexItem.rating_key) .filter_by(show_id=it.show_id, kind="episode") .order_by(PlexItem.season_number.asc().nullsfirst(), PlexItem.episode_number.asc().nullsfirst()) .all() ] try: i = keys.index(it.rating_key) except ValueError: return None, None return (keys[i - 1] if i > 0 else None), (keys[i + 1] if i < len(keys) - 1 else None) @router.get("/item/{rating_key}") def item_detail( rating_key: str, user: User = Depends(current_user), db: Session = Depends(get_db), ) -> dict: """Everything the player needs: metadata, cast, intro/credit markers (from Plex), episode prev/next, and the per-user resume position + watch status.""" it = _item_or_404(db, rating_key) st = db.query(PlexState).filter_by(user_id=user.id, item_id=it.id).first() rich: dict = {} markers: list[dict] = [] audio_streams: list[dict] = [] subtitle_streams: list[dict] = [] try: with PlexClient(db) as plex: meta = plex.metadata(it.rating_key, markers=True) or {} rich = _rich_meta(meta) for m in meta.get("Marker") or []: if m.get("type") in ("intro", "credits"): markers.append( { "type": m.get("type"), "start_s": round((m.get("startTimeOffset") or 0) / 1000, 1), "end_s": round((m.get("endTimeOffset") or 0) / 1000, 1), } ) # Audio + subtitle tracks (ordinals among their stream type — used to -map the selection). part = ((meta.get("Media") or [{}])[0].get("Part") or [{}])[0] for s in part.get("Stream") or []: if s.get("streamType") == 2: audio_streams.append( { "ord": len(audio_streams), "label": s.get("displayTitle") or s.get("language") or f"Audio {len(audio_streams) + 1}", "language": s.get("languageTag") or s.get("language"), "default": bool(s.get("selected") or s.get("default")), } ) elif s.get("streamType") == 3: subtitle_streams.append( { "ord": len(subtitle_streams), "label": s.get("displayTitle") or s.get("language") or f"Subtitle {len(subtitle_streams) + 1}", "language": s.get("languageTag") or s.get("language"), } ) except (PlexError, PlexNotConfigured): pass # metadata extras are best-effort; playback works without them prev_id, next_id = _episode_nav(db, it) show = db.get(PlexShow, it.show_id) if it.kind == "episode" and it.show_id else None return { "id": it.rating_key, "kind": it.kind, "title": it.title, "summary": it.summary, "year": it.year, "duration_seconds": it.duration_s, "playable": it.playable, "thumb": f"/api/plex/image/{it.rating_key}", "art": f"/api/plex/image/{it.rating_key}?variant=art", "cast": rich.get("cast", []), "imdb_rating": rich.get("imdb_rating"), "imdb_id": rich.get("imdb_id"), "imdb_url": rich.get("imdb_url"), "content_rating": rich.get("content_rating"), "genres": rich.get("genres", []), "directors": rich.get("directors", []), "studio": rich.get("studio"), "tagline": rich.get("tagline"), "markers": markers, "audio_streams": audio_streams, "subtitle_streams": subtitle_streams, "status": st.status if st else "new", "position_seconds": st.position_seconds if st else 0, "show_title": show.title if show else None, "season_number": it.season_number, "episode_number": it.episode_number, "prev_id": prev_id, "next_id": next_id, } @router.post("/item/{rating_key}/progress") def item_progress( rating_key: str, payload: dict, user: User = Depends(current_user), db: Session = Depends(get_db), ) -> dict: """Checkpoint the resume position while playing (near the end → mark watched + clear it).""" it = _item_or_404(db, rating_key) try: position = max(0, int(payload.get("position_seconds") or 0)) duration = int(payload.get("duration_seconds") or 0) except (TypeError, ValueError): raise HTTPException(status_code=400, detail="position_seconds must be an integer") now = datetime.now(timezone.utc) st = db.query(PlexState).filter_by(user_id=user.id, item_id=it.id).first() 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 db.commit() 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 db.commit() return {"position_seconds": position} @router.post("/item/{rating_key}/state") def item_state( rating_key: str, payload: dict, user: User = Depends(current_user), db: Session = Depends(get_db), ) -> dict: """Set the per-user watch status (new|watched|hidden).""" it = _item_or_404(db, rating_key) status = payload.get("status") if status not in ("new", "watched", "hidden"): raise HTTPException(status_code=400, detail="Invalid status") st = db.query(PlexState).filter_by(user_id=user.id, item_id=it.id).first() if status == "new": if st is not None: db.delete(st) db.commit() 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 db.commit() return {"status": status} @router.post("/stream/{rating_key}/session") def stream_session( rating_key: str, start: float = 0.0, audio: int | None = None, subtitle: int | None = None, 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. Selecting a non-default audio/subtitle track (`audio`/`subtitle` = stream ordinals) forces the HLS path so the chosen tracks can be muxed in.""" plex_stream.reap_idle() # opportunistic cleanup of idle sessions on each new playback it = _item_or_404(db, rating_key) if it.playable == "transcode": raise HTTPException(status_code=501, detail="This format needs transcoding (a later phase).") select = audio is not None or subtitle is not None if it.playable == "direct" and not select: 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, subtitle) 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}/hls/{s.entry}"} @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: only the entry playlist for a remux item may auto-start one at 0. it = _item_or_404(db, rating_key) if filename != "index.m3u8" or it.playable != "remux": raise HTTPException(status_code=404, detail="No active playback session") 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") 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"} )