mirror of
https://github.com/myronblair/kino-app
synced 2026-07-27 21:18:43 -05:00
Myron Blair
3d2b7a4122
Admin can now Approve a pending request as Movie or TV, which looks up the title via Radarr/Sonarr's own lookup API, adds the best match with search-on-add enabled, and tracks the resulting radarr_id/sonarr_id on the request. A live status line (searching/queued/downloading/downloaded, polled every 15s) shows next to each approved request.
99 lines
4.0 KiB
Python
99 lines
4.0 KiB
Python
"""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
|