127 lines
5.4 KiB
Python
127 lines
5.4 KiB
Python
from typing import Any, Dict, Optional
|
|
from .base import ApiClient
|
|
|
|
|
|
class SonarrClient(ApiClient):
|
|
async def get_system_status(self) -> Optional[Dict[str, Any]]:
|
|
return await self.get("/api/v3/system/status")
|
|
|
|
async def get_series_by_tvdb_id(self, tvdb_id: int) -> Optional[Dict[str, Any]]:
|
|
return await self.get("/api/v3/series", params={"tvdbId": tvdb_id})
|
|
|
|
async def lookup_series_by_tvdb_id(self, tvdb_id: int) -> Optional[Dict[str, Any]]:
|
|
result = await self.get("/api/v3/series/lookup", params={"term": f"tvdb:{tvdb_id}"})
|
|
if not isinstance(result, list):
|
|
return None
|
|
for item in result:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
try:
|
|
if int(item.get("tvdbId")) == tvdb_id:
|
|
return item
|
|
except (TypeError, ValueError):
|
|
continue
|
|
return next((item for item in result if isinstance(item, dict)), None)
|
|
|
|
async def get_series(self, series_id: int) -> Optional[Dict[str, Any]]:
|
|
return await self.get(f"/api/v3/series/{series_id}")
|
|
|
|
async def get_root_folders(self) -> Optional[Dict[str, Any]]:
|
|
return await self.get("/api/v3/rootfolder")
|
|
|
|
async def get_quality_profiles(self) -> Optional[Dict[str, Any]]:
|
|
return await self.get("/api/v3/qualityprofile")
|
|
|
|
async def get_queue(self, series_id: int) -> Optional[Dict[str, Any]]:
|
|
records = []
|
|
page = 1
|
|
while True:
|
|
result = await self.get("/api/v3/queue", params={
|
|
"seriesIds": series_id, "includeEpisode": "true",
|
|
"page": page, "pageSize": 100,
|
|
})
|
|
if not isinstance(result, dict) or not isinstance(result.get("records"), list):
|
|
raise ValueError("Sonarr returned an invalid queue")
|
|
batch = result["records"]
|
|
records.extend(batch)
|
|
if not batch or len(records) >= int(result.get("totalRecords", len(records))):
|
|
return {**result, "records": records, "totalRecords": len(records)}
|
|
page += 1
|
|
if page > 100:
|
|
raise ValueError("Sonarr queue exceeded the safe paging limit")
|
|
|
|
async def get_indexers(self) -> Optional[Dict[str, Any]]:
|
|
return await self.get("/api/v3/indexer")
|
|
|
|
async def get_episodes(self, series_id: int) -> Optional[Dict[str, Any]]:
|
|
return await self.get("/api/v3/episode", params={"seriesId": series_id})
|
|
|
|
async def get_episode_files(self, series_id: int) -> Optional[Dict[str, Any]]:
|
|
return await self.get("/api/v3/episodefile", params={"seriesId": series_id})
|
|
|
|
async def search_releases(self, series_id: int, season_number: int) -> Optional[Any]:
|
|
return await self.get(
|
|
"/api/v3/release",
|
|
params={"seriesId": series_id, "seasonNumber": season_number},
|
|
timeout_seconds=90.0,
|
|
)
|
|
|
|
async def search(self, series_id: int) -> Optional[Dict[str, Any]]:
|
|
return await self.post("/api/v3/command", payload={"name": "SeriesSearch", "seriesId": series_id})
|
|
|
|
async def search_episodes(self, episode_ids: list[int]) -> Optional[Dict[str, Any]]:
|
|
return await self.post("/api/v3/command", payload={"name": "EpisodeSearch", "episodeIds": episode_ids})
|
|
|
|
async def monitor_episodes(
|
|
self, episode_ids: list[int], monitored: bool = True
|
|
) -> Optional[Dict[str, Any]]:
|
|
return await self.put(
|
|
"/api/v3/episode/monitor",
|
|
payload={"episodeIds": episode_ids, "monitored": monitored},
|
|
)
|
|
|
|
async def delete_episode_file(self, episode_file_id: int) -> Optional[Any]:
|
|
return await self.delete(
|
|
f"/api/v3/episodefile/{episode_file_id}",
|
|
params={"deleteFromClient": "true"},
|
|
)
|
|
|
|
async def add_series(
|
|
self,
|
|
tvdb_id: int,
|
|
quality_profile_id: int,
|
|
root_folder: str,
|
|
monitored: bool = True,
|
|
title: Optional[str] = None,
|
|
search_missing: bool = True,
|
|
) -> Optional[Dict[str, Any]]:
|
|
lookup = await self.lookup_series_by_tvdb_id(tvdb_id)
|
|
resolved_title = str((lookup or {}).get("title") or "").strip() or (title or "").strip()
|
|
if not resolved_title:
|
|
raise ValueError("Sonarr could not resolve a title for this TVDB ID")
|
|
payload = {
|
|
"tvdbId": tvdb_id,
|
|
"title": resolved_title,
|
|
"qualityProfileId": quality_profile_id,
|
|
"rootFolderPath": root_folder,
|
|
"monitored": monitored,
|
|
"seasonFolder": True,
|
|
"addOptions": {"searchForMissingEpisodes": search_missing},
|
|
}
|
|
return await self.post("/api/v3/series", payload=payload)
|
|
|
|
async def update_series(self, payload: Dict[str, Any]) -> Optional[Dict[str, Any]]:
|
|
return await self.put("/api/v3/series", payload=payload)
|
|
|
|
async def grab_release(self, guid: str, indexer_id: int) -> Optional[Dict[str, Any]]:
|
|
return await self.post("/api/v3/release", payload={"guid": guid, "indexerId": indexer_id})
|
|
|
|
async def push_release(self, payload: Dict[str, Any]) -> Optional[Dict[str, Any]]:
|
|
return await self.post("/api/v3/release/push", payload=payload)
|
|
|
|
async def download_release(self, guid: str, indexer_id: int) -> Optional[Dict[str, Any]]:
|
|
return await self.post(
|
|
"/api/v3/command",
|
|
payload={"name": "DownloadRelease", "guid": guid, "indexerId": indexer_id},
|
|
)
|