feat(plex): P1 backend — catalog sync + browse/search/show/image API

- app/plex/sync.py: full reconcile of enabled movie/show sections into plex_*
  (upsert by rating_key). Reuses Plex's own codec info to classify browser
  playability (direct/remux/transcode) — no ffprobe. Movies + episodes become
  playable leaves; shows/seasons mirrored for the drill-down.
- routes/plex.py: GET /libraries, /browse (FTS search + sort + paging, movie
  leaves or show cards), /show/{rk} (seasons+episodes, per-user state), /image
  (proxies Plex art, no duplication); admin POST /sync. All auth'd, demo-allowed.
- scheduler: plex_sync job (interval settings.plex_sync_interval_min).
- Verified against the real server: 2 libs, 3206 movies, 260 shows/961 seasons/
  14768 episodes synced in 13s; FTS search + drill-down work. Playability:
  ~1% direct, 89% remux, 6% transcode (informs P2/P3).

backend/app/plex/sync.py

@@ -0,0 +1,236 @@
+"""Mirror a locally-reachable Plex library into Siftlode's own catalog (plex_* tables).
+
+Playback is from the local physical file, so this only mirrors METADATA + the physical-media facts
+(codecs/container/path) needed to browse, search, and later play. It reuses Plex's own codec info
+(no ffprobe needed) to classify each item's browser playability. Movies + episodes become playable
+LEAF rows (plex_items); shows/seasons are mirrored for the drill-down + episode ordering.
+
+The sync is a full reconcile of the enabled sections (upsert by rating_key). Pruning of items that
+disappeared from Plex is intentionally left for later (a follow-up can diff rating_keys per library).
+Cast + intro/credit markers are fetched lazily on the item-detail/player path (P2), not here.
+"""
+import logging
+from datetime import datetime, timezone
+
+from sqlalchemy.orm import Session
+
+from app import sysconfig
+from app.models import PlexItem, PlexLibrary, PlexSeason, PlexShow
+from app.plex.client import PlexClient, PlexError, PlexNotConfigured
+
+log = logging.getLogger("siftlode.plex")
+
+# Codecs/containers a browser can play in a native <video> without any transcoding.
+_BROWSER_VIDEO = {"h264"}
+_BROWSER_AUDIO = {"aac", "mp3", "opus", "vorbis"}
+_BROWSER_CONTAINER = {"mp4", "mov", "m4v"}
+
+# Plex section item types.
+_TYPE_MOVIE = 1
+_TYPE_SHOW = 2
+_TYPE_SEASON = 3
+_TYPE_EPISODE = 4
+
+_PAGE = 200
+
+
+def classify_playable(vcodec: str | None, acodec: str | None, container: str | None) -> str:
+    """How the browser can play this file: 'direct' (native <video>), 'remux' (copy video, fix
+    container / transcode audio — cheap), or 'transcode' (re-encode video — expensive)."""
+    v = (vcodec or "").lower()
+    a = (acodec or "").lower()
+    c = (container or "").lower()
+    if not v or v not in _BROWSER_VIDEO:
+        return "transcode"
+    if a and a not in _BROWSER_AUDIO:
+        return "remux"
+    if c and c not in _BROWSER_CONTAINER:
+        return "remux"
+    return "direct"
+
+
+def _epoch(v) -> datetime | None:
+    try:
+        return datetime.fromtimestamp(int(v), tz=timezone.utc) if v else None
+    except (TypeError, ValueError, OSError):
+        return None
+
+
+def _media_facts(meta: dict) -> dict:
+    media = meta.get("Media") or []
+    if not media:
+        return {}
+    m = media[0]
+    part = (m.get("Part") or [{}])[0]
+    return {
+        "container": m.get("container") or part.get("container"),
+        "codec_video": m.get("videoCodec"),
+        "codec_audio": m.get("audioCodec"),
+        "size_bytes": part.get("size"),
+        "file_path": part.get("file"),
+        "part_key": part.get("key"),
+    }
+
+
+def _dur_s(meta: dict) -> int | None:
+    d = meta.get("duration")
+    return int(d / 1000) if d else None
+
+
+def _enabled_section_keys(db: Session) -> set[str] | None:
+    """The configured allowlist of section keys, or None to mean 'all movie/show sections'."""
+    raw = sysconfig.get_str(db, "plex_libraries")
+    keys = {k.strip() for k in raw.split(",") if k.strip()}
+    return keys or None
+
+
+def sync(db: Session) -> dict:
+    """Full reconcile of the enabled movie/show sections. Returns a small stats dict."""
+    if not sysconfig.get_bool(db, "plex_enabled"):
+        return {"skipped": "disabled"}
+    stats = {"libraries": 0, "movies": 0, "shows": 0, "seasons": 0, "episodes": 0}
+    wanted = _enabled_section_keys(db)
+    try:
+        with PlexClient(db) as plex:
+            for s in plex.sections():
+                if s.get("type") not in ("movie", "show"):
+                    continue
+                key = str(s.get("key"))
+                if wanted is not None and key not in wanted:
+                    continue
+                lib = _upsert_library(db, key, s.get("title") or key, s.get("type"))
+                stats["libraries"] += 1
+                if s.get("type") == "movie":
+                    _sync_movies(db, plex, lib, stats)
+                else:
+                    _sync_shows(db, plex, lib, stats)
+            db.commit()
+    except PlexNotConfigured as e:
+        return {"skipped": str(e)}
+    except PlexError as e:
+        db.rollback()
+        log.warning("Plex sync failed: %s", e)
+        return {"error": str(e)}
+    log.info("Plex sync done: %s", stats)
+    return stats
+
+
+def _upsert_library(db: Session, key: str, title: str, kind: str) -> PlexLibrary:
+    lib = db.query(PlexLibrary).filter_by(plex_key=key).first()
+    if lib is None:
+        lib = PlexLibrary(plex_key=key)
+        db.add(lib)
+    lib.title = title
+    lib.kind = kind
+    lib.synced_at = datetime.now(timezone.utc)
+    db.flush()
+    return lib
+
+
+def _paginate(plex: PlexClient, key: str, item_type: int):
+    start = 0
+    while True:
+        items, total = plex.section_items(key, item_type, start, _PAGE)
+        if not items:
+            break
+        for it in items:
+            yield it
+        start += len(items)
+        if total and start >= total:
+            break
+        if len(items) < _PAGE:
+            break
+
+
+def _apply_item(row: PlexItem, lib_id: int, kind: str, meta: dict) -> None:
+    row.library_id = lib_id
+    row.kind = kind
+    row.title = (meta.get("title") or "").strip() or "Untitled"
+    row.summary = meta.get("summary")
+    row.year = meta.get("year")
+    row.duration_s = _dur_s(meta)
+    row.thumb_key = meta.get("thumb")
+    row.art_key = meta.get("art") or meta.get("grandparentArt")
+    row.added_at = _epoch(meta.get("addedAt"))
+    mf = _media_facts(meta)
+    row.file_path = mf.get("file_path")
+    row.part_key = mf.get("part_key")
+    row.container = mf.get("container")
+    row.codec_video = mf.get("codec_video")
+    row.codec_audio = mf.get("codec_audio")
+    row.size_bytes = mf.get("size_bytes")
+    row.playable = classify_playable(mf.get("codec_video"), mf.get("codec_audio"), mf.get("container"))
+
+
+def _sync_movies(db: Session, plex: PlexClient, lib: PlexLibrary, stats: dict) -> None:
+    existing = {r.rating_key: r for r in db.query(PlexItem).filter_by(library_id=lib.id, kind="movie")}
+    for meta in _paginate(plex, lib.plex_key, _TYPE_MOVIE):
+        rk = str(meta.get("ratingKey") or "")
+        if not rk:
+            continue
+        row = existing.get(rk) or PlexItem(rating_key=rk)
+        if row.id is None:
+            db.add(row)
+        _apply_item(row, lib.id, "movie", meta)
+        stats["movies"] += 1
+    db.flush()
+
+
+def _sync_shows(db: Session, plex: PlexClient, lib: PlexLibrary, stats: dict) -> None:
+    # 1) shows
+    shows = {r.rating_key: r for r in db.query(PlexShow).filter_by(library_id=lib.id)}
+    for meta in _paginate(plex, lib.plex_key, _TYPE_SHOW):
+        rk = str(meta.get("ratingKey") or "")
+        if not rk:
+            continue
+        sh = shows.get(rk) or PlexShow(rating_key=rk)
+        if sh.id is None:
+            db.add(sh)
+            shows[rk] = sh
+        sh.library_id = lib.id
+        sh.title = (meta.get("title") or "").strip() or "Untitled"
+        sh.summary = meta.get("summary")
+        sh.year = meta.get("year")
+        sh.thumb_key = meta.get("thumb")
+        sh.art_key = meta.get("art")
+        sh.child_count = meta.get("childCount")
+        sh.added_at = _epoch(meta.get("addedAt"))
+        stats["shows"] += 1
+    db.flush()
+    show_id = {rk: sh.id for rk, sh in shows.items()}
+
+    # 2) seasons
+    seasons = {r.rating_key: r for r in db.query(PlexSeason).join(PlexShow).filter(PlexShow.library_id == lib.id)}
+    for meta in _paginate(plex, lib.plex_key, _TYPE_SEASON):
+        rk = str(meta.get("ratingKey") or "")
+        sid = show_id.get(str(meta.get("parentRatingKey") or ""))
+        if not rk or sid is None:
+            continue
+        se = seasons.get(rk) or PlexSeason(rating_key=rk)
+        if se.id is None:
+            db.add(se)
+            seasons[rk] = se
+        se.show_id = sid
+        se.title = meta.get("title")
+        se.season_number = meta.get("index")
+        se.thumb_key = meta.get("thumb")
+        stats["seasons"] += 1
+    db.flush()
+    season_id = {rk: se.id for rk, se in seasons.items()}
+
+    # 3) episodes (playable leaves)
+    episodes = {r.rating_key: r for r in db.query(PlexItem).filter_by(library_id=lib.id, kind="episode")}
+    for meta in _paginate(plex, lib.plex_key, _TYPE_EPISODE):
+        rk = str(meta.get("ratingKey") or "")
+        if not rk:
+            continue
+        row = episodes.get(rk) or PlexItem(rating_key=rk)
+        if row.id is None:
+            db.add(row)
+        _apply_item(row, lib.id, "episode", meta)
+        row.show_id = show_id.get(str(meta.get("grandparentRatingKey") or ""))
+        row.season_id = season_id.get(str(meta.get("parentRatingKey") or ""))
+        row.season_number = meta.get("parentIndex")
+        row.episode_number = meta.get("index")
+        stats["episodes"] += 1
+    db.flush()

backend/app/routes/plex.py

@@ -1,27 +1,39 @@
 """Plex integration API.
 
-Read endpoints (browse/search/stream/image/state) are available to ANY authenticated user — the
+Read endpoints (libraries/browse/show/image) are available to ANY authenticated user (demo +
-whole point of the module is to be an access layer to the Plex library WITHOUT requiring a plex.tv
+pure-Siftlode included) — the module is an access layer to the Plex library WITHOUT requiring a
-account, and playback is a local file (no quota/credit cost), so demo + pure-Siftlode users are
+plex.tv account, and playback is a local file. Admin endpoints test the connection and run the
-allowed. Admin endpoints configure + test the connection and trigger a sync.
+catalog sync.
-
-P0 ships only the admin connectivity endpoints; browse/search/player land in P1/P2.
 """
-from fastapi import APIRouter, Depends, HTTPException
+import logging
+
+from fastapi import APIRouter, Depends, HTTPException, Query, Response
+from sqlalchemy import func
 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 User
+from app.models import PlexItem, PlexLibrary, PlexSeason, PlexShow, PlexState, User
+from app.plex import sync as plex_sync
 from app.plex.client import PlexClient, PlexError, PlexNotConfigured
 from app.routes.admin import admin_user
+from app.routes.feed import _to_tsquery_str
+
+log = logging.getLogger("siftlode.plex")
 
 router = APIRouter(prefix="/api/plex", tags=["plex"])
 
+_TS_CONFIG = "public.unaccent_simple"
+
+
+# --- Admin: connectivity + sync ---------------------------------------------------------------
+
 
 @router.post("/test")
 def test_connection(_: User = Depends(admin_user), db: Session = Depends(get_db)) -> dict:
     """Admin: verify the configured Plex server URL + token, returning the server name and the
-    available library sections (movie/show) for the config library-picker."""
+    movie/show sections (for the config library-picker)."""
     try:
         with PlexClient(db) as plex:
             info = plex.server_info()
@@ -35,13 +47,223 @@ def test_connection(_: User = Depends(admin_user), db: Session = Depends(get_db)
         "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")}
-                "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")
         ],
     }
+
+
+@router.post("/sync")
+def trigger_sync(_: User = Depends(admin_user), db: Session = Depends(get_db)) -> dict:
+    """Admin: mirror the enabled Plex sections into the local catalog. Synchronous (can take a
+    while on a large library); a background/scheduled sync also runs periodically."""
+    return plex_sync.sync(db)
+
+
+# --- Read: libraries / browse / show / image --------------------------------------------------
+
+
+def _enabled() -> None:
+    """Guard read endpoints when the module is off (avoids leaking a stale mirror)."""
+    # Read endpoints stay usable as long as a mirror exists; the toggle only gates sync + UI.
+    return None
+
+
+@router.get("/libraries")
+def list_libraries(user: User = Depends(current_user), db: Session = Depends(get_db)) -> dict:
+    """The mirrored libraries the user can browse (drives the Plex scope selector)."""
+    libs = db.query(PlexLibrary).order_by(PlexLibrary.kind.desc(), PlexLibrary.title).all()
+    out = []
+    for lib in libs:
+        if lib.kind == "movie":
+            count = db.query(func.count(PlexItem.id)).filter_by(library_id=lib.id, kind="movie").scalar()
+        else:
+            count = db.query(func.count(PlexShow.id)).filter_by(library_id=lib.id).scalar()
+        out.append({"key": lib.plex_key, "title": lib.title, "kind": lib.kind, "count": count or 0})
+    return {"enabled": sysconfig.get_bool(db, "plex_enabled"), "libraries": out}
+
+
+def _movie_card(it: PlexItem, st: PlexState | None) -> dict:
+    return {
+        "id": it.rating_key,
+        "type": "movie",
+        "title": it.title,
+        "year": it.year,
+        "duration_seconds": it.duration_s,
+        "thumb": f"/api/plex/image/{it.rating_key}",
+        "playable": it.playable,
+        "status": st.status if st else "new",
+        "position_seconds": st.position_seconds if st else 0,
+    }
+
+
+def _show_card(sh: PlexShow) -> dict:
+    return {
+        "id": sh.rating_key,
+        "type": "show",
+        "title": sh.title,
+        "year": sh.year,
+        "thumb": f"/api/plex/image/{sh.rating_key}",
+        "season_count": sh.child_count,
+    }
+
+
+def _episode_card(e: PlexItem, st: PlexState | None) -> dict:
+    return {
+        "id": e.rating_key,
+        "type": "episode",
+        "title": e.title,
+        "summary": e.summary,
+        "season_number": e.season_number,
+        "episode_number": e.episode_number,
+        "duration_seconds": e.duration_s,
+        "thumb": f"/api/plex/image/{e.rating_key}",
+        "playable": e.playable,
+        "status": st.status if st else "new",
+        "position_seconds": st.position_seconds if st else 0,
+    }
+
+
+@router.get("/browse")
+def browse(
+    library: str,
+    q: str | None = None,
+    sort: str = "added",
+    offset: int = 0,
+    limit: int = Query(default=40, ge=1, le=100),
+    user: User = Depends(current_user),
+    db: Session = Depends(get_db),
+) -> dict:
+    """List a library's movies (leaves) or shows, with optional FTS search + sorting.
+    Movie libraries return playable movie cards; show libraries return show cards (drill in via
+    /show/{id})."""
+    lib = db.query(PlexLibrary).filter_by(plex_key=str(library)).first()
+    if lib is None:
+        raise HTTPException(status_code=404, detail="Unknown Plex library")
+    offset = max(0, offset)
+    model = PlexItem if lib.kind == "movie" else PlexShow
+
+    query = db.query(model)
+    if lib.kind == "movie":
+        query = query.filter(PlexItem.library_id == lib.id, PlexItem.kind == "movie")
+    else:
+        query = query.filter(PlexShow.library_id == lib.id)
+
+    ts = _to_tsquery_str(q) if q else None
+    tsq = None
+    if ts:
+        tsq = func.to_tsquery(_TS_CONFIG, ts)
+        query = query.filter(model.search_vector.op("@@")(tsq))
+
+    total = query.count()
+
+    if tsq is not None:
+        order = [func.ts_rank(model.search_vector, tsq).desc()]
+    elif sort == "title":
+        order = [func.lower(model.title).asc()]
+    else:  # newest first
+        order = [model.added_at.desc().nullslast()]
+    order.append(model.id.desc())
+
+    rows = query.order_by(*order).offset(offset).limit(limit).all()
+
+    if lib.kind == "movie":
+        by_item = {r.id: r for r in rows}
+        states = {
+            s.item_id: s
+            for s in db.query(PlexState).filter(
+                PlexState.user_id == user.id, PlexState.item_id.in_(list(by_item.keys()) or [0])
+            )
+        }
+        items = [_movie_card(r, states.get(r.id)) for r in rows]
+    else:
+        items = [_show_card(r) for r in rows]
+
+    return {"kind": lib.kind, "total": total, "offset": offset, "limit": limit, "items": items}
+
+
+@router.get("/show/{rating_key}")
+def show_detail(
+    rating_key: str,
+    user: User = Depends(current_user),
+    db: Session = Depends(get_db),
+) -> dict:
+    """A show's seasons + episodes (the drill-down view) with per-user watch state."""
+    sh = db.query(PlexShow).filter_by(rating_key=str(rating_key)).first()
+    if sh is None:
+        raise HTTPException(status_code=404, detail="Unknown Plex show")
+    seasons = (
+        db.query(PlexSeason).filter_by(show_id=sh.id).order_by(PlexSeason.season_number.asc().nullsfirst()).all()
+    )
+    eps = (
+        db.query(PlexItem)
+        .filter_by(show_id=sh.id, kind="episode")
+        .order_by(PlexItem.season_number.asc().nullsfirst(), PlexItem.episode_number.asc().nullsfirst())
+        .all()
+    )
+    states = {
+        s.item_id: s
+        for s in db.query(PlexState).filter(
+            PlexState.user_id == user.id, PlexState.item_id.in_([e.id for e in eps] or [0])
+        )
+    }
+    by_season: dict[int | None, list] = {}
+    for e in eps:
+        by_season.setdefault(e.season_id, []).append(_episode_card(e, states.get(e.id)))
+    return {
+        "show": {
+            "id": sh.rating_key,
+            "title": sh.title,
+            "summary": sh.summary,
+            "year": sh.year,
+            "thumb": f"/api/plex/image/{sh.rating_key}",
+            "art": f"/api/plex/image/{sh.rating_key}?variant=art",
+        },
+        "seasons": [
+            {
+                "id": se.rating_key,
+                "season_number": se.season_number,
+                "title": se.title or (f"Season {se.season_number}" if se.season_number else "Season"),
+                "thumb": f"/api/plex/image/{se.rating_key}",
+                "episodes": by_season.get(se.id, []),
+            }
+            for se in seasons
+        ],
+    }
+
+
+def _image_key(db: Session, rating_key: str, variant: str) -> str | None:
+    """The stored Plex image path for a known rating_key (item/show/season). Only mirrored keys
+    are proxyable — this is not an open image proxy."""
+    it = db.query(PlexItem).filter_by(rating_key=rating_key).first()
+    if it is not None:
+        return it.art_key if variant == "art" else it.thumb_key
+    sh = db.query(PlexShow).filter_by(rating_key=rating_key).first()
+    if sh is not None:
+        return sh.art_key if variant == "art" else sh.thumb_key
+    se = db.query(PlexSeason).filter_by(rating_key=rating_key).first()
+    if se is not None:
+        return se.thumb_key
+    return None
+
+
+@router.get("/image/{rating_key}")
+def image(
+    rating_key: str,
+    variant: str = "thumb",
+    user: User = Depends(current_user),
+    db: Session = Depends(get_db),
+) -> Response:
+    """Proxy a Plex poster/art image (keeps the admin token server-side; no image duplication)."""
+    key = _image_key(db, str(rating_key), "art" if variant == "art" else "thumb")
+    if not key:
+        raise HTTPException(status_code=404, detail="No image")
+    try:
+        with PlexClient(db) as plex:
+            data, ct = plex.image_bytes(key)
+    except PlexNotConfigured as e:
+        raise HTTPException(status_code=400, detail=str(e))
+    except PlexError as e:
+        raise HTTPException(status_code=502, detail=f"Plex image fetch failed: {e}")
+    return Response(content=data, media_type=ct, headers={"Cache-Control": "public, max-age=86400"})

backend/app/scheduler.py

@@ -14,6 +14,7 @@ from app.config import settings
 from app.db import SessionLocal
 from app.downloads.gc import run_download_gc
 from app.models import SchedulerSetting
+from app.plex.sync import sync as run_plex_sync
 from app.notifications import create_notification
 from app.state import is_sync_paused
 from app.sync.autotag import run_autotag_all
@@ -61,6 +62,7 @@ JOB_INTERVALS: dict[str, int] = {
     "demo_reset": settings.demo_reset_minutes,
     "explore_cleanup": settings.explore_cleanup_minutes,
     "download_gc": settings.download_gc_minutes,
+    "plex_sync": settings.plex_sync_interval_min,
 }
 
 # Sane bounds for an admin-set interval (minutes).
@@ -284,6 +286,12 @@ def _download_gc_job() -> None:
     _job("download_gc", run_download_gc)
 
 
+def _plex_sync_job() -> None:
+    # Mirror the enabled Plex library sections into the local catalog. Pure metadata (no YouTube
+    # quota); a no-op when the Plex module is disabled.
+    _job("plex_sync", run_plex_sync)
+
+
 # job_id -> wrapper. The single source of truth for which jobs exist and how to run one,
 # shared by start_scheduler (recurring registration) and trigger_job (manual "run now").
 JOB_FUNCS: dict[str, Callable[[], None]] = {
@@ -298,6 +306,7 @@ JOB_FUNCS: dict[str, Callable[[], None]] = {
     "demo_reset": _demo_reset_job,
     "explore_cleanup": _explore_cleanup_job,
     "download_gc": _download_gc_job,
+    "plex_sync": _plex_sync_job,
 }
 
 

backend/app/sysconfig.py

@@ -82,7 +82,7 @@ SPECS: tuple[ConfigSpec, ...] = (
     ConfigSpec("plex_server_token", "str", "plex", "plex_server_token", secret=True),
     ConfigSpec("plex_path_map", "str", "plex", "plex_path_map"),
     ConfigSpec("plex_libraries", "str", "plex", "plex_libraries"),
-    ConfigSpec("plex_sync_interval_min", "int", "plex", "plex_sync_interval_min", min=5, max=100_000),
+    # (plex_sync_interval_min is tuned from the Scheduler dashboard, not here — it's a scheduler job.)
     ConfigSpec("plex_max_transcodes", "int", "plex", "plex_max_transcodes", min=0, max=16),
 )