from typing import Any, Dict, Optional import logging import time import httpx from ..logging_config import sanitize_headers, sanitize_value from ..services.operation_progress import finish_remote_call, start_remote_call from ..metrics import record_remote _SERVICE_NAMES = { "JellyseerrClient": "Seerr", "SonarrClient": "Sonarr", "RadarrClient": "Radarr", "BazarrClient": "Bazarr", "ProwlarrClient": "Prowlarr", "JellyfinClient": "Jellyfin", "QBittorrentClient": "qBittorrent", } def _result_items(result: Any, *keys: str) -> list[Any]: if isinstance(result, list): return result if not isinstance(result, dict): return [] for key in keys: value = result.get(key) if isinstance(value, list): return value return [] def _result_title(result: Any, payload: Optional[Dict[str, Any]] = None) -> Optional[str]: candidates = result if isinstance(result, list) else [result] for candidate in candidates: if not isinstance(candidate, dict): continue title = str(candidate.get("title") or candidate.get("name") or "").strip() if title: return title if isinstance(payload, dict): title = str(payload.get("title") or payload.get("name") or "").strip() if title: return title return None def _count_message(count: int, singular: str, plural: Optional[str] = None) -> str: noun = singular if count == 1 else (plural or f"{singular}s") return f"{count} {noun}" def _queue_result_message(service: str, result: Any) -> str: records = _result_items(result, "records", "items") total = result.get("totalRecords") if isinstance(result, dict) else None count = int(total) if isinstance(total, int) else len(records) if count == 0: return f"{service} has no matching downloads in its queue." first = next((item for item in records if isinstance(item, dict)), None) progress_text = "" if first: size = first.get("size") size_left = first.get("sizeleft") if isinstance(size, (int, float)) and size > 0 and isinstance(size_left, (int, float)): progress = max(0, min(100, round((1 - (size_left / size)) * 100))) progress_text = f" The first is {progress}% complete." return f"{service} found {_count_message(count, 'matching download')} in its queue.{progress_text}" def _command_name(payload: Optional[Dict[str, Any]]) -> str: raw_name = str((payload or {}).get("name") or "").strip() names = { "MoviesSearch": "movie search", "SeriesSearch": "series search", "EpisodeSearch": "episode search", "DownloadRelease": "release download", "RefreshMovie": "movie refresh", "RescanMovie": "movie rescan", "RefreshSeries": "series refresh", "RescanSeries": "series rescan", } return names.get(raw_name, "command") def _operation_result_message( service: str, method: str, path: str, result: Any, *, params: Optional[Dict[str, Any]] = None, payload: Optional[Dict[str, Any]] = None, ) -> str: normalized_path = path.lower().split("?", 1)[0].rstrip("/") normalized_method = method.upper() title = _result_title(result, payload) title_text = f' "{title}"' if title else "" if service == "Seerr": if normalized_path.endswith("/request") and normalized_method == "POST": request_id = result.get("id") if isinstance(result, dict) else None suffix = f" #{request_id}" if isinstance(request_id, int) else "" return f"Seerr created the request{suffix} and passed it into the collection workflow." if "/request/" in normalized_path and normalized_method == "GET": status_names = {1: "waiting for approval", 2: "approved", 3: "declined"} status = result.get("status") if isinstance(result, dict) else None status_text = status_names.get(status) return ( f"Seerr found the request; it is currently {status_text}." if status_text else "Seerr found the request and returned its current status." ) if service in {"Radarr", "Sonarr"}: media_name = "movie" if service == "Radarr" else "series" media_path = "/movie" if service == "Radarr" else "/series" if "/queue" in normalized_path and normalized_method == "GET": return _queue_result_message(service, result) if "/command" in normalized_path and normalized_method == "POST": return f"{service} accepted the {_command_name(payload)} and put it in line to run. This does not mean a download has started." if "/release" in normalized_path: if normalized_method == "GET": count = len(_result_items(result, "records", "items")) return ( f"{service} found {_count_message(count, 'download option')}." if count else f"{service} could not find a suitable download option." ) return f"{service} accepted the selected release and sent it to the download client." if "/qualityprofile" in normalized_path and normalized_method == "GET": count = len(_result_items(result)) return f"{service} returned {_count_message(count, 'download quality setting')}." if "/rootfolder" in normalized_path and normalized_method == "GET": count = len(_result_items(result)) return f"{service} returned {_count_message(count, 'library folder')}." if "/indexer" in normalized_path and normalized_method == "GET": count = len(_result_items(result)) return f"{service} reports {_count_message(count, 'configured search source')}." if service == "Sonarr" and "/episodefile" in normalized_path: if normalized_method == "DELETE": return "Sonarr removed the existing episode file so it can be replaced." count = len(_result_items(result)) return f"Sonarr found {_count_message(count, 'downloaded episode file')}." if service == "Sonarr" and normalized_path.endswith("/episode/monitor") and normalized_method == "PUT": return "Sonarr marked the selected episodes as wanted." if service == "Sonarr" and "/episode" in normalized_path and normalized_method == "GET": episodes = _result_items(result) available = sum(1 for item in episodes if isinstance(item, dict) and item.get("hasFile") is True) return f"Sonarr reports {available} of {len(episodes)} episodes downloaded." if service == "Radarr" and "/moviefile/" in normalized_path and normalized_method == "DELETE": return "Radarr removed the existing movie file so it can be replaced." is_media_endpoint = normalized_path.endswith(media_path) or f"{media_path}/" in normalized_path if is_media_endpoint: if normalized_method == "GET": found = bool(result) if not isinstance(result, list) else len(result) > 0 return ( f"{service} found{title_text} in its library list." if found else f"This {media_name} is not currently in {service}." ) if normalized_method == "POST": search_key = "searchForMovie" if service == "Radarr" else "searchForMissingEpisodes" search_requested = bool(((payload or {}).get("addOptions") or {}).get(search_key)) search_text = " and started looking for a download" if search_requested else "" subject = title_text or f" the {media_name}" return f"{service} added{subject}{search_text}." if normalized_method == "PUT": return f"{service} saved the updated settings for{title_text or f' the {media_name}'}." if "/system/status" in normalized_path: version = str(result.get("version") or "").strip() if isinstance(result, dict) else "" return f"Connected to {service}{f' version {version}' if version else ''}." if service == "Prowlarr": if "/health" in normalized_path: issues = _result_items(result) if not issues: return "The download search sources are working normally." first = next((item for item in issues if isinstance(item, dict)), {}) detail = str(first.get("message") or first.get("source") or "").strip() suffix = f" First issue: {detail}" if detail else "" return f"Prowlarr reports {_count_message(len(issues), 'indexer issue')}.{suffix}" if "/search" in normalized_path: results = _result_items(result, "results", "records") return ( f"Prowlarr found {_count_message(len(results), 'possible download')}." if results else "Prowlarr did not find any possible downloads." ) if service == "Bazarr" and normalized_method == "PATCH" and "/subtitles" in normalized_path: target = "movie" if "/movies/" in normalized_path else "selected episode" language = str((params or {}).get("language") or "the requested language").upper() return f"Bazarr accepted a fresh {language} subtitle search for the {target}." if normalized_method == "GET": return f"{service} finished this check without reporting a problem." if normalized_method == "POST": return f"{service} received the request. Its result will be checked separately." if normalized_method == "PUT": return f"{service} saved the requested changes." if normalized_method == "DELETE": return f"{service} confirmed the item was removed." return f"{service} completed the request successfully." def _operation_error_message(service: str, status_code: Optional[int]) -> str: explanations = { 400: "rejected the request because some details were invalid", 401: "rejected Magent's login details", 403: "refused permission for this action", 404: "could not find the requested item", 409: "reported a conflict, usually because the item already exists", 422: "could not use the details Magent supplied", 429: "is busy and asked Magent to try again later", 500: "encountered an internal error while processing the request", 502: "could not reach one of its own dependent services", 503: "is temporarily unavailable", 504: "did not finish before the request timed out", } explanation = explanations.get(status_code) if explanation: return f"{service} {explanation}." if status_code: return f"{service} could not complete the request (response code {status_code})." return f"Magent could not get a usable response from {service}." def _operation_messages(service: str, method: str, path: str) -> tuple[str, str]: normalized_path = path.lower() normalized_method = method.upper() if service == "Seerr" and "/request/" in normalized_path and normalized_method == "GET": return "Reading the request from Seerr…", "Seerr returned the current request record" if service == "Radarr" and normalized_path.endswith("/movie") and normalized_method == "GET": return "Checking Radarr for the movie…", "Radarr returned the movie record" if service == "Sonarr" and normalized_path.endswith("/series") and normalized_method == "GET": return "Checking Sonarr for the series…", "Sonarr returned the series record" if service in {"Radarr", "Sonarr"} and "/queue" in normalized_path: return f"Checking {service}'s download queue…", f"{service} returned its queue state" if service == "Sonarr" and "/episode" in normalized_path: return "Checking episode availability in Sonarr…", "Sonarr returned episode availability" if service in {"Radarr", "Sonarr"} and "/release" in normalized_path: return f"Checking releases through {service}…", f"{service} returned release information" if service in {"Radarr", "Sonarr"} and "/command" in normalized_path: if normalized_method == "GET": return f"Checking {service}'s search activity…", f"{service} returned its current activity" return f"Sending a command to {service}…", f"{service} accepted the command" if service == "Bazarr" and "/subtitles" in normalized_path and normalized_method == "PATCH": return "Asking Bazarr for fresh subtitles…", "Bazarr started the subtitle search" if service == "Prowlarr" and "/health" in normalized_path: return "Checking whether the download search sources are working…", "Prowlarr returned its indexer health" return f"Contacting {service}…", f"{service} responded" class ApiClient: def __init__(self, base_url: Optional[str], api_key: Optional[str] = None): self.base_url = base_url.rstrip("/") if base_url else None self.api_key = api_key self.logger = logging.getLogger(f"{__name__}.{self.__class__.__name__}") def configured(self) -> bool: return bool(self.base_url) def headers(self) -> Dict[str, str]: return {"X-Api-Key": self.api_key} if self.api_key else {} def _response_summary(self, response: Optional[httpx.Response]) -> Optional[Any]: if response is None: return None try: payload = sanitize_value(response.json()) except ValueError: payload = sanitize_value(response.text) if isinstance(payload, str) and len(payload) > 500: return f"{payload[:500]}..." return payload async def _send_request( self, client: httpx.AsyncClient, method: str, url: str, *, headers: Dict[str, str], params: Optional[Dict[str, Any]], payload: Optional[Dict[str, Any]], ) -> httpx.Response: return await client.request( method, url, headers=headers, params=params, json=payload, ) async def _request( self, method: str, path: str, *, params: Optional[Dict[str, Any]] = None, payload: Optional[Dict[str, Any]] = None, timeout_seconds: float = 10.0, ) -> Optional[Any]: if not self.base_url: self.logger.warning("client request skipped method=%s path=%s reason=not-configured", method, path) return None url = f"{self.base_url}{path}" started_at = time.perf_counter() service_name = _SERVICE_NAMES.get(self.__class__.__name__, self.__class__.__name__.removesuffix("Client")) active_message, _ = _operation_messages(service_name, method, path) operation_event_id = start_remote_call(service_name, active_message) metric_status = 'error' self.logger.debug( "outbound request started method=%s url=%s params=%s payload=%s headers=%s", method, url, sanitize_value(params), sanitize_value(payload), sanitize_headers(self.headers()), ) try: async with httpx.AsyncClient(timeout=timeout_seconds) as client: response = await self._send_request( client, method, url, headers=self.headers(), params=params, payload=payload, ) metric_status = str(response.status_code) response.raise_for_status() duration_ms = round((time.perf_counter() - started_at) * 1000, 2) self.logger.debug( "outbound request completed method=%s url=%s status=%s duration_ms=%s", method, url, response.status_code, duration_ms, ) result = response.json() if response.content else None finish_remote_call( operation_event_id, success=True, status_code=response.status_code, message=_operation_result_message( service_name, method, path, result, params=params, payload=payload, ), ) return result except httpx.HTTPStatusError as exc: duration_ms = round((time.perf_counter() - started_at) * 1000, 2) response = exc.response status = response.status_code if response is not None else "unknown" log_fn = self.logger.error if isinstance(status, int) and status >= 500 else self.logger.warning log_fn( "outbound request returned error method=%s url=%s status=%s duration_ms=%s response=%s", method, url, status, duration_ms, self._response_summary(response), ) finish_remote_call( operation_event_id, success=False, status_code=status if isinstance(status, int) else None, message=_operation_error_message( service_name, status if isinstance(status, int) else None, ), ) raise except Exception: duration_ms = round((time.perf_counter() - started_at) * 1000, 2) self.logger.exception( "outbound request failed method=%s url=%s duration_ms=%s", method, url, duration_ms, ) finish_remote_call( operation_event_id, success=False, message=_operation_error_message(service_name, None), ) raise finally: record_remote(service_name, method, metric_status, time.perf_counter() - started_at) async def get( self, path: str, params: Optional[Dict[str, Any]] = None, timeout_seconds: float = 10.0, ) -> Optional[Any]: return await self._request( "GET", path, params=params, timeout_seconds=timeout_seconds ) async def post(self, path: str, payload: Optional[Dict[str, Any]] = None) -> Optional[Any]: return await self._request("POST", path, payload=payload) async def put(self, path: str, payload: Optional[Dict[str, Any]] = None) -> Optional[Any]: return await self._request("PUT", path, payload=payload) async def delete( self, path: str, params: Optional[Dict[str, Any]] = None, ) -> Optional[Any]: return await self._request("DELETE", path, params=params)