siftlode/backend/app/plex/client.py

"""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 collections(self, section_key: str) -> list[dict]:
        """All collections in a library section (title/summary/thumb/childCount/smart)."""
        mc = self._get(f"/library/sections/{section_key}/collections")
        return mc.get("Metadata", []) or []

    def collection_children(self, rating_key: str) -> list[dict]:
        """The member items (movies / shows) of a collection."""
        mc = self._get(f"/library/collections/{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")