"""Jellyfin HTTP API client. This module is deliberately independent from Streamlit. It wraps only the API calls the app currently needs and returns plain Python dictionaries/lists so a future FastAPI/React frontend can reuse the same client. """ from __future__ import annotations import logging from typing import Any, cast import requests from media_library_viewer_api.clients.http_timeout import DEFAULT_READ_TIMEOUT, http_timeout logger = logging.getLogger(__name__) # Jellyfin validates Fields against its ItemFields enum. Keep this list to # documented/commonly supported optional fields; invalid names cause 400s. DEFAULT_FIELDS = ",".join( [ "DateCreated", "Genres", "MediaSources", "Overview", "Path", "People", "PremiereDate", "ProviderIds", "Tags", ] ) class JellyfinClient: """Small wrapper around the Jellyfin/Emby-compatible HTTP API.""" def __init__(self, base_url: str, api_key: str, timeout: float = DEFAULT_READ_TIMEOUT): if not base_url: raise ValueError("Jellyfin URL is required") if not api_key: raise ValueError("Jellyfin API key is required") # Use the server root, not the web UI path. Users often paste # https://host/web; API endpoints live at https://host/... self.base_url = base_url.rstrip("/") if self.base_url.endswith("/web"): self.base_url = self.base_url[:-4] self.api_key = api_key # timeout is the per-response READ timeout (seconds); connect timeout is fixed at 5s. self.timeout = http_timeout(timeout) self.session = requests.Session() self.session.headers.update( { "X-Emby-Token": api_key, "Accept": "application/json", "X-Emby-Authorization": ( 'MediaBrowser Client="MediaLibraryViewer", Device="Streamlit", DeviceId="streamlit", Version="0.1"' ), } ) def get(self, path: str, **params: Any) -> Any: """GET a Jellyfin endpoint and include useful response text on errors.""" clean_params = {k: v for k, v in params.items() if v is not None and v != ""} logger.debug("Jellyfin GET %s params=%s", path, sorted(clean_params.keys())) response = self.session.get(f"{self.base_url}{path}", params=clean_params, timeout=self.timeout) try: response.raise_for_status() except requests.HTTPError as exc: detail = response.text[:500] logger.warning("Jellyfin GET %s failed status=%s url=%s", path, response.status_code, response.url) raise requests.HTTPError( f"{response.status_code} for {response.url}: {detail}", response=response, ) from exc logger.debug("Jellyfin GET %s ok status=%s", path, response.status_code) return response.json() def users(self) -> list[dict[str, Any]]: """List users visible to this API key. Jellyfin API keys are server-level tokens, not user session tokens, so /Users/Me often fails with API-key auth. The user id selected here is then used for user-scoped library endpoints. """ users = self.get("/Users") logger.info("Jellyfin returned %s visible users", len(users)) return users def resolve_user_id(self, identifier: str | None) -> str: """Resolve a configured user identifier to Jellyfin's internal Id. The service ``user_id`` config field accepts either the internal Jellyfin Id (a hash) or a username (e.g. ``'admin'``). Jellyfin's ``/Users/{id}/...`` endpoints reject usernames with HTTP 400 (``"The value 'admin' is not valid."``), so any caller must resolve usernames to the real Id before hitting user-scoped endpoints. Resolution order: exact ``Id`` match → ``Name`` match → first visible user. Raises if the API key cannot see any users. """ users = self.users() if not users: raise RuntimeError("No Jellyfin users visible to this API key") if identifier: if any(str(u.get("Id")) == identifier for u in users): return identifier match = next((u for u in users if str(u.get("Name", "")) == identifier), None) if match: resolved = str(match["Id"]) logger.info("Resolved Jellyfin username %r to Id %s", identifier, resolved) return resolved logger.warning("Jellyfin user identifier %r not found; using first user", identifier) return str(users[0]["Id"]) def libraries(self, user_id: str) -> list[dict[str, Any]]: """Return top-level library views visible to the selected Jellyfin user.""" items = self.get(f"/Users/{user_id}/Views").get("Items", []) logger.info("Jellyfin returned %s libraries for user_id=%s", len(items), user_id) return items def items( self, user_id: str, parent_id: str | None = None, start_index: int = 0, limit: int = 50, search: str | None = None, include_item_types: str | None = None, recursive: bool = True, sort_by: str = "SortName", sort_order: str = "Ascending", ) -> dict[str, Any]: """Return a paginated item list for a user/library. This is used by both the visual library browser and the media-index builder. Keep arguments close to Jellyfin's own query parameters so the service layer can request server-side pagination and basic sorting. """ logger.debug( "Jellyfin items user_id=%s parent_id=%s start=%s limit=%s types=%s search=%s sort=%s/%s", user_id, parent_id or "", start_index, limit, include_item_types or "", search or "", sort_by, sort_order, ) return self.get( f"/Users/{user_id}/Items", ParentId=parent_id, StartIndex=start_index, Limit=limit, SearchTerm=search, IncludeItemTypes=include_item_types, Recursive=str(recursive).lower(), Fields=DEFAULT_FIELDS, SortBy=sort_by, SortOrder=sort_order, ) def item_count(self, user_id: str, include_item_types: str, parent_id: str | None = None) -> int: """Return a count using Jellyfin's TotalRecordCount without fetching rows.""" response = self.get( f"/Users/{user_id}/Items", ParentId=parent_id, Recursive="true", IncludeItemTypes=include_item_types, Limit=0, ) count = int(response.get("TotalRecordCount", 0)) logger.debug( "Jellyfin item count user_id=%s parent_id=%s types=%s count=%s", user_id, parent_id or "", include_item_types, count, ) return count def media_counts(self, user_id: str) -> dict[str, int]: """Return dashboard-level counts for the main media types.""" return { "movies": self.item_count(user_id, "Movie"), "series": self.item_count(user_id, "Series"), "episodes": self.item_count(user_id, "Episode"), } def library_item_counts(self, user_id: str, libraries: list[dict[str, Any]]) -> list[dict[str, Any]]: """Return per-library item counts broken down by type for the dashboard.""" results = [] for lib in libraries: lib_id = lib.get("Id") lib_name = lib.get("Name", "Unknown") lib_type = lib.get("CollectionType", "") if not lib_id: continue movies = self.item_count(user_id, "Movie", parent_id=lib_id) series = self.item_count(user_id, "Series", parent_id=lib_id) episodes = self.item_count(user_id, "Episode", parent_id=lib_id) total = self.item_count(user_id, "Movie,Episode,Video,Audio,Series", parent_id=lib_id) results.append( { "library": lib_name, "type": lib_type, "movies": movies, "series": series, "episodes": episodes, "total": total, } ) return results def sessions(self, active_within_seconds: int | None = None) -> list[dict[str, Any]]: """Return Jellyfin sessions (playing and idle/logged-in). When ``active_within_seconds`` is None, no recency filter is sent and Jellyfin decides which sessions to include. """ payload: Any = self.get("/Sessions", ActiveWithinSeconds=active_within_seconds) return cast(list[dict[str, Any]], payload) if isinstance(payload, list) else [] def active_sessions(self, active_within_seconds: int = 300) -> list[dict[str, Any]]: """Return sessions that currently have a now-playing item.""" sessions = [session for session in self.sessions(active_within_seconds) if session.get("NowPlayingItem")] logger.info("Jellyfin active sessions within %ss: %s", active_within_seconds, len(sessions)) return sessions def image_url(self, item_id: str, image_type: str = "Primary") -> str: """Build an authenticated image URL suitable for st.image/browser use.""" return f"{self.base_url}/Items/{item_id}/Images/{image_type}?api_key={self.api_key}"