Developer
277 lines
11 KiB
Python
277 lines
11 KiB
Python
"""Jellyseerr HTTP API client.
|
|
|
|
Jellyseerr is optional. When configured, it can enrich the Jellyfin user list
|
|
with email addresses, avatars, permissions, and request metadata.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Any
|
|
|
|
import requests
|
|
|
|
from media_library_viewer_api.clients.http_timeout import DEFAULT_READ_TIMEOUT, http_timeout
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Jellyseerr numeric status enums (see Overseerr/Jellyseerr source).
|
|
_REQUEST_STATUS: dict[int, str] = {1: "pending", 2: "approved", 3: "declined"}
|
|
_MEDIA_STATUS: dict[int, str] = {
|
|
1: "unknown",
|
|
2: "pending",
|
|
3: "processing",
|
|
4: "partially_available",
|
|
5: "available",
|
|
}
|
|
_REQUEST_TYPE: dict[int, str] = {1: "movie", 2: "tv"}
|
|
|
|
|
|
def _label(value: Any, table: dict[int, str]) -> str:
|
|
try:
|
|
return table.get(int(value), str(value))
|
|
except (TypeError, ValueError):
|
|
return str(value) if value is not None else ""
|
|
|
|
|
|
class JellyseerrClient:
|
|
"""Small wrapper around the Jellyseerr REST API."""
|
|
|
|
def __init__(self, base_url: str, api_key: str, timeout: float = DEFAULT_READ_TIMEOUT):
|
|
if not base_url:
|
|
raise ValueError("Jellyseerr URL is required")
|
|
if not api_key:
|
|
raise ValueError("Jellyseerr API key is required")
|
|
|
|
self.base_url = base_url.rstrip("/")
|
|
if self.base_url.endswith("/api/v1"):
|
|
self.base_url = self.base_url[:-7]
|
|
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-Api-Key": api_key,
|
|
"Accept": "application/json",
|
|
}
|
|
)
|
|
self._title_cache: dict[tuple[str, str], str] = {}
|
|
|
|
def get(self, path: str, **params: Any) -> Any:
|
|
"""GET a Jellyseerr 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("Jellyseerr GET %s params=%s", path, sorted(clean_params.keys()))
|
|
response = self.session.get(f"{self.base_url}/api/v1{path}", params=clean_params, timeout=self.timeout)
|
|
try:
|
|
response.raise_for_status()
|
|
except requests.HTTPError as exc:
|
|
detail = response.text[:500]
|
|
logger.warning("Jellyseerr 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("Jellyseerr GET %s ok status=%s", path, response.status_code)
|
|
return response.json()
|
|
|
|
def absolute_url(self, path: str | None) -> str:
|
|
"""Return an absolute URL for Jellyseerr-relative assets."""
|
|
if not path:
|
|
return ""
|
|
if path.startswith("http://") or path.startswith("https://"):
|
|
return path
|
|
if not path.startswith("/"):
|
|
path = f"/{path}"
|
|
return f"{self.base_url}{path}"
|
|
|
|
def _resolve_title(self, media_type: Any, tmdb_id: Any) -> str:
|
|
"""Resolve a media title via /movie/{tmdbId} or /tv/{tmdbId}, cached.
|
|
|
|
Jellyseerr's /request list doesn't include titles; they live on the
|
|
Movie/Series records. Cached per (type, tmdbId) so repeated polls reuse.
|
|
"""
|
|
if not tmdb_id:
|
|
return ""
|
|
key = (str(media_type or ""), str(tmdb_id))
|
|
if key in self._title_cache:
|
|
return self._title_cache[key]
|
|
try:
|
|
is_tv = str(media_type) in ("2", "tv")
|
|
data = self.get(f"/{'tv' if is_tv else 'movie'}/{tmdb_id}")
|
|
title = str(data.get("name" if is_tv else "title") or "")
|
|
except Exception:
|
|
title = ""
|
|
self._title_cache[key] = title
|
|
return title
|
|
|
|
def jellyfin_users(self) -> list[dict[str, Any]]:
|
|
"""Return Jellyfin-linked users known to Jellyseerr.
|
|
|
|
Jellyseerr has used both a top-level list payload and a wrapped
|
|
`{ "users": [...] }` payload in different versions/docs, so accept
|
|
either shape.
|
|
"""
|
|
payload = self.get("/settings/jellyfin/users")
|
|
if isinstance(payload, list):
|
|
users = [item for item in payload if isinstance(item, dict)]
|
|
logger.info("Jellyseerr returned %s Jellyfin-linked users", len(users))
|
|
return users
|
|
if isinstance(payload, dict):
|
|
users = payload.get("users")
|
|
if isinstance(users, list):
|
|
mapped = [item for item in users if isinstance(item, dict)]
|
|
logger.info("Jellyseerr returned %s Jellyfin-linked users (wrapped payload)", len(mapped))
|
|
return mapped
|
|
logger.info("Jellyseerr returned no Jellyfin-linked users")
|
|
return []
|
|
|
|
def users(self, page_size: int = 1000) -> list[dict[str, Any]]:
|
|
"""Return Jellyseerr users via the paginated /user list endpoint.
|
|
|
|
Jellyseerr's list endpoint uses ``take`` and ``skip`` query params,
|
|
not ``page``.
|
|
"""
|
|
results: list[dict[str, Any]] = []
|
|
take = max(1, int(page_size))
|
|
skip = 0
|
|
total_results: int | None = None
|
|
|
|
while True:
|
|
payload = self.get("/user", take=take, skip=skip)
|
|
if not isinstance(payload, dict):
|
|
return results
|
|
|
|
page_results = payload.get("results") or []
|
|
page_items = (
|
|
[item for item in page_results if isinstance(item, dict)] if isinstance(page_results, list) else []
|
|
)
|
|
results.extend(page_items)
|
|
|
|
page_info = payload.get("pageInfo") or {}
|
|
if isinstance(page_info, dict):
|
|
try:
|
|
page_total = int(page_info.get("results") or 0)
|
|
if page_total:
|
|
total_results = page_total
|
|
except (TypeError, ValueError):
|
|
pass
|
|
|
|
logger.debug(
|
|
"Jellyseerr user page skip=%s take=%s -> %s results (total=%s)",
|
|
skip,
|
|
take,
|
|
len(page_items),
|
|
total_results if total_results is not None else "unknown",
|
|
)
|
|
|
|
if not page_items:
|
|
break
|
|
skip += len(page_items)
|
|
if len(page_items) < take:
|
|
break
|
|
if total_results is not None and skip >= total_results:
|
|
break
|
|
|
|
logger.info("Jellyseerr returned %s users", len(results))
|
|
return results
|
|
|
|
def request_count(self) -> dict[str, int]:
|
|
"""Return normalized request counts from /api/v1/request/count.
|
|
|
|
Jellyseerr reports pending/approved/declined/processing/available/total.
|
|
Missing keys default to 0 so callers can rely on a stable shape.
|
|
"""
|
|
payload = self.get("/request/count")
|
|
if not isinstance(payload, dict):
|
|
payload = {}
|
|
keys = ("total", "pending", "approved", "declined", "processing", "available")
|
|
counts = {k: int(payload.get(k) or 0) for k in keys}
|
|
logger.info(
|
|
"Jellyseerr request counts total=%s pending=%s processing=%s",
|
|
counts["total"],
|
|
counts["pending"],
|
|
counts["processing"],
|
|
)
|
|
return counts
|
|
|
|
def recent_requests(self, take: int = 20) -> list[dict[str, Any]]:
|
|
"""Return the most recently modified requests with resolved titles."""
|
|
take = max(1, min(int(take), 100))
|
|
payload = self.get("/request", sort="modified", skip=0, take=take)
|
|
if not isinstance(payload, dict):
|
|
return []
|
|
results = payload.get("results") or []
|
|
items = [r for r in results if isinstance(r, dict)] if isinstance(results, list) else []
|
|
mapped: list[dict[str, Any]] = []
|
|
for r in items:
|
|
media = r.get("media") or {}
|
|
tmdb_id = media.get("tmdbId")
|
|
name = r.get("title") or media.get("title") or media.get("name") or ""
|
|
if not name and tmdb_id:
|
|
name = self._resolve_title(r.get("type"), tmdb_id)
|
|
if not name:
|
|
name = media.get("externalServiceSlug") or ""
|
|
mapped.append(
|
|
{
|
|
"id": r.get("id"),
|
|
"type": _label(r.get("type"), _REQUEST_TYPE),
|
|
"name": name or "—",
|
|
"status": _label(r.get("status"), _REQUEST_STATUS),
|
|
"media_status": _label((media or {}).get("status"), _MEDIA_STATUS),
|
|
"created_at": r.get("createdAt"),
|
|
}
|
|
)
|
|
return mapped
|
|
|
|
def open_requests(self, max_per_filter: int = 100) -> list[dict[str, Any]]:
|
|
"""Return open (pending + approved) requests with resolved titles.
|
|
|
|
Fetches pending and approved requests via Jellyseerr's filter param
|
|
(not all 800+ historical requests), then resolves titles from
|
|
/movie/{tmdbId} or /tv/{tmdbId}. Titles are cached on the client so
|
|
subsequent polls are instant.
|
|
"""
|
|
results: list[dict[str, Any]] = []
|
|
take = 50
|
|
for filter_val in ("pending", "approved"):
|
|
skip = 0
|
|
while skip < max_per_filter:
|
|
payload = self.get("/request", filter=filter_val, sort="added", skip=skip, take=take)
|
|
if not isinstance(payload, dict):
|
|
break
|
|
page = payload.get("results") or []
|
|
items = [r for r in page if isinstance(r, dict)] if isinstance(page, list) else []
|
|
for r in items:
|
|
media = r.get("media") or {}
|
|
tmdb_id = media.get("tmdbId") or r.get("tmdbId")
|
|
# Diagnostic: log the first request's shape once so we can verify tmdbId.
|
|
if not results and filter_val == "pending":
|
|
logger.info(
|
|
"Jellyseerr request sample: keys=%s media_keys=%s tmdbId=%s",
|
|
sorted(r.keys()),
|
|
sorted(media.keys()) if isinstance(media, dict) else "N/A",
|
|
tmdb_id,
|
|
)
|
|
name = r.get("title") or media.get("title") or media.get("name") or ""
|
|
if not name and tmdb_id:
|
|
name = self._resolve_title(r.get("type"), tmdb_id)
|
|
if not name:
|
|
name = media.get("externalServiceSlug") or ""
|
|
results.append(
|
|
{
|
|
"id": r.get("id"),
|
|
"type": _label(r.get("type"), _REQUEST_TYPE),
|
|
"name": name or "—",
|
|
"status": _label(r.get("status"), _REQUEST_STATUS),
|
|
"media_status": _label((media or {}).get("status"), _MEDIA_STATUS),
|
|
"created_at": r.get("createdAt"),
|
|
}
|
|
)
|
|
if len(items) < take:
|
|
break
|
|
skip += len(items)
|
|
results.sort(key=lambda r: r.get("created_at") or 0, reverse=True)
|
|
logger.info("Jellyseerr returned %s open requests (with titles)", len(results))
|
|
return results
|