import aiohttp import asyncio import config import urllib.parse import json import logging from datetime import datetime, timedelta from typing import Dict, List, Optional, Any, Union logger = logging.getLogger('jellyseerr_api') class JellyseerrAPI: """Client for interacting with the Jellyseerr API. This class handles authentication, session management, and provides methods for interacting with various Jellyseerr API endpoints. """ def __init__(self, base_url: str, email: str = None, password: str = None): """Initialize the Jellyseerr API client. Args: base_url: The base URL of the Jellyseerr instance email: Email for authentication with Jellyseerr password: Password for authentication with Jellyseerr """ self.base_url = base_url.rstrip('/') self.email = email self.password = password self.headers = { 'Content-Type': 'application/json' } self.session: Optional[aiohttp.ClientSession] = None self.cookie_jar: Optional[aiohttp.CookieJar] = None self.auth_cookie: Optional[str] = None self.last_login: Optional[datetime] = None async def __aenter__(self): if not self.cookie_jar: self.cookie_jar = aiohttp.CookieJar() self.session = aiohttp.ClientSession(cookie_jar=self.cookie_jar) await self.login() return self async def __aexit__(self, exc_type, exc_val, exc_tb): if self.session: await self.session.close() async def login(self) -> bool: """Log in to Jellyseerr using local authentication. Returns: bool: True if login was successful Raises: Exception: If authentication fails for any reason """ # Skip login if we have a valid cookie and it's not expired cookie_expiry_days = config.AUTH_COOKIE_EXPIRY if (self.auth_cookie and self.last_login and datetime.now() < self.last_login + timedelta(days=cookie_expiry_days)): logger.debug("Using existing auth cookie") return True if not self.session: if not self.cookie_jar: self.cookie_jar = aiohttp.CookieJar() self.session = aiohttp.ClientSession(cookie_jar=self.cookie_jar) try: login_data = { 'email': self.email, 'password': self.password } logger.info(f"Logging in to Jellyseerr with email: {self.email}") logger.debug(f"Login URL: {self.base_url}/api/v1/auth/local") # Print request info for debugging logger.debug(f"Request headers: {self.session.headers}") logger.debug(f"Login payload (without password): {{'email': '{self.email}'}}") async with self.session.post( f"{self.base_url}/api/v1/auth/local", json=login_data ) as response: response_text = await response.text() logger.debug(f"Login response status: {response.status}") logger.debug(f"Login response headers: {response.headers}") logger.debug(f"Login response body: {response_text}") if response.status == 200: # Login successful, cookies are automatically stored in the cookie jar self.last_login = datetime.now() logger.info("Successfully logged in to Jellyseerr") return True else: # Try to parse response as JSON, but handle case where it's not JSON try: error_data = json.loads(response_text) error_message = error_data.get('message', 'Unknown error') except json.JSONDecodeError: error_message = f"Non-JSON response (Status {response.status}): {response_text}" logger.error(f"Login failed: {error_message}") if response.status == 401: raise Exception(f"Authentication failed: Invalid email or password") elif response.status == 403: raise Exception(f"Authentication failed: Access denied. Account may be disabled or lacks permissions") elif response.status == 500: raise Exception(f"Authentication failed: Server error. Check Jellyseerr logs") else: raise Exception(f"Authentication failed ({response.status}): {error_message}") except aiohttp.ClientConnectorError as e: logger.error(f"Connection error: {str(e)}") raise Exception(f"Failed to connect to Jellyseerr at {self.base_url}: {str(e)}") except aiohttp.ClientError as e: logger.error(f"Client error: {str(e)}") raise Exception(f"HTTP client error: {str(e)}") except Exception as e: logger.error(f"Login error: {str(e)}", exc_info=True) raise Exception(f"Failed to authenticate with Jellyseerr: {str(e)}") async def _request(self, method: str, endpoint: str, params: Optional[Dict[str, Any]] = None, data: Optional[Dict[str, Any]] = None) -> Any: """Make a request to the Jellyseerr API. Args: method: HTTP method (GET, POST, etc.) endpoint: API endpoint (without the base URL and /api/v1) params: Query parameters data: JSON request body Returns: Any: The JSON response data, or None for 204 responses Raises: Exception: For API errors, connection issues, or other failures """ # Ensure we're logged in if not self.last_login: try: await self.login() except Exception as e: logger.warning(f"Authentication failed, attempting request without authentication: {str(e)}") # Continue anyway - some endpoints might work without auth # Create session if it doesn't exist if self.session is None: if not self.cookie_jar: self.cookie_jar = aiohttp.CookieJar() self.session = aiohttp.ClientSession(cookie_jar=self.cookie_jar) # Debug info logger.debug(f"Making request: {method} {self.base_url}/api/v1{endpoint}") if params: logger.debug(f"Request params: {params}") if data: logger.debug(f"Request data: {data}") url = f"{self.base_url}/api/v1{endpoint}" # URL encode parameters - only encode string values, not numbers or booleans if params: encoded_params = {} for key, value in params.items(): if isinstance(value, str): encoded_params[key] = urllib.parse.quote(value) else: encoded_params[key] = value params = encoded_params try: # Set timeout for the request timeout = aiohttp.ClientTimeout(total=config.REQUEST_TIMEOUT) # Attempt the request async with self.session.request( method, url, headers=self.headers, params=params, json=data, timeout=timeout ) as response: if response.status == 204: # No content return None try: response_data = await response.json() # If unauthorized, try to login again and retry if response.status == 401: logger.warning("Received unauthorized response, attempting to re-login") try: await self.login() # Retry the request after re-login async with self.session.request( method, url, headers=self.headers, params=params, json=data, timeout=timeout ) as retry_response: logger.debug(f"Retry response status: {retry_response.status}") if retry_response.status == 204: return None try: retry_data = await retry_response.json() if not retry_response.ok: error_message = retry_data.get('message', 'Unknown error') raise Exception(f"API Error ({retry_response.status}): {error_message}") return retry_data except aiohttp.ContentTypeError: # Handle non-JSON responses retry_text = await retry_response.text() logger.error(f"Non-JSON response on retry: {retry_text}") raise Exception(f"API returned non-JSON response: {retry_text[:100]}...") except Exception as login_err: logger.error(f"Re-login failed: {str(login_err)}") raise Exception(f"Authentication error: {str(login_err)}") if not response.ok: error_message = response_data.get('message', 'Unknown error') raise Exception(f"API Error ({response.status}): {error_message}") return response_data except aiohttp.ContentTypeError: # Not JSON, get the text response error_text = await response.text() raise Exception(f"API Error ({response.status}): Not a valid JSON response: {error_text[:100]}...") except aiohttp.ClientResponseError as e: raise Exception(f"API Error ({response.status}): {str(e)}") except aiohttp.ClientConnectorError as e: logger.error(f"Connection error: {str(e)}") raise Exception(f"Connection error: Could not connect to {self.base_url}: {str(e)}") except aiohttp.ClientError as e: logger.error(f"HTTP client error: {str(e)}") raise Exception(f"Connection error: {str(e)}") except asyncio.TimeoutError: logger.error(f"Request timed out after {config.REQUEST_TIMEOUT} seconds") raise Exception(f"Request timed out after {config.REQUEST_TIMEOUT} seconds. Check your network connection and Jellyseerr server status.") # Search endpoints async def search(self, query: str, page: int = 1, language: str = 'en') -> Dict[str, Any]: """Search for movies, TV shows, or people. Args: query: The search term page: Page number for results language: Language code for results (e.g., 'en', 'fr') Returns: Dict containing search results """ params = { 'query': query, 'page': page, 'language': language } return await self._request('GET', '/search', params=params) # Media information endpoints async def get_movie_details(self, movie_id: int, language: str = 'en') -> Dict[str, Any]: """Get detailed information about a movie. Args: movie_id: TMDB ID of the movie language: Language code for results (e.g., 'en', 'fr') Returns: Dict containing movie details """ params = {'language': language} return await self._request('GET', f'/movie/{movie_id}', params=params) async def get_tv_details(self, tv_id: int, language: str = 'en') -> Dict[str, Any]: """Get detailed information about a TV show. Args: tv_id: TMDB ID of the TV show language: Language code for results (e.g., 'en', 'fr') Returns: Dict containing TV show details """ params = {'language': language} return await self._request('GET', f'/tv/{tv_id}', params=params) async def get_season_details(self, tv_id: int, season_id: int, language: str = 'en') -> Dict[str, Any]: """Get detailed information about a TV season. Args: tv_id: TMDB ID of the TV show season_id: Season number language: Language code for results (e.g., 'en', 'fr') Returns: Dict containing season details """ params = {'language': language} return await self._request('GET', f'/tv/{tv_id}/season/{season_id}', params=params) # Recommendation endpoints async def get_movie_recommendations(self, movie_id: int, page: int = 1, language: str = 'en') -> Dict[str, Any]: """Get movie recommendations based on a movie. Args: movie_id: TMDB ID of the movie page: Page number for results language: Language code for results (e.g., 'en', 'fr') Returns: Dict containing recommended movies """ params = { 'page': page, 'language': language } return await self._request('GET', f'/movie/{movie_id}/recommendations', params=params) async def get_tv_recommendations(self, tv_id: int, page: int = 1, language: str = 'en') -> Dict[str, Any]: """Get TV show recommendations based on a TV show. Args: tv_id: TMDB ID of the TV show page: Page number for results language: Language code for results (e.g., 'en', 'fr') Returns: Dict containing recommended TV shows """ params = { 'page': page, 'language': language } return await self._request('GET', f'/tv/{tv_id}/recommendations', params=params) # Request endpoints async def get_requests(self, filter_status: str = 'all', page: int = 1, page_size: int = 10) -> Dict: """Get all requests with pagination""" params = { 'filter': filter_status, 'take': page_size, 'skip': (page - 1) * page_size } return await self._request('GET', '/request', params=params) async def create_request(self, media_type: str, media_id: int, seasons: Union[List[int], str] = None, is_4k: bool = False) -> Dict[str, Any]: """Create a new media request. Args: media_type: Type of media ('movie' or 'tv') media_id: TMDB ID of the media seasons: For TV shows, specific seasons to request or 'all' is_4k: Whether to request 4K version Returns: Dict containing the created request information Raises: Exception: If the request creation fails """ data = { 'mediaType': media_type, 'mediaId': media_id, 'is4k': is_4k } if media_type == 'tv' and seasons: data['seasons'] = seasons return await self._request('POST', '/request', data=data) async def get_request(self, request_id: int) -> Dict: """Get details of a specific request""" return await self._request('GET', f'/request/{request_id}') # Methods for approving and declining requests have been removed # These actions should be performed in the Jellyseerr web interface # Discover endpoints async def discover_movies(self, page: int = 1, language: str = 'en', genre: Optional[str] = None, sort_by: str = 'popularity.desc') -> Dict: """Discover movies with various filters""" params = { 'page': page, 'language': language, 'sortBy': sort_by } if genre: params['genre'] = genre return await self._request('GET', '/discover/movies', params=params) async def discover_tv(self, page: int = 1, language: str = 'en', genre: Optional[str] = None, sort_by: str = 'popularity.desc') -> Dict: """Discover TV shows with various filters""" params = { 'page': page, 'language': language, 'sortBy': sort_by } if genre: params['genre'] = genre return await self._request('GET', '/discover/tv', params=params) async def discover_trending(self, page: int = 1, language: str = 'en') -> Dict: """Get trending movies and TV shows""" params = { 'page': page, 'language': language } return await self._request('GET', '/discover/trending', params=params) # Genre endpoints async def get_movie_genres(self, language: str = 'en') -> List: """Get a list of movie genres""" params = {'language': language} return await self._request('GET', '/genres/movie', params=params) async def get_tv_genres(self, language: str = 'en') -> List: """Get a list of TV show genres""" params = {'language': language} return await self._request('GET', '/genres/tv', params=params) async def get_media(self, filter_status: str = 'all', page: int = 1, page_size: int = 10) -> Dict: """Get all media with pagination""" params = { 'filter': filter_status, 'take': page_size, 'skip': (page - 1) * page_size } return await self._request('GET', '/media', params=params) # Create an instance for import (but don't initialize cookie_jar yet) jellyseerr_api = JellyseerrAPI(config.JELLYSEERR_URL, config.JELLYSEERR_EMAIL, config.JELLYSEERR_PASSWORD)