202 lines
8.2 KiB
Python
202 lines
8.2 KiB
Python
from typing import Any, Dict, Optional
|
|
import httpx
|
|
from .base import ApiClient
|
|
|
|
|
|
class JellyfinClient(ApiClient):
|
|
def __init__(self, base_url: Optional[str], api_key: Optional[str]):
|
|
super().__init__(base_url, api_key)
|
|
|
|
def configured(self) -> bool:
|
|
return bool(self.base_url and self.api_key)
|
|
|
|
def _emby_headers(self) -> Dict[str, str]:
|
|
return {"X-Emby-Token": self.api_key} if self.api_key else {}
|
|
|
|
@staticmethod
|
|
def _extract_user_id(payload: Any) -> Optional[str]:
|
|
if not isinstance(payload, dict):
|
|
return None
|
|
candidate = payload.get("User") if isinstance(payload.get("User"), dict) else payload
|
|
if not isinstance(candidate, dict):
|
|
return None
|
|
for key in ("Id", "id", "UserId", "userId"):
|
|
value = candidate.get(key)
|
|
if value is None:
|
|
continue
|
|
if isinstance(value, (str, int)):
|
|
text = str(value).strip()
|
|
if text:
|
|
return text
|
|
return None
|
|
|
|
async def get_users(self) -> Optional[Dict[str, Any]]:
|
|
if not self.base_url:
|
|
return None
|
|
url = f"{self.base_url}/Users"
|
|
headers = self._emby_headers()
|
|
async with httpx.AsyncClient(timeout=10.0) as client:
|
|
response = await client.get(url, headers=headers)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
async def get_user(self, user_id: str) -> Optional[Dict[str, Any]]:
|
|
if not self.base_url or not self.api_key:
|
|
return None
|
|
url = f"{self.base_url}/Users/{user_id}"
|
|
headers = self._emby_headers()
|
|
async with httpx.AsyncClient(timeout=10.0) as client:
|
|
response = await client.get(url, headers=headers)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
async def find_user_by_name(self, username: str) -> Optional[Dict[str, Any]]:
|
|
users = await self.get_users()
|
|
if not isinstance(users, list):
|
|
return None
|
|
target = username.strip().lower()
|
|
for user in users:
|
|
if not isinstance(user, dict):
|
|
continue
|
|
name = str(user.get("Name") or "").strip().lower()
|
|
if name and name == target:
|
|
return user
|
|
return None
|
|
|
|
async def authenticate_by_name(self, username: str, password: str) -> Optional[Dict[str, Any]]:
|
|
if not self.base_url:
|
|
return None
|
|
url = f"{self.base_url}/Users/AuthenticateByName"
|
|
headers = self._emby_headers()
|
|
payload = {"Username": username, "Pw": password}
|
|
async with httpx.AsyncClient(timeout=10.0) as client:
|
|
response = await client.post(url, headers=headers, json=payload)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
async def create_user(self, username: str) -> Optional[Dict[str, Any]]:
|
|
if not self.base_url or not self.api_key:
|
|
return None
|
|
url = f"{self.base_url}/Users/New"
|
|
headers = self._emby_headers()
|
|
payload = {"Name": username}
|
|
async with httpx.AsyncClient(timeout=10.0) as client:
|
|
response = await client.post(url, headers=headers, json=payload)
|
|
response.raise_for_status()
|
|
if not response.content:
|
|
return None
|
|
return response.json()
|
|
|
|
async def set_user_password(self, user_id: str, password: str) -> None:
|
|
if not self.base_url or not self.api_key:
|
|
return None
|
|
headers = self._emby_headers()
|
|
payloads = [
|
|
{"CurrentPw": "", "NewPw": password},
|
|
{"CurrentPwd": "", "NewPw": password},
|
|
{"CurrentPw": "", "NewPw": password, "ResetPassword": False},
|
|
{"CurrentPwd": "", "NewPw": password, "ResetPassword": False},
|
|
{"NewPw": password, "ResetPassword": False},
|
|
]
|
|
paths = [
|
|
f"/Users/{user_id}/Password",
|
|
f"/Users/{user_id}/EasyPassword",
|
|
]
|
|
last_error: Exception | None = None
|
|
async with httpx.AsyncClient(timeout=10.0) as client:
|
|
for path in paths:
|
|
url = f"{self.base_url}{path}"
|
|
for payload in payloads:
|
|
try:
|
|
response = await client.post(url, headers=headers, json=payload)
|
|
response.raise_for_status()
|
|
return
|
|
except httpx.HTTPStatusError as exc:
|
|
last_error = exc
|
|
continue
|
|
except Exception as exc:
|
|
last_error = exc
|
|
continue
|
|
if last_error:
|
|
raise last_error
|
|
|
|
async def set_user_disabled(self, user_id: str, disabled: bool = True) -> None:
|
|
if not self.base_url or not self.api_key:
|
|
return None
|
|
user = await self.get_user(user_id)
|
|
if not isinstance(user, dict):
|
|
raise RuntimeError("Jellyfin user details not available")
|
|
policy = user.get("Policy") if isinstance(user.get("Policy"), dict) else {}
|
|
payload = {**policy, "IsDisabled": bool(disabled)}
|
|
url = f"{self.base_url}/Users/{user_id}/Policy"
|
|
headers = self._emby_headers()
|
|
async with httpx.AsyncClient(timeout=10.0) as client:
|
|
response = await client.post(url, headers=headers, json=payload)
|
|
response.raise_for_status()
|
|
|
|
async def delete_user(self, user_id: str) -> None:
|
|
if not self.base_url or not self.api_key:
|
|
return None
|
|
url = f"{self.base_url}/Users/{user_id}"
|
|
headers = self._emby_headers()
|
|
async with httpx.AsyncClient(timeout=10.0) as client:
|
|
response = await client.delete(url, headers=headers)
|
|
response.raise_for_status()
|
|
|
|
async def create_user_with_password(self, username: str, password: str) -> Optional[Dict[str, Any]]:
|
|
created = await self.create_user(username)
|
|
user_id = self._extract_user_id(created)
|
|
if not user_id:
|
|
users = await self.get_users()
|
|
if isinstance(users, list):
|
|
for user in users:
|
|
if not isinstance(user, dict):
|
|
continue
|
|
name = str(user.get("Name") or "").strip()
|
|
if name.lower() == username.strip().lower():
|
|
created = user
|
|
user_id = self._extract_user_id(user)
|
|
break
|
|
if not user_id:
|
|
raise RuntimeError("Jellyfin user created but user ID was not returned")
|
|
await self.set_user_password(user_id, password)
|
|
return created
|
|
|
|
async def search_items(
|
|
self, term: str, item_types: Optional[list[str]] = None, limit: int = 20
|
|
) -> Optional[Dict[str, Any]]:
|
|
if not self.base_url or not self.api_key:
|
|
return None
|
|
url = f"{self.base_url}/Items"
|
|
params = {
|
|
"SearchTerm": term,
|
|
"IncludeItemTypes": ",".join(item_types or []),
|
|
"Recursive": "true",
|
|
"Limit": limit,
|
|
}
|
|
headers = self._emby_headers()
|
|
async with httpx.AsyncClient(timeout=10.0) as client:
|
|
response = await client.get(url, headers=headers, params=params)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
async def get_system_info(self) -> Optional[Dict[str, Any]]:
|
|
if not self.base_url or not self.api_key:
|
|
return None
|
|
url = f"{self.base_url}/System/Info"
|
|
headers = self._emby_headers()
|
|
async with httpx.AsyncClient(timeout=10.0) as client:
|
|
response = await client.get(url, headers=headers)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
async def refresh_library(self, recursive: bool = True) -> None:
|
|
if not self.base_url or not self.api_key:
|
|
return None
|
|
url = f"{self.base_url}/Library/Refresh"
|
|
headers = self._emby_headers()
|
|
params = {"Recursive": "true" if recursive else "false"}
|
|
async with httpx.AsyncClient(timeout=10.0) as client:
|
|
response = await client.post(url, headers=headers, params=params)
|
|
response.raise_for_status()
|