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()