waxwing/app/services/plex/client.py

"""Plex Media Server HTTP client for Waxwing knowledge ingestion.

Responsibilities:
- Parse Plex webhook payloads (multipart/form-data, single 'payload' field)
- Fetch media file bytes from Plex's local HTTP API
- List library sections so callers can identify gardening content
"""
from __future__ import annotations

import json
import logging
import re
from dataclasses import dataclass
from typing import Any
from urllib.parse import urlparse

import httpx

# Plex part keys are always /library/parts/<numeric id>/file — nothing else.
# Rejecting anything outside this pattern prevents SSRF via forged webhook payloads.
_PART_KEY_RE = re.compile(r"^/library/parts/\d+/file/?$")

logger = logging.getLogger(__name__)

# Plex sends webhooks for many events; only these matter for knowledge ingestion.
INGEST_EVENTS: frozenset[str] = frozenset({"library.new", "media.play"})

# Library types that may contain gardening video content.
_VIDEO_TYPES: frozenset[str] = frozenset({"show", "movie"})


@dataclass(frozen=True)
class PlexMediaItem:
    """Minimal metadata extracted from a Plex webhook payload."""
    event: str
    rating_key: str        # Plex item ID
    part_key: str          # path suffix for file download: /library/parts/{id}/file
    title: str
    show_title: str | None  # None for movies
    library_section: str
    duration_ms: int | None
    thumb: str | None      # poster path, unused but useful for future UI


def parse_webhook(raw_payload: str) -> PlexMediaItem | None:
    """Parse the JSON string from Plex's multipart 'payload' field.

    Returns None for events we don't act on (playback progress, rating, etc.).
    """
    try:
        data: dict[str, Any] = json.loads(raw_payload)
    except json.JSONDecodeError:
        logger.warning("Plex webhook: invalid JSON payload")
        return None

    event = data.get("event", "")
    if event not in INGEST_EVENTS:
        return None

    metadata: dict[str, Any] = data.get("Metadata", {})
    media_type = metadata.get("type", "")
    if media_type not in _VIDEO_TYPES:
        logger.debug("Plex webhook: skipping non-video type %r", media_type)
        return None

    # Plex nests file parts under Media[0].Part[0]
    parts: list[dict] = []
    for media in metadata.get("Media", []):
        parts.extend(media.get("Part", []))

    if not parts:
        logger.warning("Plex webhook: no media parts found for %r", metadata.get("title"))
        return None

    part = parts[0]
    part_key = part.get("key", "")
    if not part_key:
        logger.warning("Plex webhook: part has no key for %r", metadata.get("title"))
        return None

    return PlexMediaItem(
        event=event,
        rating_key=str(metadata.get("ratingKey", "")),
        part_key=part_key,
        title=metadata.get("title", "Unknown"),
        show_title=metadata.get("grandparentTitle") or metadata.get("parentTitle"),
        library_section=metadata.get("librarySectionTitle", ""),
        duration_ms=metadata.get("duration"),
        thumb=metadata.get("thumb"),
    )


class PlexClient:
    """Thin async client for Plex Media Server's local HTTP API."""

    def __init__(self, base_url: str, token: str) -> None:
        # base_url: e.g. "http://10.1.10.117:32400"
        self._base = base_url.rstrip("/")
        self._token = token
        self._headers = {
            "X-Plex-Token": token,
            "Accept": "application/json",
        }

    async def list_sections(self) -> list[dict[str, Any]]:
        """Return all library sections (useful for discovering gardening libraries)."""
        async with httpx.AsyncClient(timeout=10.0) as client:
            resp = await client.get(
                f"{self._base}/library/sections",
                headers=self._headers,
            )
            resp.raise_for_status()
            data = resp.json()
            return data.get("MediaContainer", {}).get("Directory", [])

    async def download_file(self, part_key: str, dest_path: str) -> int:
        """Stream a media file from Plex to dest_path on the local filesystem.

        part_key comes from PlexMediaItem.part_key (e.g. /library/parts/12345/file).
        Returns the number of bytes written.

        Raises ValueError if part_key does not match the expected Plex path format —
        this guards against SSRF via a forged webhook payload supplying an arbitrary URL.
        """
        if not _PART_KEY_RE.match(part_key):
            raise ValueError(f"Rejected unsafe part_key: {part_key!r}")

        url = f"{self._base}{part_key}"
        # Sanity-check: constructed URL must stay on the configured Plex host.
        if urlparse(url).hostname != urlparse(self._base).hostname:
            raise ValueError(f"part_key would redirect off Plex host: {url!r}")

        params = {"X-Plex-Token": self._token}
        written = 0

        async with httpx.AsyncClient(timeout=None) as client:
            async with client.stream("GET", url, params=params) as resp:
                resp.raise_for_status()
                with open(dest_path, "wb") as fh:
                    async for chunk in resp.aiter_bytes(chunk_size=1 << 20):  # 1 MB
                        fh.write(chunk)
                        written += len(chunk)

        logger.info("Plex download: %d bytes → %s", written, dest_path)
        return written

    async def get_item_metadata(self, rating_key: str) -> dict[str, Any]:
        """Fetch full metadata for a single library item."""
        async with httpx.AsyncClient(timeout=10.0) as client:
            resp = await client.get(
                f"{self._base}/library/metadata/{rating_key}",
                headers=self._headers,
            )
            resp.raise_for_status()
            data = resp.json()
            items = data.get("MediaContainer", {}).get("Metadata", [])
            return items[0] if items else {}