manage/archive/src/media_library_viewer/clients/jellyfin.py

"""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

from typing import Any

import requests


# 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: int = 30):
        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
        self.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) -> dict[str, 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 != ""}
        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]
            raise requests.HTTPError(
                f"{response.status_code} for {response.url}: {detail}",
                response=response,
            ) from exc
        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.
        """
        return self.get("/Users")

    def libraries(self, user_id: str) -> list[dict[str, Any]]:
        """Return top-level library views visible to the selected Jellyfin user."""
        return self.get(f"/Users/{user_id}/Views").get("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.
        """
        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,
        )
        return int(response.get("TotalRecordCount", 0))

    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 active_sessions(self, active_within_seconds: int = 300) -> list[dict[str, Any]]:
        """Return currently active sessions that have a now-playing item."""
        payload = self.get("/Sessions", ActiveWithinSeconds=active_within_seconds)
        sessions = payload if isinstance(payload, list) else []
        return [session for session in sessions if session.get("NowPlayingItem")]

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