Developer
208 lines
8.0 KiB
Python
208 lines
8.0 KiB
Python
"""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
|
|
|
|
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: 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) -> 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 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 "<root>",
|
|
start_index,
|
|
limit,
|
|
include_item_types or "<all>",
|
|
search or "<none>",
|
|
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 "<root>",
|
|
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}"
|