87 lines
4.3 KiB
Python
87 lines
4.3 KiB
Python
"""Jellystat API adapter. Credentials and raw history never leave the backend."""
|
|
|
|
import asyncio
|
|
import json
|
|
import re
|
|
from datetime import datetime
|
|
|
|
import httpx
|
|
|
|
from .base import ApiClient
|
|
|
|
|
|
class JellystatError(Exception):
|
|
pass
|
|
|
|
|
|
class HistoryLimitError(JellystatError):
|
|
pass
|
|
|
|
|
|
def same_user_id(left, right) -> bool:
|
|
return bool(left and right) and str(left).replace("-", "").lower() == str(right).replace("-", "").lower()
|
|
|
|
|
|
class JellystatClient(ApiClient):
|
|
PAGE_SIZE = 200
|
|
MAX_PAGES = 50
|
|
|
|
def configured(self) -> bool:
|
|
return bool(self.base_url and self.api_key)
|
|
|
|
async def _read(self, client: httpx.AsyncClient, method: str, path: str, **kwargs):
|
|
try:
|
|
response = await client.request(method, f"{self.base_url}{path}",
|
|
headers={"x-api-token": self.api_key}, **kwargs)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except (httpx.HTTPError, ValueError) as exc:
|
|
raise JellystatError("Jellystat did not return a valid response") from exc
|
|
|
|
async def test_connection(self) -> dict:
|
|
# This protected endpoint confirms API authentication without returning user data.
|
|
async with httpx.AsyncClient(timeout=10.0) as client:
|
|
result = await self._read(client, "GET", "/api/getLibraries")
|
|
if not isinstance(result, list):
|
|
raise JellystatError("Jellystat returned an unexpected library response")
|
|
return {"connected": True}
|
|
|
|
async def get_user_history(self, user_id: str, start: datetime, end: datetime) -> tuple[list, list]:
|
|
if not re.fullmatch(r"[A-Za-z0-9_-]{1,128}", user_id):
|
|
raise JellystatError("Invalid linked Jellyfin identity")
|
|
# Only fixed, user-scoped endpoints are used. Never pass browser search/filters through.
|
|
filters = json.dumps([{"field": "ActivityDateInserted", "min": start.isoformat(), "max": end.isoformat()}])
|
|
try:
|
|
async with asyncio.timeout(30):
|
|
async with httpx.AsyncClient(timeout=10.0) as client:
|
|
libraries = await self._read(client, "GET", "/api/getLibraries")
|
|
if not isinstance(libraries, list) or any(not isinstance(row, dict) for row in libraries):
|
|
raise JellystatError("Jellystat returned an unexpected library response")
|
|
history = []
|
|
for page in range(1, self.MAX_PAGES + 1):
|
|
payload = await self._read(client, "POST", "/api/getUserHistory",
|
|
json={"userid": user_id}, params={"page": page, "size": self.PAGE_SIZE,
|
|
"sort": "ActivityDateInserted", "desc": "true", "filters": filters})
|
|
if not isinstance(payload, dict) or not isinstance(payload.get("results"), list):
|
|
raise JellystatError("Jellystat returned an unexpected history response")
|
|
rows = payload["results"]
|
|
try:
|
|
pages = int(payload["pages"])
|
|
except (KeyError, TypeError, ValueError) as exc:
|
|
raise JellystatError("Jellystat did not return history pagination") from exc
|
|
if pages < 0 or (pages == 0 and rows) or len(rows) > self.PAGE_SIZE:
|
|
raise JellystatError("Jellystat returned invalid history pagination")
|
|
if pages > self.MAX_PAGES:
|
|
raise HistoryLimitError("Select a shorter period to view this history")
|
|
for row in rows:
|
|
if not isinstance(row, dict) or not same_user_id(row.get("UserId"), user_id):
|
|
raise JellystatError("Jellystat returned history for an unexpected account")
|
|
history.extend(rows)
|
|
if page >= pages:
|
|
return history, libraries
|
|
if not rows:
|
|
raise JellystatError("Jellystat returned incomplete history")
|
|
except TimeoutError as exc:
|
|
raise JellystatError("Jellystat took too long to return history") from exc
|
|
raise HistoryLimitError("Select a shorter period to view this history")
|