"""Sonarr v3 API client for series/episode management. Provides series listing, episode enumeration, file paths, tag filtering, and RescanSeries commands after new subtitle files are created. """ import logging import time import requests from config import get_settings from mediainfo_utils import arr_video_fps logger = logging.getLogger(__name__) REQUEST_TIMEOUT = 15 MAX_RETRIES = 3 BACKOFF_BASE = 2 _client = None _clients_cache = {} # Cache for multi-instance clients: {instance_name: SonarrClient} def get_sonarr_client(instance_name=None): """Get or create a Sonarr client. Returns None if not configured. Args: instance_name: Optional instance name. If None, uses first available instance or legacy config. Returns: SonarrClient instance or None """ global _client, _clients_cache # If instance_name is specified, use multi-instance logic if instance_name is not None: if instance_name in _clients_cache: return _clients_cache[instance_name] from config import get_sonarr_instances instances = get_sonarr_instances() for inst in instances: if inst.get("name") == instance_name: client = SonarrClient(inst["url"], inst["api_key"]) _clients_cache[instance_name] = client return client # Name not found — fall back to first available instance if instances: logger.warning( "Sonarr instance '%s' not found; falling back to first instance '%s'", instance_name, instances[0].get("name"), ) inst = instances[0] client = SonarrClient(inst["url"], inst["api_key"]) _clients_cache[instance_name] = client return client logger.warning("Sonarr instance '%s' not found and no instances configured", instance_name) return None # Legacy singleton behavior (for backward compatibility) if _client is not None: return _client from config import get_sonarr_instances instances = get_sonarr_instances() if instances: # Use first instance inst = instances[0] _client = SonarrClient(inst["url"], inst["api_key"]) return _client # Fallback to legacy config settings = get_settings() if not settings.sonarr_url or not settings.sonarr_api_key: logger.debug("Sonarr not configured (sonarr_url or sonarr_api_key missing)") return None _client = SonarrClient(settings.sonarr_url, settings.sonarr_api_key) return _client def invalidate_client(): """Reset all client caches so the next call creates fresh instances.""" global _client, _clients_cache _client = None _clients_cache = {} from sonarr_diagnostics import _SonarrDiagnosticsMixin # noqa: E402 def _season_size(series: dict, season: int | None) -> int | None: """Episodes Sonarr lists for ``season`` (its statistics), or None.""" for entry in series.get("seasons") or []: if entry.get("seasonNumber") == season: stats = entry.get("statistics") or {} return stats.get("totalEpisodeCount") or stats.get("episodeCount") or None return None def series_aliases(series: dict) -> list[tuple[str, int | None, int | None]]: """A series' alternate titles as (title, seasonNumber, sceneSeasonNumber). The episode gate reads a season-scoped title ("Strike Witches - Road to Berlin", scene season 3) as that season; -1 means every season. Library and scene season can differ ("Criminal Minds Evolution": 16 and 1). """ return [ (alt["title"], alt.get("seasonNumber"), alt.get("sceneSeasonNumber")) for alt in series.get("alternateTitles") or [] if alt.get("title") ] class SonarrClient(_SonarrDiagnosticsMixin): """Sonarr v3 REST API Client.""" def __init__(self, url, api_key): self.url = url.rstrip("/") self.api_key = api_key self.session = requests.Session() self.session.headers["X-Api-Key"] = api_key def _get(self, path, params=None, timeout=REQUEST_TIMEOUT): """GET request with retry logic.""" url = f"{self.url}/api/v3{path}" for attempt in range(1, MAX_RETRIES + 1): try: resp = self.session.get(url, params=params, timeout=timeout) # Handle rate limiting (429) before raise_for_status if resp.status_code == 429: retry_after = resp.headers.get("Retry-After") if retry_after: try: wait_seconds = int(retry_after) except ValueError: wait_seconds = 60 else: wait_seconds = 60 logger.warning( "Sonarr GET %s rate limited (attempt %d), waiting %ds", path, attempt, wait_seconds, ) if attempt < MAX_RETRIES: time.sleep(wait_seconds) continue else: logger.error( "Sonarr GET %s rate limited after %d attempts", path, MAX_RETRIES ) return None resp.raise_for_status() return resp.json() except requests.ConnectionError as e: logger.warning("Sonarr GET %s failed (attempt %d): %s", path, attempt, e) except requests.Timeout: logger.warning("Sonarr GET %s timed out (attempt %d)", path, attempt) except requests.HTTPError as e: logger.error("Sonarr HTTP error on GET %s: %s", path, e) return None except Exception as e: logger.error("Sonarr unexpected error on GET %s: %s", path, e) return None if attempt < MAX_RETRIES: time.sleep(BACKOFF_BASE * attempt) logger.error("Sonarr GET %s failed after %d attempts", path, MAX_RETRIES) return None def _post(self, path, data=None, timeout=REQUEST_TIMEOUT): """POST request with retry logic.""" url = f"{self.url}/api/v3{path}" for attempt in range(1, MAX_RETRIES + 1): try: resp = self.session.post(url, json=data, timeout=timeout) # Handle rate limiting (429) before raise_for_status if resp.status_code == 429: retry_after = resp.headers.get("Retry-After") if retry_after: try: wait_seconds = int(retry_after) except ValueError: wait_seconds = 60 else: wait_seconds = 60 logger.warning( "Sonarr POST %s rate limited (attempt %d), waiting %ds", path, attempt, wait_seconds, ) if attempt < MAX_RETRIES: time.sleep(wait_seconds) continue else: logger.error( "Sonarr POST %s rate limited after %d attempts", path, MAX_RETRIES ) return None resp.raise_for_status() return resp.json() if resp.content else {} except requests.ConnectionError as e: logger.warning("Sonarr POST %s failed (attempt %d): %s", path, attempt, e) except requests.Timeout: logger.warning("Sonarr POST %s timed out (attempt %d)", path, attempt) except Exception as e: logger.error("Sonarr unexpected error on POST %s: %s", path, e) return None if attempt < MAX_RETRIES: time.sleep(BACKOFF_BASE * attempt) return None def health_check(self): """Check if Sonarr is reachable. Returns: tuple: (is_healthy: bool, message: str) """ result = self._get("/system/status") if result is not None: return True, "OK" return False, f"Cannot connect to Sonarr at {self.url}" def test_connection(self) -> dict: """Test Sonarr connection and return a result dict.""" healthy, message = self.health_check() return {"success": healthy, "message": message} def get_series(self): """Get all series. Returns: list: List of series dicts, or empty list on error """ result = self._get("/series") return result or [] def list_series(self): """All series, or None when Sonarr did not answer. ``get_series`` answers [] for both an outage and an empty library; callers that must tell them apart use this. """ result = self._get("/series") return result if isinstance(result, list) else None def get_series_by_id(self, series_id): """Get a single series by ID.""" return self._get(f"/series/{series_id}") def lookup_series(self, series_id): """One GET /series/{id} without retries: ``(status, body)``. Unlike ``get_series_by_id`` (None for every failure) this keeps a 404 — Sonarr does not know the series — apart from an outage. ``status`` is None when Sonarr did not answer (connection error, timeout, bad JSON). """ try: resp = self.session.get( f"{self.url}/api/v3/series/{series_id}", timeout=REQUEST_TIMEOUT ) except requests.RequestException as e: logger.warning("Sonarr lookup of series %s failed: %s", series_id, e) return None, None if resp.status_code != 200: return resp.status_code, None try: return 200, resp.json() except ValueError: logger.warning("Sonarr lookup of series %s: response is not JSON", series_id) return None, None def get_episodes(self, series_id): """Get all episodes for a series. Returns: list: List of episode dicts """ result = self._get("/episode", params={"seriesId": series_id}) return result or [] def list_episodes(self, series_id): """All episodes of a series, or None when Sonarr did not answer.""" result = self._get("/episode", params={"seriesId": series_id}) return result if isinstance(result, list) else None def get_episode_by_id(self, episode_id): """Get a single episode by ID. Returns: dict: Episode info including episodeFileId and hasFile """ return self._get(f"/episode/{episode_id}") def get_episode_file_path(self, episode_id): """Get the file path for a specific episode by episode ID. Looks up the episode, then fetches the episode file details. Returns: str or None: Full file path, or None if not found """ episode = self.get_episode_by_id(episode_id) if not episode: return None # Try direct episodeFile.path first (Sonarr v3 often includes it) ep_file = episode.get("episodeFile") if ep_file and ep_file.get("path"): return ep_file["path"] # Fallback: get episodeFile separately file_id = episode.get("episodeFileId") if not file_id or file_id == 0: logger.debug("Episode %d has no file", episode_id) return None file_info = self.get_episode_file(file_id) if file_info and file_info.get("path"): return file_info["path"] return None def get_episode_file(self, episode_file_id): """Get file info for an episode file. Returns: dict: Episode file info including path """ return self._get(f"/episodefile/{episode_file_id}") def get_episode_files_by_series(self, series_id): """Get all episode files for a series in one request. Uses /episodefile?seriesId=X which returns full file info including path and mediaInfo. Much more efficient than fetching per-episode. Returns: dict: Mapping of episodeFileId -> file info dict (with path, mediaInfo) """ result = self._get("/episodefile", params={"seriesId": series_id}) files = result or [] return {f["id"]: f for f in files if f.get("id")} def get_tags(self): """Get all tags. Returns: list: List of tag dicts (id, label) """ result = self._get("/tag") return result or [] def get_anime_series(self, anime_tag="anime"): """Get series filtered by anime tag, seriesType=anime, or 'Anime' genre. Returns: list: List of anime series dicts """ tags = self.get_tags() anime_tag_ids = set( t["id"] for t in tags if t.get("label", "").lower() == anime_tag.lower() ) all_series = self.get_series() anime_series = [] for series in all_series: has_tag = anime_tag_ids and any( tag_id in anime_tag_ids for tag_id in series.get("tags", []) ) is_anime_type = series.get("seriesType", "").lower() == "anime" has_anime_genre = "anime" in [g.lower() for g in series.get("genres", [])] if has_tag or is_anime_type or has_anime_genre: anime_series.append(series) logger.info("Found %d anime series (tag + seriesType + genre)", len(anime_series)) return anime_series def rescan_series(self, series_id): """Trigger a RescanSeries command. Returns: bool: True if command was sent successfully """ result = self._post( "/command", data={ "name": "RescanSeries", "seriesId": series_id, }, ) if result is not None: logger.info("Triggered Sonarr RescanSeries for series %d", series_id) return True return False def get_episode_metadata(self, series_id, episode_id): """Get rich metadata for building a VideoQuery. Returns: dict: {series_title, season, episode, absolute_episode, year, imdb_id, tvdb_id, anidb_id, anilist_id, title, fps} or None on error """ series = self.get_series_by_id(series_id) if not series: return None episode = self.get_episode_by_id(episode_id) if not episode: return None # Extract AniDB ID using mapper service anidb_id = None tvdb_id = series.get("tvdbId") series_title = series.get("title", "") try: from anidb_mapper import get_anidb_id anidb_id = get_anidb_id(tvdb_id=tvdb_id, series_title=series_title, series=series) except Exception as e: logger.debug("Failed to resolve AniDB ID: %s", e) # Extract AniList ID from Custom Fields (if available) anilist_id = None custom_fields = series.get("customFields", {}) if custom_fields: # Try various field name variations for field_name in ["anilist_id", "anilistId", "AniList", "AniList ID", "anilist"]: value = custom_fields.get(field_name) if value: try: anilist_id = int(value) if isinstance(value, (int, str)) else None if anilist_id and anilist_id > 0: break except (ValueError, TypeError): continue return { "series_title": series_title, "year": series.get("year"), "season": episode.get("seasonNumber"), "episode": episode.get("episodeNumber"), # Sonarr numbers every series continuously, anime or not. "absolute_episode": episode.get("absoluteEpisodeNumber"), "title": episode.get("title", ""), "series_aliases": series_aliases(series), "scene_season": episode.get("sceneSeasonNumber"), "season_size": _season_size(series, episode.get("seasonNumber")), "imdb_id": series.get("imdbId", ""), "tvdb_id": tvdb_id, "anidb_id": anidb_id, "anilist_id": anilist_id, # GET /episode/{id} embeds the episode file and its mediaInfo. "fps": arr_video_fps(episode.get("episodeFile")), } def get_episode_history(self, episode_id): """History records (grabs, imports, ...) of one episode, as a list. GET /history?episodeId=X answers a page ``{"records": [...]}``; one page of 50 covers an episode's grabs and imports in practice. """ result = self._get("/history", params={"episodeId": episode_id, "pageSize": 50}) records = result.get("records") if isinstance(result, dict) else None return records if isinstance(records, list) else [] # extended_health_check + get_library_info live on # _SonarrDiagnosticsMixin — see sonarr_diagnostics.py.