119 lines
6.0 KiB
Python
119 lines
6.0 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 check_user_ids(self, user_ids: list[str]) -> dict:
|
|
"""Read metadata for known identities; never scan everyone's playback history."""
|
|
if not self.configured():
|
|
return {user_id: {"state": "not_configured"} for user_id in user_ids}
|
|
results = {user_id: {"state": "unavailable"} for user_id in user_ids}
|
|
semaphore = asyncio.Semaphore(6)
|
|
async with httpx.AsyncClient(timeout=8.0) as client:
|
|
async def check(user_id):
|
|
if not re.fullmatch(r"[a-f0-9]{32}", user_id):
|
|
return
|
|
async with semaphore:
|
|
try:
|
|
response = await client.post(f"{self.base_url}/api/getUserDetails",
|
|
headers={"x-api-token": self.api_key}, json={"userid": user_id})
|
|
if response.status_code == 404 or (response.status_code == 200 and not response.content.strip()):
|
|
results[user_id] = {"state": "missing"}
|
|
return
|
|
response.raise_for_status()
|
|
row = response.json()
|
|
if row is None:
|
|
results[user_id] = {"state": "missing"}
|
|
elif isinstance(row, dict) and same_user_id(row.get("Id"), user_id):
|
|
results[user_id] = {"state": "matched", "id": user_id, "name": str(row.get("Name") or "")[:200]}
|
|
except (httpx.HTTPError, ValueError):
|
|
pass
|
|
try:
|
|
async with asyncio.timeout(25):
|
|
await asyncio.gather(*(check(user_id) for user_id in user_ids))
|
|
except TimeoutError:
|
|
pass
|
|
return results
|
|
|
|
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")
|