diff --git a/backend/app/routes/plex/__init__.py b/backend/app/routes/plex/__init__.py new file mode 100644 index 00000000..6bfe2fae --- /dev/null +++ b/backend/app/routes/plex/__init__.py @@ -0,0 +1,13 @@ +"""Plex route package — assembles the admin/browse/playback sub-routers. + +Split out of the former single routes/plex.py (R12 S1). `router` keeps the same shape the app +wiring expects (`plex_routes.router`), so main.py is unchanged. +""" +from fastapi import APIRouter + +from app.routes.plex import admin, browse, playback + +router = APIRouter() +router.include_router(admin.router) +router.include_router(browse.router) +router.include_router(playback.router) diff --git a/backend/app/routes/plex/_shared.py b/backend/app/routes/plex/_shared.py new file mode 100644 index 00000000..c56f9d24 --- /dev/null +++ b/backend/app/routes/plex/_shared.py @@ -0,0 +1,16 @@ +"""Shared helpers, serializers and constants for the plex route package.""" + +import logging +from fastapi import HTTPException +from sqlalchemy.orm import Session +from app.models import ( + PlexItem, +) + +log = logging.getLogger("siftlode.plex") + +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 diff --git a/backend/app/routes/plex/admin.py b/backend/app/routes/plex/admin.py new file mode 100644 index 00000000..55a64020 --- /dev/null +++ b/backend/app/routes/plex/admin.py @@ -0,0 +1,143 @@ +"""Plex admin: connection test, catalog sync, and owner watch-link config.""" + +from datetime import datetime, timezone +from fastapi import APIRouter, Depends, HTTPException +from pydantic import BaseModel +from sqlalchemy.orm import Session +from app import sysconfig +from app.auth import current_user +from app.db import get_db +from app.models import ( + PlexItem, + PlexLink, + User, +) +from app.plex import paths as plex_paths +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.utils import iso + +router = APIRouter(prefix="/api/plex", tags=["plex"]) + +@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) + +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": iso(link.last_watch_sync_at) if link 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) + +class WatchLinkIn(BaseModel): + enabled: bool = False + +@router.post("/watch/link") +def watch_link_set( + body: WatchLinkIn, + 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 = body.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} diff --git a/backend/app/routes/plex.py b/backend/app/routes/plex/browse.py similarity index 66% rename from backend/app/routes/plex.py rename to backend/app/routes/plex/browse.py index 63a46821..0315e192 100644 --- a/backend/app/routes/plex.py +++ b/backend/app/routes/plex/browse.py @@ -1,35 +1,21 @@ -"""Plex integration API. +"""Plex browse: libraries, facets, collections, playlists, show/item detail, people.""" -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 tempfile import threading import time from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass from datetime import datetime, timedelta, timezone - -import httpx -from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query, Request, Response -from fastapi.responses import FileResponse +from fastapi import APIRouter, Depends, HTTPException, Query from pydantic import BaseModel 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.db import SessionLocal, get_db +from app.auth import current_user +from app.db import get_db from app.models import ( PlexCollection, PlexItem, PlexLibrary, - PlexLink, PlexPerson, PlexPlaylist, PlexPlaylistItem, @@ -38,167 +24,22 @@ from app.models import ( PlexState, User, ) -from app.plex import paths as plex_paths from app.plex import people as plex_people -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.plex import imgcache, rollup, subtitles +from app.plex import rollup, subtitles from app.routes._common import owned_or_404 from app.routes.admin import admin_user from app.routes.feed import _to_tsquery_str -from app.utils import iso - -log = logging.getLogger("siftlode.plex") router = APIRouter(prefix="/api/plex", tags=["plex"]) +from app.routes.plex._shared import ( + _item_or_404, +) + _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": iso(link.last_watch_sync_at) if link 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) - - -class WatchLinkIn(BaseModel): - enabled: bool = False - - -@router.post("/watch/link") -def watch_link_set( - body: WatchLinkIn, - 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 = body.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).""" @@ -212,7 +53,6 @@ def list_libraries(user: User = Depends(current_user), db: Session = Depends(get 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, @@ -226,7 +66,6 @@ def _movie_card(it: PlexItem, st: PlexState | None) -> dict: "position_seconds": st.position_seconds if st else 0, } - def _show_card(sh: PlexShow, status: str = "new") -> dict: return { "id": sh.rating_key, @@ -239,7 +78,6 @@ def _show_card(sh: PlexShow, status: str = "new") -> dict: "status": status, } - 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).""" @@ -248,7 +86,6 @@ def _lib_key(db: Session, library_id: int | None) -> str | 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, @@ -264,7 +101,6 @@ def _episode_card(e: PlexItem, st: PlexState | None) -> dict: "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: @@ -277,19 +113,15 @@ def _leaf_card( 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 @@ -324,12 +156,10 @@ def _meta_filter_conds(model, p: dict) -> list: 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 @@ -362,7 +192,6 @@ class LibraryFilters: "studios": self.studios, "collection": self.collection, } - @router.get("/library") def unified_library( scope: str = "both", @@ -531,7 +360,6 @@ def unified_library( return {"scope": scope, "total": total, "offset": offset, "limit": limit, "items": items, "episodes": episodes} - @router.get("/facets") def facets( scope: str = "both", @@ -674,17 +502,14 @@ def facets( "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, @@ -698,7 +523,6 @@ def _collection_card(c: PlexCollection) -> dict: "can_edit": _collection_editable(c), # whether the edit controls should be offered (admin UI) } - @router.get("/collections") def collections( library: str | None = None, @@ -728,8 +552,6 @@ def collections( ).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: @@ -738,13 +560,11 @@ def _editable_collection_or_403(db: Session, rating_key: str) -> PlexCollection: raise HTTPException(status_code=403, detail="This collection is read-only") return col - class CreateCollectionIn(BaseModel): library: str = "" title: str = "" item_rating_key: str = "" - @router.post("/collections") def create_collection( body: CreateCollectionIn, _: User = Depends(admin_user), db: Session = Depends(get_db) @@ -769,7 +589,6 @@ def create_collection( raise HTTPException(status_code=422, 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) @@ -786,7 +605,6 @@ def collection_add_item( raise HTTPException(status_code=422, 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) @@ -803,11 +621,9 @@ def collection_remove_item( raise HTTPException(status_code=422, detail=f"Plex write failed: {e}") return _collection_card(col) - class CollectionRenameIn(BaseModel): title: str = "" - @router.patch("/collections/{rating_key}") def collection_rename( rating_key: str, body: CollectionRenameIn, _: User = Depends(admin_user), db: Session = Depends(get_db) @@ -827,7 +643,6 @@ def collection_rename( raise HTTPException(status_code=422, 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) @@ -844,11 +659,9 @@ def collection_delete( raise HTTPException(status_code=422, detail=f"Plex write failed: {e}") return {"deleted": rating_key} - class CollectionEditableIn(BaseModel): editable: bool = False - @router.post("/collections/{rating_key}/editable") def collection_set_editable( rating_key: str, body: CollectionEditableIn, _: User = Depends(admin_user), db: Session = Depends(get_db) @@ -865,12 +678,9 @@ def collection_set_editable( 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: return owned_or_404(db, PlexPlaylist, pid, user, detail="Playlist not found") - def _playlist_card(pl: PlexPlaylist, count: int, thumb_rk: str | None) -> dict: return { "id": pl.id, @@ -879,7 +689,6 @@ def _playlist_card(pl: PlexPlaylist, count: int, thumb_rk: str | None) -> dict: "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) @@ -888,7 +697,6 @@ def _playlist_item_query(db: Session, pid: int): .order_by(PlexPlaylistItem.position, PlexPlaylistItem.id) ) - @router.get("/playlists") def playlists( contains: str | None = None, @@ -932,12 +740,10 @@ def playlists( resp["group_size"] = len(group_ids) return resp - class CreatePlexPlaylistIn(BaseModel): title: str = "" item_rating_key: str = "" - @router.post("/playlists") def create_playlist( body: CreatePlexPlaylistIn, user: User = Depends(current_user), db: Session = Depends(get_db) @@ -959,7 +765,6 @@ def create_playlist( 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) @@ -986,11 +791,9 @@ def playlist_detail(pid: int, user: User = Depends(current_user), db: Session = ] return {"id": pl.id, "title": pl.title, "items": cards} - class RenamePlexPlaylistIn(BaseModel): title: str = "" - @router.patch("/playlists/{pid}") def rename_playlist( pid: int, body: RenamePlexPlaylistIn, user: User = Depends(current_user), db: Session = Depends(get_db) @@ -1005,7 +808,6 @@ def rename_playlist( 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) @@ -1013,11 +815,9 @@ def delete_playlist(pid: int, user: User = Depends(current_user), db: Session = db.commit() return {"deleted": pid} - class PlexItemRefIn(BaseModel): item_rating_key: str = "" - @router.post("/playlists/{pid}/items") def playlist_add_item( pid: int, body: PlexItemRefIn, user: User = Depends(current_user), db: Session = Depends(get_db) @@ -1034,7 +834,6 @@ def playlist_add_item( 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) @@ -1045,11 +844,9 @@ def playlist_remove_item(pid: int, item_rating_key: str, user: User = Depends(cu db.commit() return {"ok": True} - class PlexItemKeysIn(BaseModel): item_rating_keys: list[str] = [] - 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)] @@ -1065,7 +862,6 @@ def _items_by_rks(db: Session, rks: list) -> list[PlexItem]: out.append(it) return out - @router.post("/playlists/{pid}/items/bulk") def playlist_add_bulk( pid: int, body: PlexItemKeysIn, user: User = Depends(current_user), db: Session = Depends(get_db) @@ -1090,7 +886,6 @@ def playlist_add_bulk( 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, body: PlexItemKeysIn, user: User = Depends(current_user), db: Session = Depends(get_db) @@ -1112,7 +907,6 @@ def playlist_remove_bulk( 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, body: PlexItemKeysIn, user: User = Depends(current_user), db: Session = Depends(get_db) @@ -1129,14 +923,11 @@ def reorder_playlist( 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() +_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 @@ -1165,7 +956,6 @@ def _show_enrichment(db: Session, rating_key: str) -> tuple[dict, list]: _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, @@ -1256,7 +1046,6 @@ def show_detail( "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).""" @@ -1273,7 +1062,6 @@ def _season_block(se: PlexSeason, cards: list[dict]) -> dict: "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.""" @@ -1297,124 +1085,6 @@ def _related_show_cards(db: Session, user_id: int, related: list[dict], exclude: break return out - -def _require_session(request: Request) -> int: - """Lightweight auth for the high-fan-out image proxies: validate the signed session cookie WITHOUT - a DB user-load, so these endpoints never hold a pool connection for the auth. Poster images are - low-sensitivity; a valid session is enough.""" - uid, _ = resolved_user_id(request) - if not uid: - raise HTTPException(status_code=401, detail="Not authenticated") - return uid - - -@router.get("/image/{rating_key}") -def image(rating_key: str, request: Request, variant: str = "thumb") -> Response: - """Proxy a Plex poster/art image (keeps the admin token server-side; no image duplication). A - thin on-disk cache serves repeat views without re-hitting Plex. - - ⚠️ This endpoint is called in BURSTS (a collections page fires 100+ concurrent poster requests), - so it must NOT hold a DB connection during the slow Plex fetch — doing so exhausted the pool - (QueuePool timeout, broken images on prod). Warm path (cache hit) touches NO DB; cold path opens a - brief session only to resolve the image key + read the Plex config, then RELEASES it before the - fetch. Auth is the signed-session check (no DB user-load).""" - _require_session(request) - v = "art" if variant == "art" else "thumb" - w, h = imgcache.SIZES[v] - # Size is part of the cache key: Plex resizes server-side (see PlexClient.image_bytes), so we cache - # the small variant. The width tag also means changing a target size just re-caches, never serves a - # stale differently-sized file (old full-size files under the un-tagged key are simply orphaned). - cache_key = f"item_{rating_key}_{v}_{w}" - hit = imgcache.read(cache_key) - if hit: - return imgcache.response(*hit) - # Cold miss: short-lived session for the key + Plex client config, released before the slow fetch. - try: - with SessionLocal() as db: - key = imgcache.key_for(db, str(rating_key), v) - if not key: - raise HTTPException(status_code=404, detail="No image") - plex = PlexClient(db) # reads base URL + token now; the fetch below needs no DB - except PlexNotConfigured as e: - raise HTTPException(status_code=400, detail=str(e)) - try: - with plex: - data, ct = plex.image_bytes(key, w, h) - except PlexError as e: - raise HTTPException(status_code=502, detail=f"Plex image fetch failed: {e}") - imgcache.write(cache_key, data, ct) - return imgcache.response(data, ct) - - -@router.get("/subtitle/{rating_key}/{sub_ord}") -def subtitle(rating_key: str, sub_ord: int, request: Request, offset: float = 0.0) -> Response: - """Serve the ord-th subtitle track as WebVTT for the browser's — so choosing a subtitle - overlays it directly (no HLS re-mux / session restart). Handles EXTERNAL sidecar subs (fetched - from Plex via the stream `key` — the old `-map 0:s` broke on these since they're not in the file) - AND embedded text subs (extracted from the local file). Image subs (PGS/VobSub) → 415. Disk-cached.""" - _require_session(request) - # Cache the ABSOLUTE-time VTT once (per rk+ord); the per-session cue-shift is applied on the way - # out, so a resume/seek to any offset reuses the same cached track (no cache-per-offset explosion). - cache_key = f"sub_{rating_key}_{sub_ord}" - hit = subtitles.cache_read(cache_key) - if hit is not None: - return Response(subtitles.shift_vtt(hit, offset), media_type="text/vtt") - try: - with SessionLocal() as db: - it = _item_or_404(db, rating_key) - with PlexClient(db) as plex: - part = ((plex.metadata(it.rating_key) or {}).get("Media") or [{}])[0].get("Part") or [{}] - subs = [s for s in (part[0].get("Stream") or []) if s.get("streamType") == 3] - if sub_ord < 0 or sub_ord >= len(subs): - raise HTTPException(status_code=404, detail="No such subtitle") - s = subs[sub_ord] - codec = (s.get("codec") or s.get("format") or "").lower() - if codec in subtitles.BITMAP_SUBS: - raise HTTPException(status_code=415, detail="Image-based subtitle can't be shown as text") - if s.get("key"): # external sidecar → fetch from Plex, convert to VTT - raw = plex.raw_get(s["key"]) - if codec in ("srt", "subrip", ""): - vtt = subtitles.srt_to_vtt(raw.decode("utf-8", "replace")) - else: - fd, tmp = tempfile.mkstemp(suffix="." + (codec or "sub")) - try: - os.write(fd, raw) - os.close(fd) - vtt = subtitles.ffmpeg_to_vtt(tmp, None) - finally: - os.unlink(tmp) - elif s.get("index") is not None: # embedded → extract from the local file - src = plex_paths.local_media_path(db, it.file_path) - if src is None: - raise HTTPException(status_code=404, detail="Media file not found on disk") - emb_ord = sum(1 for x in subs[:sub_ord] if x.get("index") is not None) - vtt = subtitles.ffmpeg_to_vtt(str(src), emb_ord) - else: - raise HTTPException(status_code=404, detail="Subtitle not fetchable") - except PlexNotConfigured as e: - raise HTTPException(status_code=400, detail=str(e)) - except PlexError as e: - raise HTTPException(status_code=502, detail=f"Subtitle fetch failed: {e}") - subtitles.cache_write(cache_key, vtt) - return Response(subtitles.shift_vtt(vtt, offset), media_type="text/vtt") - - -# --- 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 - - 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 @@ -1452,7 +1122,6 @@ def _rich_meta(meta: dict) -> dict: "cast": plex_people.cast_from_meta(meta), } - @router.get("/people/cards") def people_cards( names: str, @@ -1497,36 +1166,8 @@ def people_cards( out.append({"name": nm, "photo": photos.get(nm), "count": count}) return {"people": out} - -@router.get("/person-image") -def person_image(u: str, request: Request) -> Response: - """Proxy a cast/crew photo from Plex's public metadata CDN (host-whitelisted; no token needed). - Signed-session auth (no DB) so it, like /image, never holds a pool connection during the fetch.""" - _require_session(request) - if not u.startswith(plex_people.PERSON_IMG_HOST): - raise HTTPException(status_code=400, detail="Unsupported image host") - cache_key = "person_" + hashlib.sha1(u.encode()).hexdigest() - hit = imgcache.read(cache_key) - if hit: - return imgcache.response(*hit) - try: - with httpx.Client(timeout=10.0, follow_redirects=True) as c: - r = c.get(u) - r.raise_for_status() - except httpx.HTTPError: - raise HTTPException(status_code=502, detail="Image fetch failed") - ct = r.headers.get("content-type", "image/jpeg") - imgcache.write(cache_key, r.content, ct) - return imgcache.response(r.content, ct) - - -# 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: @@ -1535,7 +1176,6 @@ def _collection_source(col: PlexCollection) -> str: 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).""" @@ -1576,7 +1216,6 @@ def _collection_strips(db: Session, it: PlexItem, user_id: int) -> list[dict]: ) 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 @@ -1593,7 +1232,6 @@ def _episode_nav(db: Session, it: PlexItem) -> tuple[str | None, str | None]: 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, @@ -1696,312 +1334,3 @@ def item_detail( "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), - ) - - -class ItemProgressIn(BaseModel): - # float, truncated in the handler — the player may post a raw currentTime (a float); a plain - # `int` field would 422 it on the periodic progress checkpoint (the old `int(...)` truncated). - position_seconds: float = 0 - duration_seconds: float = 0 - final: bool = False - - -@router.post("/item/{rating_key}/progress") -def item_progress( - rating_key: str, - body: ItemProgressIn, - background: BackgroundTasks, - user: User = Depends(current_user), - db: Session = Depends(get_db), -) -> dict: - """Checkpoint the resume position while playing (near the end → mark watched + clear it). - - `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) - position = max(0, int(body.position_seconds)) - duration = int(body.duration_seconds) - final = body.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} - - -class ItemStateIn(BaseModel): - status: str = "" - - -@router.post("/item/{rating_key}/state") -def item_state( - rating_key: str, - body: ItemStateIn, - background: BackgroundTasks, - user: User = Depends(current_user), - db: Session = Depends(get_db), -) -> dict: - """Set the per-user watch status (new|watched|hidden). watched→scrobble / →new→unscrobble push - to a linked Plex account immediately; `hidden` is a Siftlode-only concept with no Plex analogue, - so it never touches Plex.""" - it = _item_or_404(db, rating_key) - status = body.status - if status not in ("new", "watched", "hidden"): - raise HTTPException(status_code=400, detail="Invalid status") - st = db.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) - - -# Shared by the show- and season-level "mark all episodes watched/unwatched" endpoints. -class WatchedIn(BaseModel): - watched: bool = False - - -@router.post("/show/{rating_key}/state") -def show_state( - rating_key: str, - body: WatchedIn, - background: BackgroundTasks, - user: User = Depends(current_user), - db: Session = Depends(get_db), -) -> dict: - """Mark a whole show watched/unwatched (all its episodes) for the current user + push to Plex.""" - sh = 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 = body.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, - body: WatchedIn, - background: BackgroundTasks, - user: User = Depends(current_user), - db: Session = Depends(get_db), -) -> dict: - """Mark a whole season watched/unwatched (all its episodes) for the current user + push to Plex.""" - se = 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 = body.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"} - ) diff --git a/backend/app/routes/plex/playback.py b/backend/app/routes/plex/playback.py new file mode 100644 index 00000000..deae08dc --- /dev/null +++ b/backend/app/routes/plex/playback.py @@ -0,0 +1,450 @@ +"""Plex playback: image/subtitle proxies, HLS streaming, and watch-state writes.""" + +import hashlib +import os +import tempfile +from datetime import datetime, timezone +import httpx +from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Request, Response +from fastapi.responses import FileResponse +from pydantic import BaseModel +from sqlalchemy.orm import Session +from app.auth import current_user, resolved_user_id +from app.db import SessionLocal, get_db +from app.models import ( + PlexItem, + PlexSeason, + PlexShow, + PlexState, + User, +) +from app.plex import paths as plex_paths +from app.plex import people as plex_people +from app.plex import stream as plex_stream +from app.plex import watch_sync as plex_watch +from app.plex.client import PlexClient, PlexError, PlexNotConfigured +from app.plex import imgcache, subtitles + +router = APIRouter(prefix="/api/plex", tags=["plex"]) + +from app.routes.plex._shared import ( + _item_or_404, +) + +def _require_session(request: Request) -> int: + """Lightweight auth for the high-fan-out image proxies: validate the signed session cookie WITHOUT + a DB user-load, so these endpoints never hold a pool connection for the auth. Poster images are + low-sensitivity; a valid session is enough.""" + uid, _ = resolved_user_id(request) + if not uid: + raise HTTPException(status_code=401, detail="Not authenticated") + return uid + +@router.get("/image/{rating_key}") +def image(rating_key: str, request: Request, variant: str = "thumb") -> Response: + """Proxy a Plex poster/art image (keeps the admin token server-side; no image duplication). A + thin on-disk cache serves repeat views without re-hitting Plex. + + ⚠️ This endpoint is called in BURSTS (a collections page fires 100+ concurrent poster requests), + so it must NOT hold a DB connection during the slow Plex fetch — doing so exhausted the pool + (QueuePool timeout, broken images on prod). Warm path (cache hit) touches NO DB; cold path opens a + brief session only to resolve the image key + read the Plex config, then RELEASES it before the + fetch. Auth is the signed-session check (no DB user-load).""" + _require_session(request) + v = "art" if variant == "art" else "thumb" + w, h = imgcache.SIZES[v] + # Size is part of the cache key: Plex resizes server-side (see PlexClient.image_bytes), so we cache + # the small variant. The width tag also means changing a target size just re-caches, never serves a + # stale differently-sized file (old full-size files under the un-tagged key are simply orphaned). + cache_key = f"item_{rating_key}_{v}_{w}" + hit = imgcache.read(cache_key) + if hit: + return imgcache.response(*hit) + # Cold miss: short-lived session for the key + Plex client config, released before the slow fetch. + try: + with SessionLocal() as db: + key = imgcache.key_for(db, str(rating_key), v) + if not key: + raise HTTPException(status_code=404, detail="No image") + plex = PlexClient(db) # reads base URL + token now; the fetch below needs no DB + except PlexNotConfigured as e: + raise HTTPException(status_code=400, detail=str(e)) + try: + with plex: + data, ct = plex.image_bytes(key, w, h) + except PlexError as e: + raise HTTPException(status_code=502, detail=f"Plex image fetch failed: {e}") + imgcache.write(cache_key, data, ct) + return imgcache.response(data, ct) + +@router.get("/subtitle/{rating_key}/{sub_ord}") +def subtitle(rating_key: str, sub_ord: int, request: Request, offset: float = 0.0) -> Response: + """Serve the ord-th subtitle track as WebVTT for the browser's — so choosing a subtitle + overlays it directly (no HLS re-mux / session restart). Handles EXTERNAL sidecar subs (fetched + from Plex via the stream `key` — the old `-map 0:s` broke on these since they're not in the file) + AND embedded text subs (extracted from the local file). Image subs (PGS/VobSub) → 415. Disk-cached.""" + _require_session(request) + # Cache the ABSOLUTE-time VTT once (per rk+ord); the per-session cue-shift is applied on the way + # out, so a resume/seek to any offset reuses the same cached track (no cache-per-offset explosion). + cache_key = f"sub_{rating_key}_{sub_ord}" + hit = subtitles.cache_read(cache_key) + if hit is not None: + return Response(subtitles.shift_vtt(hit, offset), media_type="text/vtt") + try: + with SessionLocal() as db: + it = _item_or_404(db, rating_key) + with PlexClient(db) as plex: + part = ((plex.metadata(it.rating_key) or {}).get("Media") or [{}])[0].get("Part") or [{}] + subs = [s for s in (part[0].get("Stream") or []) if s.get("streamType") == 3] + if sub_ord < 0 or sub_ord >= len(subs): + raise HTTPException(status_code=404, detail="No such subtitle") + s = subs[sub_ord] + codec = (s.get("codec") or s.get("format") or "").lower() + if codec in subtitles.BITMAP_SUBS: + raise HTTPException(status_code=415, detail="Image-based subtitle can't be shown as text") + if s.get("key"): # external sidecar → fetch from Plex, convert to VTT + raw = plex.raw_get(s["key"]) + if codec in ("srt", "subrip", ""): + vtt = subtitles.srt_to_vtt(raw.decode("utf-8", "replace")) + else: + fd, tmp = tempfile.mkstemp(suffix="." + (codec or "sub")) + try: + os.write(fd, raw) + os.close(fd) + vtt = subtitles.ffmpeg_to_vtt(tmp, None) + finally: + os.unlink(tmp) + elif s.get("index") is not None: # embedded → extract from the local file + src = plex_paths.local_media_path(db, it.file_path) + if src is None: + raise HTTPException(status_code=404, detail="Media file not found on disk") + emb_ord = sum(1 for x in subs[:sub_ord] if x.get("index") is not None) + vtt = subtitles.ffmpeg_to_vtt(str(src), emb_ord) + else: + raise HTTPException(status_code=404, detail="Subtitle not fetchable") + except PlexNotConfigured as e: + raise HTTPException(status_code=400, detail=str(e)) + except PlexError as e: + raise HTTPException(status_code=502, detail=f"Subtitle fetch failed: {e}") + subtitles.cache_write(cache_key, vtt) + return Response(subtitles.shift_vtt(vtt, offset), media_type="text/vtt") + +_PROGRESS_MIN_S = 5 + +_FINISH_MARGIN_S = 30 + +@router.get("/person-image") +def person_image(u: str, request: Request) -> Response: + """Proxy a cast/crew photo from Plex's public metadata CDN (host-whitelisted; no token needed). + Signed-session auth (no DB) so it, like /image, never holds a pool connection during the fetch.""" + _require_session(request) + if not u.startswith(plex_people.PERSON_IMG_HOST): + raise HTTPException(status_code=400, detail="Unsupported image host") + cache_key = "person_" + hashlib.sha1(u.encode()).hexdigest() + hit = imgcache.read(cache_key) + if hit: + return imgcache.response(*hit) + try: + with httpx.Client(timeout=10.0, follow_redirects=True) as c: + r = c.get(u) + r.raise_for_status() + except httpx.HTTPError: + raise HTTPException(status_code=502, detail="Image fetch failed") + ct = r.headers.get("content-type", "image/jpeg") + imgcache.write(cache_key, r.content, ct) + return imgcache.response(r.content, ct) + +def _push_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), + ) + +class ItemProgressIn(BaseModel): + # float, truncated in the handler — the player may post a raw currentTime (a float); a plain + # `int` field would 422 it on the periodic progress checkpoint (the old `int(...)` truncated). + position_seconds: float = 0 + duration_seconds: float = 0 + final: bool = False + +@router.post("/item/{rating_key}/progress") +def item_progress( + rating_key: str, + body: ItemProgressIn, + background: BackgroundTasks, + user: User = Depends(current_user), + db: Session = Depends(get_db), +) -> dict: + """Checkpoint the resume position while playing (near the end → mark watched + clear it). + + `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) + position = max(0, int(body.position_seconds)) + duration = int(body.duration_seconds) + final = body.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} + +class ItemStateIn(BaseModel): + status: str = "" + +@router.post("/item/{rating_key}/state") +def item_state( + rating_key: str, + body: ItemStateIn, + background: BackgroundTasks, + user: User = Depends(current_user), + db: Session = Depends(get_db), +) -> dict: + """Set the per-user watch status (new|watched|hidden). watched→scrobble / →new→unscrobble push + to a linked Plex account immediately; `hidden` is a Siftlode-only concept with no Plex analogue, + so it never touches Plex.""" + it = _item_or_404(db, rating_key) + status = body.status + if status not in ("new", "watched", "hidden"): + raise HTTPException(status_code=400, detail="Invalid status") + st = db.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) + +class WatchedIn(BaseModel): + watched: bool = False + +@router.post("/show/{rating_key}/state") +def show_state( + rating_key: str, + body: WatchedIn, + background: BackgroundTasks, + user: User = Depends(current_user), + db: Session = Depends(get_db), +) -> dict: + """Mark a whole show watched/unwatched (all its episodes) for the current user + push to Plex.""" + sh = 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 = body.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, + body: WatchedIn, + background: BackgroundTasks, + user: User = Depends(current_user), + db: Session = Depends(get_db), +) -> dict: + """Mark a whole season watched/unwatched (all its episodes) for the current user + push to Plex.""" + se = 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 = body.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"} + )