"""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//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 {}