"""Radarr v3 API client. https://radarr.video/docs/api/""" import requests from typing import List, Dict, Optional def _h(api_key: str) -> dict: return {"X-Api-Key": api_key, "Content-Type": "application/json"} def list_movies(base_url: str, api_key: str) -> List[Dict]: base_url = base_url.rstrip("/") r = requests.get(f"{base_url}/api/v3/movie", headers=_h(api_key), timeout=20) r.raise_for_status() data = r.json() or [] out = [] for m in data: poster = "" for img in m.get("images", []) or []: if img.get("coverType") == "poster": poster = img.get("remoteUrl") or img.get("url") or "" break movie_file = m.get("movieFile") or {} out.append({ "radarr_id": m["id"], "title": m.get("title") or "", "year": m.get("year"), "overview": m.get("overview") or "", "has_file": bool(m.get("hasFile")), "file_path": movie_file.get("path") if movie_file else None, "poster_url": poster, "tmdb_id": m.get("tmdbId"), }) return out def search_movie(base_url: str, api_key: str, term: str) -> List[Dict]: """Radarr's own title lookup (proxies TMDB) — used to resolve a free-text request to a real movie.""" base_url = base_url.rstrip("/") r = requests.get(f"{base_url}/api/v3/movie/lookup", params={"term": term}, headers=_h(api_key), timeout=20) r.raise_for_status() out = [] for m in r.json() or []: poster = "" for img in m.get("images", []) or []: if img.get("coverType") == "poster": poster = img.get("remoteUrl") or img.get("url") or "" break out.append({ "tmdb_id": m.get("tmdbId"), "title": m.get("title") or "", "year": m.get("year"), "overview": m.get("overview") or "", "poster_url": poster, }) return out def add_movie(base_url: str, api_key: str, tmdb_id: int, title: str, year: Optional[int]) -> Dict: """Adds a movie by tmdb_id with monitoring + an immediate search enabled, using the first configured quality profile and root folder — same defaults Radarr's own UI would pre-fill.""" base_url = base_url.rstrip("/") h = _h(api_key) profiles = requests.get(f"{base_url}/api/v3/qualityprofile", headers=h, timeout=15).json() roots = requests.get(f"{base_url}/api/v3/rootfolder", headers=h, timeout=15).json() if not profiles or not roots: raise RuntimeError("Radarr has no quality profile or root folder configured") payload = { "title": title, "tmdbId": tmdb_id, "year": year or 0, "qualityProfileId": profiles[0]["id"], "rootFolderPath": roots[0]["path"], "monitored": True, "addOptions": {"searchForMovie": True}, } r = requests.post(f"{base_url}/api/v3/movie", json=payload, headers=h, timeout=20) r.raise_for_status() added = r.json() return {"radarr_id": added["id"], "title": added.get("title"), "year": added.get("year")} def movie_status(base_url: str, api_key: str, radarr_id: int) -> Dict: base_url = base_url.rstrip("/") h = _h(api_key) m = requests.get(f"{base_url}/api/v3/movie/{radarr_id}", headers=h, timeout=15).json() if m.get("hasFile"): return {"state": "downloaded"} q = requests.get(f"{base_url}/api/v3/queue", params={"movieIds": radarr_id}, headers=h, timeout=15).json() items = q.get("records") if isinstance(q, dict) else q if items: it = items[0] size = it.get("size") or 0 sizeleft = it.get("sizeleft") or 0 pct = round(100 * (1 - (sizeleft / size)), 1) if size else 0 return {"state": (it.get("status") or "queued").lower(), "progress_pct": pct} return {"state": "searching"} def test_connection(base_url: str, api_key: str) -> bool: base_url = base_url.rstrip("/") try: r = requests.get(f"{base_url}/api/v3/system/status", headers=_h(api_key), timeout=10) return r.ok except Exception: return False