124 lines
4.5 KiB
Python
124 lines
4.5 KiB
Python
import httpx
|
|
import os
|
|
from typing import Optional, List, Dict, Any
|
|
|
|
|
|
class JellyfinClient:
|
|
"""A client to interact with the Jellyfin API."""
|
|
|
|
def __init__(self, base_url: str, api_key: str):
|
|
"""
|
|
Initializes the JellyfinClient.
|
|
|
|
Args:
|
|
base_url (str): The base URL of the Jellyfin server.
|
|
api_key (str): The API key for authentication.
|
|
"""
|
|
self.base_url = base_url.rstrip("/")
|
|
self.api_key = api_key
|
|
self.headers = {"Authorization": f"MediaBrowser Token=\"{self.api_key}\"", "Accept": "application/json"}
|
|
|
|
async def _request(
|
|
self, method: str, endpoint: str, params: Optional[Dict[str, Any]] = None
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Sends an authenticated request to the Jellyfin API.
|
|
|
|
Args:
|
|
method (str): The HTTP method (GET, POST, etc.).
|
|
endpoint (str): The API endpoint.
|
|
params (Optional[Dict[str, Any]]): Query parameters.
|
|
|
|
Returns:
|
|
Dict[str, Any]: The JSON response from the API.
|
|
|
|
Raises:
|
|
httpx.HTTPStatusError: If the request was unsuccessful.
|
|
"""
|
|
url = f"{self.base_url}/{endpoint.lstrip('/')}"
|
|
async with httpx.AsyncClient(headers=self.headers) as client:
|
|
response = await client.request(method, url, params=params)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
async def search_items(
|
|
self, query: str, item_types: Optional[List[str]] = None
|
|
) -> List[Dict[str, Any]]:
|
|
"""
|
|
Searches for items in the Jellyfin library.
|
|
|
|
Args:
|
|
query (str): The search term.
|
|
item_types (Optional[List[str]]): A list of item types to filter by (e.g., ["Movie", "Series"]).
|
|
|
|
Returns:
|
|
List[Dict[str, Any]]: A list of matching items.
|
|
"""
|
|
params = {
|
|
"searchTerm": query,
|
|
"Recursive": "true",
|
|
}
|
|
|
|
if item_types:
|
|
params["IncludeItemTypes"] = ",".join(item_types)
|
|
|
|
response = await self._request("GET", "/Items", params=params)
|
|
# Jellyfin returns an object containing 'Items' list.
|
|
return response.get("Items", [])
|
|
|
|
async def list_active_sessions(self) -> List[Dict[str, Any]]:
|
|
"""
|
|
Retrieves a list of active user sessions.
|
|
|
|
Returns:
|
|
List[Dict[str, Any]]: A list of active sessions.
|
|
"""
|
|
response = await self._request("GET", "/Sessions")
|
|
# Jellyfin returns an object containing 'Sessions' list.
|
|
return response
|
|
|
|
async def search_by_genre(
|
|
self, genre: str, item_types: Optional[List[str]] = None
|
|
) -> List[Dict[str, Any]]:
|
|
params = {"Genres": genre, "Recursive": "true"}
|
|
if item_types:
|
|
params["IncludeItemTypes"] = ",".join(item_types)
|
|
response = await self._request("GET", "/Items", params=params)
|
|
return response.get("Items", [])
|
|
|
|
async def search_by_director(
|
|
self, director: str, item_types: Optional[List[str]] = None
|
|
) -> List[Dict[str, Any]]:
|
|
params = {"Person": director, "PersonTypes": "Director", "Recursive": "true"}
|
|
if item_types:
|
|
params["IncludeItemTypes"] = ",".join(item_types)
|
|
response = await self._request("GET", "/Items", params=params)
|
|
return response.get("Items", [])
|
|
|
|
async def search_by_cast(
|
|
self, person: str, item_types: Optional[List[str]] = None
|
|
) -> List[Dict[str, Any]]:
|
|
params = {"Person": person, "PersonTypes": "Actor", "Recursive": "true"}
|
|
if item_types:
|
|
params["IncludeItemTypes"] = ",".join(item_types)
|
|
response = await self._request("GET", "/Items", params=params)
|
|
return response.get("Items", [])
|
|
|
|
async def get_users(self) -> List[Dict[str, Any]]:
|
|
return await self._request("GET", "/Users")
|
|
|
|
async def get_user_item_data(self, user_id: str, item_id: str) -> Dict[str, Any]:
|
|
return await self._request("GET", f"/Users/{user_id}/Items/{item_id}/UserData")
|
|
|
|
async def get_series_seasons(self, series_id: str) -> List[Dict[str, Any]]:
|
|
response = await self._request(
|
|
"GET", f"/Shows/{series_id}/Seasons", params={"Fields": "ChildCount"}
|
|
)
|
|
return response.get("Items", [])
|
|
|
|
async def get_season_episodes(self, series_id: str, season_number: int) -> List[Dict[str, Any]]:
|
|
response = await self._request(
|
|
"GET", f"/Shows/{series_id}/Episodes", params={"season": season_number}
|
|
)
|
|
return response.get("Items", [])
|