backend/sonarr_client.py
Rama master · réplica de lectura del repositorio original.
"""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.