412 lines
18 KiB
Python
412 lines
18 KiB
Python
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
|
|
|
|
|
|
_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 queued it for processing."
|
|
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, 'available quality profile')}."
|
|
if "/rootfolder" in normalized_path and normalized_method == "GET":
|
|
count = len(_result_items(result))
|
|
return f"{service} returned {_count_message(count, 'configured library location')}."
|
|
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 "Prowlarr reports that all configured indexers are healthy."
|
|
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 release')}."
|
|
if results
|
|
else "Prowlarr did not find any possible releases."
|
|
)
|
|
|
|
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} completed the check successfully."
|
|
if normalized_method == "POST":
|
|
return f"{service} accepted the request and started processing it."
|
|
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:
|
|
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 Prowlarr indexer health…", "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)
|
|
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,
|
|
)
|
|
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
|
|
|
|
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)
|