feat(plex): P0 backend foundations — catalog model, config, client

Optional Plex module groundwork (design locked 2026-07-05): plays the LOCAL
physical file; Plex is metadata + optional watch-sync only.

- models + migration 0044: plex_libraries/plex_shows/plex_seasons/plex_items
  (playable leaf, ffprobe playability class, markers, weighted FTS search_vector)
  + plex_states (per-user watch state, mirrors video_states)
- sysconfig 'plex' group + config.py env defaults (server url/token[secret]/
  path-map/libraries/sync-interval/max-transcodes)
- app/plex/client.py (PlexClient httpx: server info, sections, items, metadata
  +markers, image) + app/plex/paths.py (Plex→local path map + file resolve)
- routes/plex.py admin POST /api/plex/test (verify connection + list sections)

backend/app/plex/client.py

@@ -0,0 +1,113 @@
+"""Thin synchronous Plex Media Server client (metadata + images + watch-state).
+
+Server-side ONLY: uses the admin ``X-Plex-Token`` from sysconfig. Playback never goes through
+this client — Siftlode plays the LOCAL physical file (see app.plex.paths). Plex is used purely for
+metadata, image proxying, and (P5) optional watch-state write-back. Requests JSON via the
+``Accept: application/json`` header (Plex serves XML by default).
+"""
+import logging
+
+import httpx
+from sqlalchemy.orm import Session
+
+from app import sysconfig
+from app.config import settings
+
+log = logging.getLogger("siftlode.plex")
+
+
+class PlexError(Exception):
+    """Any failure talking to the Plex server (network, HTTP status, bad payload)."""
+
+
+class PlexNotConfigured(PlexError):
+    """The Plex server URL or token isn't set — the module can't do anything."""
+
+
+class PlexClient:
+    """Bound to the configured server + admin token. Use as a context manager so the underlying
+    httpx client is closed."""
+
+    def __init__(self, db: Session):
+        self.db = db
+        self.base = sysconfig.get_str(db, "plex_server_url").rstrip("/")
+        self.token = sysconfig.get_str(db, "plex_server_token")
+        if not self.base or not self.token:
+            raise PlexNotConfigured("Plex server URL or token is not configured")
+        self._http = httpx.Client(
+            timeout=30.0,
+            headers={
+                "Accept": "application/json",
+                "X-Plex-Token": self.token,
+                "X-Plex-Client-Identifier": settings.plex_client_id,
+                "X-Plex-Product": "Siftlode",
+            },
+        )
+
+    def __enter__(self) -> "PlexClient":
+        return self
+
+    def __exit__(self, *exc) -> None:
+        self._http.close()
+
+    def close(self) -> None:
+        self._http.close()
+
+    def _get(self, path: str, params: dict | None = None) -> dict:
+        try:
+            r = self._http.get(f"{self.base}{path}", params=params)
+            r.raise_for_status()
+        except httpx.HTTPError as e:
+            raise PlexError(str(e)) from e
+        try:
+            return (r.json() or {}).get("MediaContainer", {}) or {}
+        except ValueError as e:
+            raise PlexError("Plex returned a non-JSON response") from e
+
+    # --- Connectivity + sections (P0) ---
+    def server_info(self) -> dict:
+        """Root identity (friendlyName / version) — used by the admin 'Test connection'."""
+        return self._get("/")
+
+    def sections(self) -> list[dict]:
+        """Library sections (Directory entries): key, title, type (movie|show|...)."""
+        mc = self._get("/library/sections")
+        return mc.get("Directory", []) or []
+
+    # --- Catalog mirror (P1) ---
+    def section_items(
+        self, key: str, item_type: int, start: int = 0, size: int = 200
+    ) -> tuple[list[dict], int]:
+        """A page of a section's top-level items. type 1 = movie, 2 = show. Returns
+        (items, total_size)."""
+        mc = self._get(
+            f"/library/sections/{key}/all",
+            params={
+                "type": item_type,
+                "X-Plex-Container-Start": start,
+                "X-Plex-Container-Size": size,
+            },
+        )
+        return mc.get("Metadata", []) or [], int(mc.get("totalSize", mc.get("size", 0)) or 0)
+
+    def children(self, rating_key: str) -> list[dict]:
+        """Direct children (show → seasons, season → episodes)."""
+        mc = self._get(f"/library/metadata/{rating_key}/children")
+        return mc.get("Metadata", []) or []
+
+    def metadata(self, rating_key: str, markers: bool = False) -> dict | None:
+        """Full metadata for one item; include intro/credit markers when requested."""
+        params = {"includeMarkers": 1} if markers else None
+        mc = self._get(f"/library/metadata/{rating_key}", params=params)
+        items = mc.get("Metadata", []) or []
+        return items[0] if items else None
+
+    def image_bytes(self, image_path: str) -> tuple[bytes, str]:
+        """Raw bytes + content-type of a Plex thumb/art image (proxied to the browser). The
+        image_path is a Plex path like ``/library/metadata/123/thumb/456``."""
+        try:
+            r = self._http.get(f"{self.base}{image_path}")
+            r.raise_for_status()
+        except httpx.HTTPError as e:
+            raise PlexError(str(e)) from e
+        return r.content, r.headers.get("Content-Type", "image/jpeg")

backend/app/plex/paths.py

@@ -0,0 +1,51 @@
+"""Map a Plex-side media path to Siftlode's local mount, so the player can open the physical file.
+
+The Plex metadata gives each media Part an absolute path as the PLEX server sees it (e.g.
+``/data/movies/Foo.mkv``). Siftlode plays the file directly, so it must translate that to the path
+as SIFTLODE sees the same storage on a local mount (e.g. ``/plex-media/movies/Foo.mkv``). The map
+is admin-configured (sysconfig ``plex_path_map``) as newline/comma-separated ``plexPrefix=localPrefix``
+pairs; the longest matching Plex prefix wins. An empty map means the paths are identical on both
+sides (Plex and Siftlode see the same filesystem).
+"""
+from pathlib import Path
+
+from sqlalchemy.orm import Session
+
+from app import sysconfig
+
+
+def parse_map(raw: str) -> list[tuple[str, str]]:
+    """Parse the ``plexPrefix=localPrefix`` pairs, longest Plex prefix first."""
+    pairs: list[tuple[str, str]] = []
+    for chunk in (raw or "").replace("\n", ",").split(","):
+        chunk = chunk.strip()
+        if not chunk or "=" not in chunk:
+            continue
+        plex, local = chunk.split("=", 1)
+        plex, local = plex.strip(), local.strip()
+        if plex and local:
+            pairs.append((plex, local))
+    pairs.sort(key=lambda p: len(p[0]), reverse=True)
+    return pairs
+
+
+def map_path(db: Session, plex_path: str) -> str:
+    """Translate a Plex-side absolute path to the local mount path (no mapping = unchanged)."""
+    for plex_prefix, local_prefix in parse_map(sysconfig.get_str(db, "plex_path_map")):
+        if plex_path == plex_prefix or plex_path.startswith(plex_prefix.rstrip("/") + "/"):
+            rest = plex_path[len(plex_prefix):].lstrip("/")
+            return str(Path(local_prefix) / rest) if rest else local_prefix
+    return plex_path
+
+
+def local_media_path(db: Session, plex_path: str | None) -> Path | None:
+    """The resolved, existing local file for a Plex media path, or None if it can't be read.
+    The path originates from the trusted Plex server + an admin-set map (not user input), so no
+    traversal check is needed beyond requiring a real regular file to exist."""
+    if not plex_path:
+        return None
+    try:
+        p = Path(map_path(db, plex_path)).resolve()
+    except OSError:
+        return None
+    return p if p.is_file() else None