"""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 hashlib import logging import os import re import subprocess import tempfile import threading import time from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass from datetime import datetime, timedelta, timezone from pathlib import Path from urllib.parse import quote import httpx from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query, Request, Response from fastapi.responses import FileResponse from sqlalchemy import Integer, String, and_, case, cast, func, literal, null, or_ from sqlalchemy.orm import Session, aliased from app import sysconfig from app.auth import current_user, resolved_user_id from app.config import settings from app.db import SessionLocal, get_db from app.models import ( PlexCollection, PlexItem, PlexLibrary, PlexLink, PlexPlaylist, PlexPlaylistItem, 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 import watch_sync as plex_watch 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) # --- Watch-state sync (Plex ↔ Siftlode) ------------------------------------------------------- # Phase A: the owner links their Siftlode account to the Plex admin account and does a one-time # "Plex is master" import. Two-way push/pull lands in later phases. Admin-only for now because it # rides the server admin token (which reads/writes the OWNER Plex account's state). def _link_dict(link: PlexLink | None) -> dict: return { "linked": link is not None, "sync_enabled": bool(link and link.sync_enabled), "uses_admin": bool(link and link.uses_admin), "initial_import_done": bool(link and link.initial_import_done), "last_watch_sync_at": ( link.last_watch_sync_at.isoformat() if link and link.last_watch_sync_at else None ), } @router.get("/watch/link") def watch_link_status( user: User = Depends(current_user), db: Session = Depends(get_db) ) -> dict: """This user's Plex watch-sync link status (drives the Settings toggle).""" link = db.query(PlexLink).filter_by(user_id=user.id).first() return _link_dict(link) @router.post("/watch/link") def watch_link_set( payload: dict, user: User = Depends(admin_user), db: Session = Depends(get_db), ) -> dict: """Admin: turn two-way Plex watch-sync on/off for THIS (owner) account. The link uses the server admin token, so no separate Plex login. Enabling it for the first time runs the one-time "Plex is master" import immediately so the owner's existing Plex history shows up in Siftlode.""" if not sysconfig.get_bool(db, "plex_enabled"): raise HTTPException(status_code=409, detail="The Plex module is disabled.") enabled = bool(payload.get("enabled")) link = db.query(PlexLink).filter_by(user_id=user.id).first() if link is None: link = PlexLink(user_id=user.id, uses_admin=True, linked_at=datetime.now(timezone.utc)) db.add(link) link.sync_enabled = enabled db.commit() result: dict = {} if enabled and not link.initial_import_done: result = plex_watch.import_owner_watch_state(db, link) return {**_link_dict(link), "import": result or None} @router.post("/watch/import") def watch_import_now( user: User = Depends(admin_user), db: Session = Depends(get_db) ) -> dict: """Admin: (re-)run the "Plex is master" import for this account — Plex re-wins on the intersection; Siftlode-only states are preserved. Idempotent.""" link = db.query(PlexLink).filter_by(user_id=user.id).first() if link is None or not link.sync_enabled: raise HTTPException(status_code=409, detail="Enable Plex watch-sync first.") result = plex_watch.import_owner_watch_state(db, link) return {**_link_dict(link), "import": result} # --- Read: libraries / browse / show / image -------------------------------------------------- @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, status: str = "new") -> 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, # Aggregate watch-state across the show's episodes (new | in_progress | watched) → grid badge. "status": status, } def _show_status(total: int, watched: int, inprog: int) -> str: """Roll a show's episodes up into a single watch state: fully watched, partially started, or new.""" if total and watched >= total: return "watched" if (watched or 0) > 0 or (inprog or 0) > 0: return "in_progress" return "new" def _rollup(cards: list[dict]) -> dict: """From ORDERED episode cards → the aggregate watch status, the "on deck" episode to Resume (the last in-progress one, else the first unwatched), the first episode (for Play from the start), and the episode count. Drives the show/season Resume + Play + mark-all buttons.""" total = len(cards) watched = sum(1 for c in cards if c.get("status") == "watched") inprog = [c for c in cards if (c.get("position_seconds") or 0) > 0 and c.get("status") != "watched"] resume = None if inprog: resume = inprog[-1] # continue the latest-started episode else: resume = next((c for c in cards if c.get("status") != "watched"), None) return { "status": _show_status(total, watched, len(inprog)), "resume": resume, "first": cards[0] if cards else None, "episode_count": total, } def _show_agg_subq(db: Session, user_id: int): """Per-show episode counts for a user: total, watched, in-progress. Reused to (a) filter the show grid by aggregate watch-state and (b) compute each card's badge.""" ps = aliased(PlexState) return ( db.query( PlexItem.show_id.label("show_id"), func.count(PlexItem.id).label("total"), func.count(case((ps.status == "watched", 1))).label("watched"), func.count( case( ( and_(ps.position_seconds > 0, or_(ps.status.is_(None), ps.status != "watched")), 1, ) ) ).label("inprog"), ) .outerjoin(ps, and_(ps.item_id == PlexItem.id, ps.user_id == user_id)) .filter(PlexItem.kind == "episode", PlexItem.show_id.isnot(None)) .group_by(PlexItem.show_id) ) def _show_state_map(db: Session, user_id: int, show_ids: list[int]) -> dict[int, str]: if not show_ids: return {} rows = _show_agg_subq(db, user_id).filter(PlexItem.show_id.in_(show_ids)).all() return {r.show_id: _show_status(r.total, r.watched, r.inprog) for r in rows} def _lib_key(db: Session, library_id: int | None) -> str | None: """The Plex section key for a library id — the info/show page passes it to the admin collection editor (in the unified cross-library view there's no single library prop otherwise).""" if library_id is None: return None lib = db.get(PlexLibrary, library_id) return lib.plex_key if lib else None 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, } def _leaf_card( it: PlexItem, st: PlexState | None, show_title: str | None = None, show_rk: str | None = None ) -> dict: """A playable-leaf card (movie or episode). Used by playlists, which can mix both kinds. `show_rk` (the show's rating_key) lets the client group a playlist's episodes by show/season.""" if it.kind == "episode": card = _episode_card(it, st) card["show_title"] = show_title card["show_id"] = show_rk return card return _movie_card(it, st) def _csv(v: str | None) -> list[str]: return [s.strip() for s in v.split(",") if s.strip()] if v else [] _ADDED_WINDOWS = {"24h": 1, "1w": 7, "1m": 30, "6m": 182, "1y": 365} def _added_cutoff(within: str | None) -> datetime | None: days = _ADDED_WINDOWS.get(within or "") return datetime.now(timezone.utc) - timedelta(days=days) if days else None def _meta_filter_conds(model, p: dict) -> list: """The shared metadata-filter conditions (identical column names on plex_items + plex_shows) for `p` (the request's filter params). Duration is movie-only, handled by the caller. Returned as a list so faceted search can apply all-but-one group. DRY for browse, unified, and /facets.""" conds: list = [] gsel = _csv(p.get("genres")) if gsel: gc = [model.genres.contains([g]) for g in gsel] conds.append(and_(*gc) if p.get("genre_mode") == "all" else or_(*gc)) crs = _csv(p.get("content_ratings")) if crs: conds.append(model.content_rating.in_(crs)) if p.get("year_min") is not None: conds.append(model.year >= p["year_min"]) if p.get("year_max") is not None: conds.append(model.year <= p["year_max"]) if p.get("rating_min") is not None: conds.append(model.rating >= p["rating_min"]) cutoff = _added_cutoff(p.get("added_within")) if cutoff is not None: conds.append(model.added_at >= cutoff) for d in _csv(p.get("directors")): # AND conds.append(model.directors.contains([d])) for a in _csv(p.get("actors")): # AND conds.append(model.cast_names.contains([a])) stds = _csv(p.get("studios")) if stds: # OR conds.append(model.studio.in_(stds)) if p.get("collection"): conds.append(model.collection_keys.contains([p["collection"]])) return conds def _apply_meta_filters(q, model, p: dict): conds = _meta_filter_conds(model, p) return q.filter(*conds) if conds else q @dataclass class LibraryFilters: """The shared sidebar filter query-params for /library + /facets, declared once and injected via ``Depends()``. `as_dict()` yields the `p` dict the query builders read; `duration_*` is included for /facets' wsq but ignored by `_meta_filter_conds` (both callers apply duration separately).""" genres: str | None = None genre_mode: str = "any" content_ratings: str | None = None year_min: int | None = None year_max: int | None = None rating_min: float | None = None duration_min: int | None = None duration_max: int | None = None added_within: str | None = None directors: str | None = None actors: str | None = None studios: str | None = None collection: str | None = None def as_dict(self) -> dict: return { "genres": self.genres, "genre_mode": self.genre_mode, "content_ratings": self.content_ratings, "year_min": self.year_min, "year_max": self.year_max, "rating_min": self.rating_min, "duration_min": self.duration_min, "duration_max": self.duration_max, "added_within": self.added_within, "directors": self.directors, "actors": self.actors, "studios": self.studios, "collection": self.collection, } @router.get("/library") def unified_library( scope: str = "both", q: str | None = None, sort: str = "added", sort_dir: str = "desc", show: str = "all", offset: int = 0, limit: int = Query(default=40, ge=1, le=100), f: LibraryFilters = Depends(), user: User = Depends(current_user), db: Session = Depends(get_db), ) -> dict: """Unified cross-library browse: movies + shows in ONE mixed, paginated feed, scoped to movie|show|both, with the shared filters/search/watch-state (a show's state is the aggregate of its episodes). On a search that also matches episodes (and shows are in scope), the matching episodes come back in a separate `episodes` list (grouped result — the 'Episodes' section).""" offset = max(0, offset) p = f.as_dict() libs = db.query(PlexLibrary).filter_by(enabled=True).all() movie_lib_ids = [lb.id for lb in libs if lb.kind == "movie"] show_lib_ids = [lb.id for lb in libs if lb.kind == "show"] want_movies = scope in ("movie", "both") and bool(movie_lib_ids) want_shows = scope in ("show", "both") and bool(show_lib_ids) ts = _to_tsquery_str(q) if q else None tsq = func.to_tsquery(_TS_CONFIG, ts) if ts else None selects = [] if want_movies: st = aliased(PlexState) m_status = case( (st.status == "watched", "watched"), (and_(st.position_seconds > 0, or_(st.status.is_(None), st.status != "watched")), "in_progress"), else_="new", ) mq = ( db.query( PlexItem.rating_key.label("rk"), literal("movie").label("kind"), PlexItem.title.label("title"), PlexItem.year.label("year"), PlexItem.rating.label("rating"), PlexItem.added_at.label("added_at"), PlexItem.originally_available_at.label("rel"), PlexItem.duration_s.label("duration_s"), PlexItem.playable.label("playable"), cast(null(), Integer).label("season_count"), func.coalesce(st.position_seconds, 0).label("position_seconds"), m_status.label("status"), (func.ts_rank(PlexItem.search_vector, tsq) if tsq is not None else literal(0.0)).label("rank"), ) .outerjoin(st, and_(st.item_id == PlexItem.id, st.user_id == user.id)) .filter(PlexItem.kind == "movie", PlexItem.library_id.in_(movie_lib_ids)) ) if show == "watched": mq = mq.filter(st.status == "watched") elif show == "in_progress": mq = mq.filter(st.position_seconds > 0, or_(st.status.is_(None), st.status != "watched")) elif show == "unwatched": mq = mq.filter(or_(st.status.is_(None), st.status.notin_(["watched", "hidden"]))) else: mq = mq.filter(or_(st.status.is_(None), st.status != "hidden")) mq = _apply_meta_filters(mq, PlexItem, p) if f.duration_min is not None: mq = mq.filter(PlexItem.duration_s >= f.duration_min) if f.duration_max is not None: mq = mq.filter(PlexItem.duration_s <= f.duration_max) if tsq is not None: mq = mq.filter(PlexItem.search_vector.op("@@")(tsq)) selects.append(mq) if want_shows: agg = _show_agg_subq(db, user.id).subquery() tot = func.coalesce(agg.c.total, 0) wat = func.coalesce(agg.c.watched, 0) inp = func.coalesce(agg.c.inprog, 0) s_status = case( (and_(tot > 0, wat >= tot), "watched"), (or_(wat > 0, inp > 0), "in_progress"), else_="new", ) sq = ( db.query( PlexShow.rating_key.label("rk"), literal("show").label("kind"), PlexShow.title.label("title"), PlexShow.year.label("year"), PlexShow.rating.label("rating"), PlexShow.added_at.label("added_at"), PlexShow.originally_available_at.label("rel"), cast(null(), Integer).label("duration_s"), cast(null(), String).label("playable"), PlexShow.child_count.label("season_count"), literal(0).label("position_seconds"), s_status.label("status"), (func.ts_rank(PlexShow.search_vector, tsq) if tsq is not None else literal(0.0)).label("rank"), ) .outerjoin(agg, agg.c.show_id == PlexShow.id) .filter(PlexShow.library_id.in_(show_lib_ids)) ) if show == "watched": sq = sq.filter(tot > 0, wat >= tot) elif show == "in_progress": sq = sq.filter(or_(wat > 0, inp > 0), wat < tot) elif show == "unwatched": sq = sq.filter(wat == 0, inp == 0) sq = _apply_meta_filters(sq, PlexShow, p) if tsq is not None: sq = sq.filter(PlexShow.search_vector.op("@@")(tsq)) selects.append(sq) if not selects: return {"scope": scope, "total": 0, "offset": offset, "limit": limit, "items": [], "episodes": []} union = selects[0] if len(selects) == 1 else selects[0].union_all(*selects[1:]) u = union.subquery() total = db.query(func.count()).select_from(u).scalar() or 0 if tsq is not None: order = [u.c.rank.desc()] else: col = { "title": func.lower(u.c.title), "year": u.c.year, "rating": u.c.rating, "duration": u.c.duration_s, "release": u.c.rel, }.get(sort, u.c.added_at) order = [(col.asc() if sort_dir == "asc" else col.desc()).nullslast()] order.append(u.c.rk.desc()) rows = db.query(u).order_by(*order).offset(offset).limit(limit).all() items = [] for r in rows: if r.kind == "movie": items.append({ "id": r.rk, "type": "movie", "title": r.title, "year": r.year, "duration_seconds": r.duration_s, "thumb": f"/api/plex/image/{r.rk}", "playable": r.playable, "status": r.status, "position_seconds": r.position_seconds, }) else: items.append({ "id": r.rk, "type": "show", "title": r.title, "year": r.year, "thumb": f"/api/plex/image/{r.rk}", "season_count": r.season_count, "status": r.status, }) episodes = [] if tsq is not None and want_shows: ep_st = aliased(PlexState) eps = ( db.query(PlexItem, ep_st, PlexShow.title.label("show_title")) .outerjoin(ep_st, and_(ep_st.item_id == PlexItem.id, ep_st.user_id == user.id)) .outerjoin(PlexShow, PlexShow.id == PlexItem.show_id) .filter( PlexItem.kind == "episode", PlexItem.library_id.in_(show_lib_ids), PlexItem.search_vector.op("@@")(tsq), ) .order_by(func.ts_rank(PlexItem.search_vector, tsq).desc()) .limit(24) .all() ) for e, s, show_title in eps: card = _episode_card(e, s) card["show_title"] = show_title episodes.append(card) return {"scope": scope, "total": total, "offset": offset, "limit": limit, "items": items, "episodes": episodes} @router.get("/facets") def facets( scope: str = "both", show: str = "all", f: LibraryFilters = Depends(), user: User = Depends(current_user), db: Session = Depends(get_db), ) -> dict: """Available filter values for the unified library (scope movie|show|both), ADAPTED to the active filters (faceted search): each group's values/bounds are computed with all the OTHER active filters applied but NOT its own — so selecting a filter narrows what the other groups offer, while a multi-select group still lets you add more. Merged across the scope's libraries; duration is movie-only. (Search text is not factored in — facets track the sidebar filters.)""" empty = { "genres": [], "content_ratings": [], "year_min": None, "year_max": None, "rating_max": None, "duration_min": None, "duration_max": None, } libs = db.query(PlexLibrary).filter_by(enabled=True).all() movie_ids = [lb.id for lb in libs if lb.kind == "movie"] if scope in ("movie", "both") else [] show_ids = [lb.id for lb in libs if lb.kind == "show"] if scope in ("show", "both") else [] if not movie_ids and not show_ids: return empty p = f.as_dict() uid = user.id def wsq(query, model, ids, pp, kind_movie, with_dur): """Apply the (self-excluded) metadata filters + library scope + the per-user watch-state to a facet aggregation query, so the facets reflect exactly the visible result set.""" query = query.filter(model.library_id.in_(ids), *_meta_filter_conds(model, pp)) if kind_movie: query = query.filter(model.kind == "movie") if with_dur and pp.get("duration_min") is not None: query = query.filter(model.duration_s >= pp["duration_min"]) if with_dur and pp.get("duration_max") is not None: query = query.filter(model.duration_s <= pp["duration_max"]) # Always join watch-state and mirror the grid's status branches (unified_library) exactly — # hidden movies never appear in the grid, so they must not inflate facet counts/bounds either. st = aliased(PlexState) query = query.outerjoin(st, and_(st.item_id == model.id, st.user_id == uid)) 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: query = query.filter(or_(st.status.is_(None), st.status != "hidden")) else: if show in ("watched", "in_progress", "unwatched"): agg = _show_agg_subq(db, uid).subquery() query = query.outerjoin(agg, agg.c.show_id == model.id) tot = func.coalesce(agg.c.total, 0) wat = func.coalesce(agg.c.watched, 0) inp = func.coalesce(agg.c.inprog, 0) if show == "watched": query = query.filter(tot > 0, wat >= tot) elif show == "in_progress": query = query.filter(or_(wat > 0, inp > 0), wat < tot) else: query = query.filter(wat == 0, inp == 0) return query genre_counts: dict[str, int] = {} cr_counts: dict[str, int] = {} years: list[int] = [] ratings: list[float] = [] durs: list[int] = [] def per_table(model, ids, kind_movie: bool) -> None: if not ids: return # Genres — in "any"/OR mode, exclude the genre filter itself so each offered genre is a valid # OR-addition that only widens (keep adding). In "all"/AND mode, KEEP the filter applied so we # offer only genres that CO-OCCUR on the current result set — a non-co-occurring genre would AND # the result down to zero, i.e. a dead-end chip. (The already-selected genres still appear, since # every matching title carries them, so they stay togglable.) genre_pp = p if p.get("genre_mode") == "all" else {**p, "genres": None} expanded = wsq( db.query(func.jsonb_array_elements_text(model.genres).label("g")), model, ids, genre_pp, kind_movie, True, ).subquery() for g, c in db.query(expanded.c.g, func.count()).group_by(expanded.c.g).all(): genre_counts[g] = genre_counts.get(g, 0) + c # Content ratings — exclude its own. crq = wsq( db.query(model.content_rating, func.count()), model, ids, {**p, "content_ratings": None}, kind_movie, True, ).filter(model.content_rating.isnot(None)).group_by(model.content_rating) for cr, c in crq.all(): cr_counts[cr] = cr_counts.get(cr, 0) + c # Year bounds — exclude the year filter. by = wsq( db.query(func.min(model.year), func.max(model.year)), model, ids, {**p, "year_min": None, "year_max": None}, kind_movie, True, ).first() if by[0] is not None: years.append(by[0]) if by[1] is not None: years.append(by[1]) # Rating max — exclude the rating filter. br = wsq(db.query(func.max(model.rating)), model, ids, {**p, "rating_min": None}, kind_movie, True).scalar() if br is not None: ratings.append(float(br)) # Duration bounds (movies) — exclude the duration filter. if kind_movie: bd = wsq( db.query(func.min(model.duration_s), func.max(model.duration_s)), model, ids, p, kind_movie, False ).first() if bd[0] is not None: durs.append(bd[0]) if bd[1] is not None: durs.append(bd[1]) per_table(PlexItem, movie_ids, True) per_table(PlexShow, show_ids, False) # Always surface the actively-selected genres (even a zero-match AND combo, where they'd otherwise # be absent): the sidebar renders genres solely from this list and hides the whole genre section — # chips AND the Any/All toggle — when it's empty, which would trap the user with no per-chip way to # undo a zero-result selection. Count 0 is fine (genre chips don't display counts). for g in _csv(f.genres): genre_counts.setdefault(g, 0) return { "genres": [{"value": g, "count": c} for g, c in sorted(genre_counts.items(), key=lambda kv: kv[0])], "content_ratings": [{"value": cr, "count": c} for cr, c in sorted(cr_counts.items(), key=lambda kv: -kv[1])], "year_min": min(years) if years else None, "year_max": max(years) if years else None, "rating_max": max(ratings) if ratings else None, "duration_min": min(durs) if durs else None, "duration_max": max(durs) if durs else None, } def _like_escape(s: str) -> str: return s.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") def _collection_editable(c: PlexCollection) -> bool: """Effective editability: the admin marked it editable AND it's a plain manual collection — never smart, never an external auto-list (IMDb/TMDb/…), which are managed outside Plex.""" return bool(c.editable) and not c.smart and _collection_source(c) == "collection" def _collection_card(c: PlexCollection) -> dict: return { "id": c.rating_key, "title": c.title, "summary": c.summary, "thumb": f"/api/plex/image/{c.rating_key}" if c.thumb_key else None, "child_count": c.child_count, "smart": c.smart, "source": _collection_source(c), "editable": c.editable, "can_edit": _collection_editable(c), # whether the edit controls should be offered (admin UI) } @router.get("/collections") def collections( library: str | None = None, q: str | None = None, user: User = Depends(current_user), db: Session = Depends(get_db), ) -> dict: """Mirrored collections as cards (most-populated first), optionally name-filtered by `q`. With a `library` plex_key → that one library's collections (used by the collection editor). WITHOUT one → the UNION across all enabled libraries (the unified-scope sidebar picker, which has no single library selected).""" if library is not None: lib = db.query(PlexLibrary).filter_by(plex_key=str(library)).first() if lib is None: return {"collections": []} query = db.query(PlexCollection).filter(PlexCollection.library_id == lib.id) else: enabled_ids = [lb.id for lb in db.query(PlexLibrary.id).filter_by(enabled=True)] if not enabled_ids: return {"collections": []} query = db.query(PlexCollection).filter(PlexCollection.library_id.in_(enabled_ids)) term = (q or "").strip() if term: query = query.filter(PlexCollection.title.ilike(f"%{_like_escape(term)}%")) rows = query.order_by( PlexCollection.child_count.desc().nullslast(), func.lower(PlexCollection.title) ).all() return {"collections": [_collection_card(c) for c in rows]} # --- Collection editing (P2, admin-only — writes back to Plex; changes are shared across all clients). def _editable_collection_or_403(db: Session, rating_key: str) -> PlexCollection: col = db.query(PlexCollection).filter_by(rating_key=str(rating_key)).first() if col is None: raise HTTPException(status_code=404, detail="Unknown collection") if not _collection_editable(col): raise HTTPException(status_code=403, detail="This collection is read-only") return col @router.post("/collections") def create_collection(payload: dict, _: User = Depends(admin_user), db: Session = Depends(get_db)) -> dict: """Create a manual collection (seeded with one item — Plex needs at least one). Editable by default.""" library = str(payload.get("library") or "") title = (payload.get("title") or "").strip() seed = str(payload.get("item_rating_key") or "") if not title or not seed: raise HTTPException(status_code=422, detail="title and item_rating_key are required") lib = db.query(PlexLibrary).filter_by(plex_key=library).first() if lib is None: raise HTTPException(status_code=404, detail="Unknown library") try: with PlexClient(db) as plex: new = plex.create_collection(lib.plex_key, title, seed) col = plex_sync.resync_collection(db, plex, lib, str(new.get("ratingKey"))) col.editable = True db.commit() except (PlexError, PlexNotConfigured) as e: db.rollback() raise HTTPException(status_code=502, detail=f"Plex write failed: {e}") return _collection_card(col) @router.post("/collections/{rating_key}/items/{item_rating_key}") def collection_add_item( rating_key: str, item_rating_key: str, _: User = Depends(admin_user), db: Session = Depends(get_db) ) -> dict: col = _editable_collection_or_403(db, rating_key) lib = db.get(PlexLibrary, col.library_id) try: with PlexClient(db) as plex: plex.add_collection_item(rating_key, str(item_rating_key)) col = plex_sync.resync_collection(db, plex, lib, rating_key) db.commit() except (PlexError, PlexNotConfigured) as e: db.rollback() raise HTTPException(status_code=502, detail=f"Plex write failed: {e}") return _collection_card(col) @router.delete("/collections/{rating_key}/items/{item_rating_key}") def collection_remove_item( rating_key: str, item_rating_key: str, _: User = Depends(admin_user), db: Session = Depends(get_db) ) -> dict: col = _editable_collection_or_403(db, rating_key) lib = db.get(PlexLibrary, col.library_id) try: with PlexClient(db) as plex: plex.remove_collection_item(rating_key, str(item_rating_key)) col = plex_sync.resync_collection(db, plex, lib, rating_key) db.commit() except (PlexError, PlexNotConfigured) as e: db.rollback() raise HTTPException(status_code=502, detail=f"Plex write failed: {e}") return _collection_card(col) @router.patch("/collections/{rating_key}") def collection_rename( rating_key: str, payload: dict, _: User = Depends(admin_user), db: Session = Depends(get_db) ) -> dict: col = _editable_collection_or_403(db, rating_key) title = (payload.get("title") or "").strip() if not title: raise HTTPException(status_code=422, detail="title is required") lib = db.get(PlexLibrary, col.library_id) try: with PlexClient(db) as plex: plex.rename_collection(lib.plex_key, rating_key, title) col = plex_sync.resync_collection(db, plex, lib, rating_key) db.commit() except (PlexError, PlexNotConfigured) as e: db.rollback() raise HTTPException(status_code=502, detail=f"Plex write failed: {e}") return _collection_card(col) @router.delete("/collections/{rating_key}") def collection_delete( rating_key: str, _: User = Depends(admin_user), db: Session = Depends(get_db) ) -> dict: col = _editable_collection_or_403(db, rating_key) lib = db.get(PlexLibrary, col.library_id) try: with PlexClient(db) as plex: plex.delete_collection(rating_key) plex_sync.delete_collection_local(db, lib, rating_key) db.commit() except (PlexError, PlexNotConfigured) as e: db.rollback() raise HTTPException(status_code=502, detail=f"Plex write failed: {e}") return {"deleted": rating_key} @router.post("/collections/{rating_key}/editable") def collection_set_editable( rating_key: str, payload: dict, _: User = Depends(admin_user), db: Session = Depends(get_db) ) -> dict: """Mark an existing Plex collection as 'mine to edit' (or unmark). Smart / auto-list collections (IMDb/TMDb/…) can never be made editable — they're regenerated outside Plex.""" col = db.query(PlexCollection).filter_by(rating_key=str(rating_key)).first() if col is None: raise HTTPException(status_code=404, detail="Unknown collection") want = bool(payload.get("editable")) if want and (col.smart or _collection_source(col) != "collection"): raise HTTPException(status_code=400, detail="Smart / auto-list collections can't be made editable") col.editable = want db.commit() return _collection_card(col) # --- Playlists (Siftlode-native, per-user, ordered — no Plex account needed; Plex sync is a later phase). def _own_playlist_or_404(db: Session, pid: int, user: User) -> PlexPlaylist: pl = db.query(PlexPlaylist).filter_by(id=pid, user_id=user.id).first() if pl is None: raise HTTPException(status_code=404, detail="Playlist not found") return pl def _playlist_card(pl: PlexPlaylist, count: int, thumb_rk: str | None) -> dict: return { "id": pl.id, "title": pl.title, "item_count": count, "thumb": f"/api/plex/image/{thumb_rk}" if thumb_rk else None, } def _playlist_item_query(db: Session, pid: int): return ( db.query(PlexPlaylistItem, PlexItem) .join(PlexItem, PlexItem.id == PlexPlaylistItem.item_id) .filter(PlexPlaylistItem.playlist_id == pid) .order_by(PlexPlaylistItem.position, PlexPlaylistItem.id) ) @router.get("/playlists") def playlists( contains: str | None = None, contains_group: str | None = None, user: User = Depends(current_user), db: Session = Depends(get_db), ) -> dict: """The current user's playlists (newest-touched first), each with an item count + a cover thumb. `contains`=an item rating_key adds `has_item` per playlist (for the single 'add to playlist' dialog). `contains_group`=a comma-separated list of rating_keys adds `group_in` per playlist (how many of the group it already holds) + a top-level `group_size`, for the 'add a whole season/show' dialog.""" contains_id = None if contains: row = db.query(PlexItem.id).filter_by(rating_key=str(contains)).first() contains_id = row[0] if row else None group_ids: set[int] | None = None if contains_group is not None: rks = [s for s in (contains_group or "").split(",") if s] rows = db.query(PlexItem.id).filter(PlexItem.rating_key.in_(rks or [""])).all() group_ids = {r[0] for r in rows} out = [] for pl in db.query(PlexPlaylist).filter_by(user_id=user.id).order_by(PlexPlaylist.updated_at.desc()): count = db.query(func.count(PlexPlaylistItem.id)).filter_by(playlist_id=pl.id).scalar() or 0 first = _playlist_item_query(db, pl.id).first() card = _playlist_card(pl, count, first[1].rating_key if first else None) if contains_id is not None: card["has_item"] = ( db.query(PlexPlaylistItem.id).filter_by(playlist_id=pl.id, item_id=contains_id).first() is not None ) if group_ids is not None: card["group_in"] = ( db.query(func.count(PlexPlaylistItem.id)) .filter(PlexPlaylistItem.playlist_id == pl.id, PlexPlaylistItem.item_id.in_(group_ids or [0])) .scalar() or 0 ) out.append(card) resp: dict = {"playlists": out} if group_ids is not None: resp["group_size"] = len(group_ids) return resp @router.post("/playlists") def create_playlist(payload: dict, user: User = Depends(current_user), db: Session = Depends(get_db)) -> dict: title = (payload.get("title") or "").strip() if not title: raise HTTPException(status_code=422, detail="title is required") pl = PlexPlaylist(user_id=user.id, title=title) db.add(pl) db.flush() seed = str(payload.get("item_rating_key") or "") count = 0 thumb_rk = None if seed: it = db.query(PlexItem).filter_by(rating_key=seed).first() if it is not None: db.add(PlexPlaylistItem(playlist_id=pl.id, item_id=it.id, position=0)) count, thumb_rk = 1, it.rating_key db.commit() return _playlist_card(pl, count, thumb_rk) @router.get("/playlists/{pid}") def playlist_detail(pid: int, user: User = Depends(current_user), db: Session = Depends(get_db)) -> dict: pl = _own_playlist_or_404(db, pid, user) rows = _playlist_item_query(db, pid).all() items = [it for (_pi, it) in rows] states = { s.item_id: s for s in db.query(PlexState).filter( PlexState.user_id == user.id, PlexState.item_id.in_([it.id for it in items] or [0]) ) } shows = { sh.id: sh for sh in db.query(PlexShow).filter(PlexShow.id.in_([it.show_id for it in items if it.show_id] or [0])) } cards = [ _leaf_card( it, states.get(it.id), shows[it.show_id].title if it.show_id in shows else None, shows[it.show_id].rating_key if it.show_id in shows else None, ) for it in items ] return {"id": pl.id, "title": pl.title, "items": cards} @router.patch("/playlists/{pid}") def rename_playlist(pid: int, payload: dict, user: User = Depends(current_user), db: Session = Depends(get_db)) -> dict: pl = _own_playlist_or_404(db, pid, user) title = (payload.get("title") or "").strip() if not title: raise HTTPException(status_code=422, detail="title is required") pl.title = title db.commit() count = db.query(func.count(PlexPlaylistItem.id)).filter_by(playlist_id=pl.id).scalar() or 0 first = _playlist_item_query(db, pl.id).first() return _playlist_card(pl, count, first[1].rating_key if first else None) @router.delete("/playlists/{pid}") def delete_playlist(pid: int, user: User = Depends(current_user), db: Session = Depends(get_db)) -> dict: pl = _own_playlist_or_404(db, pid, user) db.delete(pl) db.commit() return {"deleted": pid} @router.post("/playlists/{pid}/items") def playlist_add_item(pid: int, payload: dict, user: User = Depends(current_user), db: Session = Depends(get_db)) -> dict: pl = _own_playlist_or_404(db, pid, user) it = db.query(PlexItem).filter_by(rating_key=str(payload.get("item_rating_key") or "")).first() if it is None: raise HTTPException(status_code=404, detail="Unknown item") exists = db.query(PlexPlaylistItem).filter_by(playlist_id=pid, item_id=it.id).first() if exists is None: maxpos = db.query(func.coalesce(func.max(PlexPlaylistItem.position), -1)).filter_by(playlist_id=pid).scalar() db.add(PlexPlaylistItem(playlist_id=pid, item_id=it.id, position=int(maxpos) + 1)) pl.updated_at = datetime.now(timezone.utc) db.commit() return {"ok": True} @router.delete("/playlists/{pid}/items/{item_rating_key}") def playlist_remove_item(pid: int, item_rating_key: str, user: User = Depends(current_user), db: Session = Depends(get_db)) -> dict: pl = _own_playlist_or_404(db, pid, user) it = db.query(PlexItem).filter_by(rating_key=str(item_rating_key)).first() if it is not None: db.query(PlexPlaylistItem).filter_by(playlist_id=pid, item_id=it.id).delete() pl.updated_at = datetime.now(timezone.utc) db.commit() return {"ok": True} def _items_by_rks(db: Session, rks: list) -> list[PlexItem]: """Resolve rating_keys to PlexItems, preserving the input order (drops unknown keys).""" order = [str(k) for k in rks if str(k)] if not order: return [] by_rk = {it.rating_key: it for it in db.query(PlexItem).filter(PlexItem.rating_key.in_(order))} seen: set[str] = set() out: list[PlexItem] = [] for rk in order: it = by_rk.get(rk) if it is not None and rk not in seen: seen.add(rk) out.append(it) return out @router.post("/playlists/{pid}/items/bulk") def playlist_add_bulk(pid: int, payload: dict, user: User = Depends(current_user), db: Session = Depends(get_db)) -> dict: """Append many items at once (a whole season/show). Already-present items are skipped; the input order is preserved for the newly-added ones. Returns how many were added + the new total.""" pl = _own_playlist_or_404(db, pid, user) items = _items_by_rks(db, payload.get("item_rating_keys") or []) existing = {row[0] for row in db.query(PlexPlaylistItem.item_id).filter_by(playlist_id=pid)} pos = int(db.query(func.coalesce(func.max(PlexPlaylistItem.position), -1)).filter_by(playlist_id=pid).scalar()) added = 0 for it in items: if it.id in existing: continue pos += 1 db.add(PlexPlaylistItem(playlist_id=pid, item_id=it.id, position=pos)) existing.add(it.id) added += 1 if added: pl.updated_at = datetime.now(timezone.utc) db.commit() total = db.query(func.count(PlexPlaylistItem.id)).filter_by(playlist_id=pid).scalar() or 0 return {"added": added, "item_count": total} @router.post("/playlists/{pid}/items/remove-bulk") def playlist_remove_bulk(pid: int, payload: dict, user: User = Depends(current_user), db: Session = Depends(get_db)) -> dict: """Remove many items at once (a whole season/show, or a multi-select). Returns how many were removed + the new total.""" pl = _own_playlist_or_404(db, pid, user) ids = [it.id for it in _items_by_rks(db, payload.get("item_rating_keys") or [])] removed = 0 if ids: removed = ( db.query(PlexPlaylistItem) .filter(PlexPlaylistItem.playlist_id == pid, PlexPlaylistItem.item_id.in_(ids)) .delete(synchronize_session=False) ) if removed: pl.updated_at = datetime.now(timezone.utc) db.commit() total = db.query(func.count(PlexPlaylistItem.id)).filter_by(playlist_id=pid).scalar() or 0 return {"removed": removed, "item_count": total} @router.put("/playlists/{pid}/order") def reorder_playlist(pid: int, payload: dict, user: User = Depends(current_user), db: Session = Depends(get_db)) -> dict: """Set the playlist order from an explicit list of item rating_keys (0-based positions).""" pl = _own_playlist_or_404(db, pid, user) order = [str(k) for k in (payload.get("item_rating_keys") or [])] id_by_rk = {it.rating_key: it.id for it in db.query(PlexItem).filter(PlexItem.rating_key.in_(order or [""]))} for pos, rk in enumerate(order): iid = id_by_rk.get(rk) if iid is not None: db.query(PlexPlaylistItem).filter_by(playlist_id=pid, item_id=iid).update({"position": pos}) pl.updated_at = datetime.now(timezone.utc) db.commit() return {"ok": True} # Live-Plex enrichment for a show (cast/IMDb + related) is user-independent raw payload, so it's cached # per rating_key for a short TTL and shared across users; per-user shaping happens in the caller. Bounds # staleness to the TTL while turning repeat opens of the same show into zero network calls. _SHOW_ENRICH_TTL_S = 300 _show_enrich_cache: dict[str, tuple[float, dict, list]] = {} _show_enrich_lock = threading.Lock() def _show_enrichment(db: Session, rating_key: str) -> tuple[dict, list]: """(meta, related_raw) for a show's live Plex enrichment — served from the per-show TTL cache when warm, otherwise fetched with the two Plex calls (metadata + related) run IN PARALLEL. Returns ({}, []) when Plex is unavailable/unconfigured (not cached, so it retries on the next open).""" now = time.monotonic() with _show_enrich_lock: hit = _show_enrich_cache.get(rating_key) if hit is not None and hit[0] > now: return hit[1], hit[2] try: with PlexClient(db) as plex: with ThreadPoolExecutor(max_workers=2) as ex: meta_f = ex.submit(plex.metadata, rating_key) related_f = ex.submit(plex.related, rating_key) # swallows its own errors → [] meta = meta_f.result() or {} related_raw = related_f.result() except (PlexError, PlexNotConfigured): return {}, [] # Don't cache an EMPTY related list: plex.related() swallows a transient failure as [] (same as a # genuinely no-related show), so caching it would serve an empty "related" section for the whole TTL # even after Plex recovers. Skip the cache in that case (retry next open, as before); shows that DO # have related items — the payload worth caching — still get cached. if related_raw: with _show_enrich_lock: _show_enrich_cache[rating_key] = (now + _SHOW_ENRICH_TTL_S, meta, related_raw) return meta, related_raw @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] = {} all_cards: list[dict] = [] # every episode, in season/episode order → show-level rollup for e in eps: card = _episode_card(e, states.get(e.id)) by_season.setdefault(e.season_id, []).append(card) all_cards.append(card) show_roll = _rollup(all_cards) # Cast + IMDb (best-effort live metadata; same shape as the movie/episode info page). The two live # Plex calls are cached per-show + fetched in parallel by _show_enrichment; the per-user shaping # (_related_show_cards adds this user's watch-state) stays out of the shared cache. rich: dict = {} related: list[dict] = [] try: meta, related_raw = _show_enrichment(db, sh.rating_key) rich = _rich_meta(meta) related = _related_show_cards(db, user.id, related_raw, exclude=sh.id) except Exception: # Best-effort enrichment only — the local episode list is already assembled, so any live-Plex # failure (not configured, HTTP/JSON shape, _rich_meta parse error) must degrade, not 500. pass 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", "library": _lib_key(db, sh.library_id), "content_rating": sh.content_rating, "rating": sh.rating, "genres": sh.genres or [], "studio": sh.studio, "season_count": sh.child_count, "imdb_rating": rich.get("imdb_rating"), "imdb_id": rich.get("imdb_id"), "imdb_url": rich.get("imdb_url"), "cast": rich.get("cast", []), # Show-level rollup for the hero action buttons. "status": show_roll["status"], "resume": show_roll["resume"], "first": show_roll["first"], "episode_count": show_roll["episode_count"], "collection_keys": sh.collection_keys or [], }, "seasons": [ _season_block(se, by_season.get(se.id, [])) for se in seasons ], "related": related, } def _season_block(se: PlexSeason, cards: list[dict]) -> dict: """A season's card for the show page (with its aggregate rollup) plus its episodes (used by the season subpage, read from the same cached show-detail payload).""" roll = _rollup(cards) return { "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}", "episode_count": roll["episode_count"], "status": roll["status"], "resume": roll["resume"], "first": roll["first"], "episodes": cards, } def _related_show_cards(db: Session, user_id: int, related: list[dict], exclude: int) -> list[dict]: """Map Plex 'related' metadata to show cards for the shows we actually mirror (so they're openable), excluding the current show. Order preserved; capped.""" rks = [str(m.get("ratingKey")) for m in related if m.get("ratingKey")] if not rks: return [] shows = { s.rating_key: s for s in db.query(PlexShow).filter(PlexShow.rating_key.in_(rks[:60]), PlexShow.id != exclude) } statuses = _show_state_map(db, user_id, [s.id for s in shows.values()]) out: list[dict] = [] seen: set[str] = set() for rk in rks: sh = shows.get(rk) if sh is None or rk in seen: continue seen.add(rk) out.append(_show_card(sh, statuses.get(sh.id, "new"))) if len(out) >= 20: break return out 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 col = db.query(PlexCollection).filter_by(rating_key=rating_key).first() if col is not None: return col.thumb_key return None # --- Thin on-disk image cache ----------------------------------------------------------------- # Posters/art/cast-photos are proxied on demand; without a cache every scroll re-hits Plex over the # LAN (janky). We cache the fetched bytes on local disk keyed by a stable id, so repeat views (and # other users) serve instantly from disk. On-demand + bounded by the library size; no eviction (a # few hundred MB at most for a big library — acceptable, matching the download disk trade-offs). _IMG_CACHE = Path(settings.download_root) / ".plex-img-cache" _EXT_CT = [("jpg", "image/jpeg"), ("png", "image/png"), ("webp", "image/webp")] _CT_EXT = {ct: ext for ext, ct in _EXT_CT} # Target box per variant — Plex resizes to these server-side. Posters fill ≤176px cells (grid/info/ # strips), so 400px covers 2× DPR; the faint backdrop art only needs ~1280px. Keeps images ~tens of KB. _IMG_SIZES = {"thumb": (400, 600), "art": (1280, 720)} # Image-based subtitle codecs (Blu-ray/DVD) — can't be turned into text WebVTT (would need burn-in). _BITMAP_SUBS = {"pgs", "hdmv_pgs_subtitle", "dvd_subtitle", "dvdsub", "vobsub", "dvbsub", "dvb_subtitle"} def _srt_to_vtt(text: str) -> str: """Minimal SubRip→WebVTT: add the header and switch the ms separator (`,`→`.`). SRT cue-index lines are valid in WebVTT, so they're left as-is.""" body = re.sub(r"(\d\d:\d\d:\d\d),(\d\d\d)", r"\1.\2", text.replace("\r\n", "\n").lstrip("")) return "WEBVTT\n\n" + body.strip() + "\n" def _vtt_ts_to_s(ts: str) -> float: parts = ts.split(":") h = "0" if len(parts) == 2 else parts[0] m, rest = parts[-2], parts[-1] sec, _, ms = rest.partition(".") return int(h) * 3600 + int(m) * 60 + int(sec) + int(ms or 0) / 1000.0 def _vtt_s_to_ts(x: float) -> str: x = max(0.0, x) h = int(x // 3600) x -= h * 3600 m = int(x // 60) x -= m * 60 s = int(x) ms = int(round((x - s) * 1000)) if ms >= 1000: s += 1 ms -= 1000 return f"{h:02d}:{m:02d}:{s:02d}.{ms:03d}" def _shift_vtt(vtt: str, offset: float) -> str: """Subtract `offset` seconds from every cue's start/end. WebVTT cues carry ABSOLUTE content times, but an HLS remux session's clock is zero-based at the real keyframe (see stream.py media_start_s), so a subtitle overlaid on that session must be shifted by that offset to line up with speech. Processes cue BLOCKS so a cue that shifts fully into the past (end <= 0) can be DROPPED entirely. (Collapsing them to 00:00:00.000-->00:00:00.000 made EVERY past cue count as active at currentTime 0 — where a resume/seek lands — piling all of them on screen until playback advanced past 0.) A cue straddling 0 is clamped to start at 0.""" if not offset: return vtt out_blocks: list[str] = [] for block in re.split(r"\n[ \t]*\n", vtt): if "-->" not in block: out_blocks.append(block) # header / NOTE / STYLE — keep verbatim continue lines = block.split("\n") keep = True for i, ln in enumerate(lines): if "-->" not in ln: continue left, _, right = ln.partition("-->") rparts = right.strip().split(None, 1) settings = (" " + rparts[1]) if len(rparts) > 1 else "" try: st = _vtt_ts_to_s(left.strip()) - offset en = _vtt_ts_to_s(rparts[0]) - offset except (ValueError, IndexError): break # unparseable timing → leave the block as-is if en <= 0: keep = False # whole cue is before the session start → drop the block else: lines[i] = f"{_vtt_s_to_ts(st)} --> {_vtt_s_to_ts(en)}{settings}" # _vtt_s_to_ts clamps st>=0 break if keep: out_blocks.append("\n".join(lines)) return "\n\n".join(out_blocks) def _ffmpeg_to_vtt(src: str, sub_ord: int | None) -> str: """Convert one subtitle stream of a file to WebVTT via ffmpeg (for embedded subs, and non-SRT external subs written to a temp file). `sub_ord` = ordinal among the file's subtitle streams.""" cmd = ["ffmpeg", "-nostdin", "-loglevel", "error", "-i", src, "-map", f"0:s:{sub_ord if sub_ord is not None else 0}", "-f", "webvtt", "pipe:1"] r = subprocess.run(cmd, capture_output=True) if r.returncode != 0 or not r.stdout: raise PlexError((r.stderr or b"").decode("utf-8", "replace")[:200] or "ffmpeg subtitle convert failed") return r.stdout.decode("utf-8", "replace") def _sub_cache_read(key: str) -> str | None: try: p = _IMG_CACHE / f"{key}.vtt" return p.read_text(encoding="utf-8") if p.exists() else None except OSError: return None def _sub_cache_write(key: str, vtt: str) -> None: try: _IMG_CACHE.mkdir(parents=True, exist_ok=True) tmp = _IMG_CACHE / f"{key}.vtt.tmp" tmp.write_text(vtt, encoding="utf-8") tmp.replace(_IMG_CACHE / f"{key}.vtt") # atomic publish except OSError: pass def _img_cache_read(key: str) -> tuple[bytes, str] | None: for ext, ct in _EXT_CT: p = _IMG_CACHE / f"{key}.{ext}" try: if p.exists() and p.stat().st_size > 0: return p.read_bytes(), ct except OSError: pass return None def _img_cache_write(key: str, data: bytes, ct: str) -> None: ext = _CT_EXT.get((ct or "").split(";")[0].strip(), "jpg") try: _IMG_CACHE.mkdir(parents=True, exist_ok=True) tmp = _IMG_CACHE / f"{key}.{ext}.tmp" tmp.write_bytes(data) tmp.replace(_IMG_CACHE / f"{key}.{ext}") # atomic publish except OSError: pass # cache is best-effort def _img_response(data: bytes, ct: str) -> Response: return Response(content=data, media_type=ct, headers={"Cache-Control": "public, max-age=86400"}) 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 = _IMG_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 = _img_cache_read(cache_key) if hit: return _img_response(*hit) # Cold miss: short-lived session for the key + Plex client config, released before the slow fetch. try: with SessionLocal() as db: key = _image_key(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}") _img_cache_write(cache_key, data, ct) return _img_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 = _sub_cache_read(cache_key) if hit is not None: return Response(_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 _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 = _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 = _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 = _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}") _sub_cache_write(cache_key, vtt) return Response(_shift_vtt(vtt, offset), media_type="text/vtt") # --- 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, 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(_PERSON_IMG_HOST): raise HTTPException(status_code=400, detail="Unsupported image host") cache_key = "person_" + hashlib.sha1(u.encode()).hexdigest() hit = _img_cache_read(cache_key) if hit: return _img_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") _img_cache_write(cache_key, r.content, ct) return _img_response(r.content, ct) # Known external list providers that seed Plex collections (via plextraktsync et al). A collection's # "source" buckets the info-page strips so the user can show/hide each type independently (e.g. hide # the noisy 'IMDb Popular'/'TMDb Trending' auto-lists but keep genuine franchise collections). Derived # from the already-synced title/smart — no schema change, no re-sync. Unknown providers stay "collection". _STRIP_PROVIDERS = ("imdb", "tmdb", "tvdb", "trakt") def _collection_source(col: PlexCollection) -> str: first = (col.title or "").strip().split(" ", 1)[0].lower() if first in _STRIP_PROVIDERS: return first if col.smart: return "smart" return "collection" def _collection_strips(db: Session, it: PlexItem, user_id: int) -> list[dict]: """The collections this movie belongs to, each with its member titles (playable cards) — the 'Avatar Collection' strip on the info page. Pure local lookup (no Plex call).""" keys = it.collection_keys or [] if not keys or it.kind != "movie": return [] cols = {c.rating_key: c for c in db.query(PlexCollection).filter(PlexCollection.rating_key.in_(keys))} # Smallest (most specific — e.g. a franchise) first; the big auto-lists (IMDb/TMDb Popular) last. ordered = sorted((c for c in cols.values()), key=lambda c: (c.child_count or 0, c.title)) out = [] for col in ordered: rk = col.rating_key members = ( db.query(PlexItem) .filter( PlexItem.library_id == it.library_id, PlexItem.kind == "movie", PlexItem.collection_keys.contains([rk]), ) .order_by(PlexItem.year.asc().nullslast(), func.lower(PlexItem.title)) .limit(60) .all() ) ids = [m.id for m in members] states = { s.item_id: s for s in db.query(PlexState).filter( PlexState.user_id == user_id, PlexState.item_id.in_(ids or [0]) ) } out.append( { "id": rk, "title": col.title, "source": _collection_source(col), "items": [_movie_card(m, states.get(m.id)) for m in members], } ) return out 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: codec = (s.get("codec") or s.get("format") or "").lower() 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"), "codec": codec, # Text subs (srt/ass/…) can be served as WebVTT tracks; image subs (PGS/VobSub) # can't be turned into text, so the player hides them (burn-in is a later phase). "text": codec not in _BITMAP_SUBS, } ) except Exception: pass # metadata extras are best-effort; any live-Plex failure must degrade, not 500 — 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", "library": _lib_key(db, it.library_id), "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, "collections": _collection_strips(db, it, user.id), } 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, ) -> 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.""" 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.query(PlexState).filter_by(user_id=user_id, item_id=item_id).first() background.add_task( plex_watch.push_state_to_plex, user_id, item_id, rating_key, action, time_ms, duration_ms, plex_watch._state_marker(st), ) @router.post("/item/{rating_key}/progress") def item_progress( rating_key: str, payload: dict, 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). `final` (sent on the pause/pagehide/unmount flush, not the periodic 10s checkpoint) is what debounces the Plex resume push: we mirror the settled position to Plex only on a final flush, so a single watch doesn't spray `/:/timeline` calls at the server every 10 seconds.""" 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") final = bool(payload.get("final")) now = datetime.now(timezone.utc) st = db.query(PlexState).filter_by(user_id=user.id, item_id=it.id).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 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 if final: st.synced_to_plex = False db.commit() if final: _push_watch( background, db, user.id, it.id, rating_key, "resume", position * 1000, duration * 1000 ) return {"position_seconds": position} @router.post("/item/{rating_key}/state") def item_state( rating_key: str, payload: dict, 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 = 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() 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.query(PlexState).filter( PlexState.user_id == user_id, PlexState.item_id.in_([e.id for e in eps] or [0]) ) } 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) @router.post("/show/{rating_key}/state") def show_state( rating_key: str, payload: dict, 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 = db.query(PlexShow).filter_by(rating_key=str(rating_key)).first() if sh is None: raise HTTPException(status_code=404, detail="Unknown Plex show") watched = bool(payload.get("watched")) eps = db.query(PlexItem).filter_by(show_id=sh.id, kind="episode").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, payload: dict, 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 = db.query(PlexSeason).filter_by(rating_key=str(rating_key)).first() if se is None: raise HTTPException(status_code=404, detail="Unknown Plex season") watched = bool(payload.get("watched")) eps = db.query(PlexItem).filter_by(season_id=se.id, kind="episode").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); transcode (HEVC/VP9) is P3. 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) if it.playable == "transcode": raise HTTPException(status_code=501, detail="This format needs transcoding (a later phase).") # `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) 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. return {"mode": "hls", "start_s": s.media_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 (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"} )