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
from urllib.parse import quote

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 related(self, rating_key: str) -> list[dict]:
        """Related / similar items for a movie or show, flattened out of Plex's Hub grouping.
        Best-effort (needs the library's related data) — returns [] if Plex has none."""
        try:
            mc = self._get(f"/library/metadata/{rating_key}/related")
        except PlexError:
            return []
        out: list[dict] = []
        for hub in mc.get("Hub", []) or []:
            out.extend(hub.get("Metadata", []) or [])
        return out

    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 []

    # --- Collection editing (P2 — write-back to Plex; all shared/global, admin-gated in the routes) ---
    @property
    def machine_id(self) -> str:
        """Server machineIdentifier — needed in the `uri` of collection add/create calls. Cached."""
        if not getattr(self, "_machine_id", None):
            self._machine_id = self._get("/identity").get("machineIdentifier") or ""
        return self._machine_id

    def _write(self, method: str, path: str) -> dict:
        try:
            r = self._http.request(method, f"{self.base}{path}")
            r.raise_for_status()
        except httpx.HTTPError as e:
            raise PlexError(str(e)) from e
        try:
            return (r.json() or {}).get("MediaContainer", {}) or {} if r.content else {}
        except ValueError:
            return {}

    def _metadata_uri(self, *rating_keys: str) -> str:
        keys = ",".join(str(k) for k in rating_keys)
        return f"server://{self.machine_id}/com.plexapp.plugins.library/library/metadata/{keys}"

    def create_collection(self, section_key: str, title: str, item_rating_key: str) -> dict:
        """Create a (dumb) collection seeded with one item; returns its Metadata (incl. ratingKey)."""
        uri = quote(self._metadata_uri(item_rating_key), safe="")
        mc = self._write(
            "POST",
            f"/library/collections?type=1&smart=0&sectionId={section_key}"
            f"&title={quote(title, safe='')}&uri={uri}",
        )
        items = mc.get("Metadata", []) or []
        if not items:
            raise PlexError("Plex did not return the created collection")
        return items[0]

    def add_collection_item(self, collection_rating_key: str, item_rating_key: str) -> None:
        uri = quote(self._metadata_uri(item_rating_key), safe="")
        self._write("PUT", f"/library/collections/{collection_rating_key}/items?uri={uri}")

    def remove_collection_item(self, collection_rating_key: str, item_rating_key: str) -> None:
        self._write("DELETE", f"/library/collections/{collection_rating_key}/children/{item_rating_key}")

    def rename_collection(self, section_key: str, collection_rating_key: str, title: str) -> None:
        self._write(
            "PUT",
            f"/library/sections/{section_key}/all?type=18&id={collection_rating_key}"
            f"&title.value={quote(title, safe='')}&title.locked=1",
        )

    def delete_collection(self, collection_rating_key: str) -> None:
        self._write("DELETE", f"/library/collections/{collection_rating_key}")

    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

    # --- Watch-state write-back (P5 Phase B — Siftlode→Plex push; acts on the token's account) ---
    _LIBRARY_IDENTIFIER = "com.plexapp.plugins.library"

    def scrobble(self, rating_key: str) -> None:
        """Mark the item *watched* on the token's Plex account (viewCount++, resume cleared)."""
        self._write(
            "GET",
            f"/:/scrobble?identifier={self._LIBRARY_IDENTIFIER}"
            f"&key={quote(str(rating_key), safe='')}",
        )

    def unscrobble(self, rating_key: str) -> None:
        """Mark the item *unwatched* on the token's Plex account (clears viewCount)."""
        self._write(
            "GET",
            f"/:/unscrobble?identifier={self._LIBRARY_IDENTIFIER}"
            f"&key={quote(str(rating_key), safe='')}",
        )

    def set_timeline(
        self, rating_key: str, time_ms: int, duration_ms: int, state: str = "stopped"
    ) -> None:
        """Report a resume position (viewOffset) for the item on the token's account. `state` is a
        Plex timeline state (playing|paused|stopped); we push the settled position as ``stopped``."""
        key = quote(f"/library/metadata/{rating_key}", safe="")
        self._write(
            "GET",
            f"/:/timeline?identifier={self._LIBRARY_IDENTIFIER}"
            f"&ratingKey={quote(str(rating_key), safe='')}&key={key}"
            f"&state={state}&time={int(time_ms)}&duration={int(duration_ms)}",
        )

    # --- Watch-state read for incremental sync (P5 Phase C) ---
    def accounts(self) -> list[dict]:
        """Server accounts (the owner + any managed/home users): id, name. Used to resolve the
        owner's accountID so the shared watch-history feed can be filtered to just their views."""
        return self._get("/accounts").get("Account", []) or []

    def watch_history(
        self, min_viewed_at: int = 0, account_id: int | None = None, size: int = 500
    ) -> list[dict]:
        """The most recent watch-history rows (ratingKey / viewedAt / accountID / type), newest
        first — works without Plex Pass. Filtered client-side to `account_id` and to rows at/after
        `min_viewed_at` (epoch seconds): the efficient "what changed since T" feed. Sorted desc, so
        we stop at the first row older than the cutoff."""
        mc = self._get(
            "/status/sessions/history/all",
            params={
                "sort": "viewedAt:desc",
                "X-Plex-Container-Start": 0,
                "X-Plex-Container-Size": size,
            },
        )
        out = []
        for r in mc.get("Metadata", []) or []:
            if int(r.get("viewedAt") or 0) < min_viewed_at:
                break
            if account_id is not None and int(r.get("accountID") or -1) != account_id:
                continue
            out.append(r)
        return out

    def on_deck(self) -> list[dict]:
        """The token account's Continue-Watching list (ratingKey / viewOffset / lastViewedAt /
        type) — the incremental source for resume-position changes made on the Plex side."""
        return self._get("/library/onDeck").get("Metadata", []) or []

    def raw_get(self, path: str) -> bytes:
        """Raw bytes of an arbitrary Plex resource path (e.g. an external subtitle stream `key` like
        ``/library/streams/553184``). Keeps the admin token server-side."""
        try:
            r = self._http.get(f"{self.base}{path}")
            r.raise_for_status()
        except httpx.HTTPError as e:
            raise PlexError(str(e)) from e
        return r.content

    def image_bytes(
        self, image_path: str, width: int | None = None, height: int | None = None
    ) -> 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``.

        With width+height, Plex's photo transcoder resizes server-side so the browser fetches a small
        image (~tens of KB) instead of the multi-MB full-res original — far faster grid scroll + a
        much smaller disk cache. ``minSize=1`` fits within the box; ``upscale=0`` never enlarges."""
        if width and height:
            path = (
                f"/photo/:/transcode?width={width}&height={height}"
                f"&minSize=1&upscale=0&url={quote(image_path, safe='')}"
            )
        else:
            path = image_path
        try:
            r = self._http.get(f"{self.base}{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")