kino-app/backend/sonarr.py

"""Sonarr v3 API client. https://sonarr.tv/docs/api/"""
import requests
from typing import List, Dict


def _h(api_key: str) -> dict:
    return {"X-Api-Key": api_key, "Content-Type": "application/json"}


def _poster(images):
    for img in images or []:
        if img.get("coverType") == "poster":
            return img.get("remoteUrl") or img.get("url") or ""
    return ""


def search_series(base_url: str, api_key: str, term: str) -> List[Dict]:
    """Sonarr's own title lookup (proxies TVDB) — used to resolve a free-text request to a real series."""
    r = requests.get(f"{base_url.rstrip('/')}/api/v3/series/lookup", params={"term": term}, headers=_h(api_key), timeout=20)
    r.raise_for_status()
    out = []
    for s in r.json() or []:
        out.append({
            "tvdb_id": s.get("tvdbId"), "tmdb_id": s.get("tmdbId"), "title": s.get("title") or "",
            "year": s.get("year"), "overview": s.get("overview") or "", "poster_url": _poster(s.get("images")),
        })
    return out


def add_series(base_url: str, api_key: str, tvdb_id: int, title: str, year) -> Dict:
    """Adds a series by tvdb_id with monitoring + an immediate missing-episode search enabled."""
    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("Sonarr has no quality profile or root folder configured")
    payload = {
        "title": title, "tvdbId": tvdb_id, "year": year or 0,
        "qualityProfileId": profiles[0]["id"], "rootFolderPath": roots[0]["path"],
        "seasonFolder": True, "monitored": True,
        "addOptions": {"searchForMissingEpisodes": True, "monitor": "all"},
    }
    r = requests.post(f"{base_url}/api/v3/series", json=payload, headers=h, timeout=20)
    r.raise_for_status()
    added = r.json()
    return {"sonarr_id": added["id"], "title": added.get("title"), "year": added.get("year")}


def series_status(base_url: str, api_key: str, sonarr_id: int) -> Dict:
    base_url = base_url.rstrip("/")
    h = _h(api_key)
    s = requests.get(f"{base_url}/api/v3/series/{sonarr_id}", headers=h, timeout=15).json()
    stats = s.get("statistics") or {}
    if stats.get("episodeFileCount", 0) > 0 and stats.get("episodeFileCount") >= stats.get("episodeCount", 1):
        return {"state": "downloaded"}
    q = requests.get(f"{base_url}/api/v3/queue", params={"seriesId": sonarr_id, "includeSeries": False}, 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}
    if stats.get("episodeFileCount", 0) > 0:
        return {"state": "partially downloaded"}
    return {"state": "searching"}


def test_connection(base_url: str, api_key: str) -> bool:
    try:
        r = requests.get(f"{base_url.rstrip('/')}/api/v3/system/status", headers=_h(api_key), timeout=10)
        return r.ok
    except Exception:
        return False


def list_series(base_url: str, api_key: str) -> List[Dict]:
    r = requests.get(f"{base_url.rstrip('/')}/api/v3/series", headers=_h(api_key), timeout=30)
    r.raise_for_status()
    out = []
    for s in r.json() or []:
        stats = s.get("statistics") or {}
        out.append({
            "sonarr_id": s["id"],
            "title": s.get("title", ""),
            "year": s.get("year"),
            "overview": s.get("overview") or "",
            "poster_url": _poster(s.get("images")),
            "tvdb_id": s.get("tvdbId"),
            "tmdb_id": s.get("tmdbId"),
            "season_count": stats.get("seasonCount", 0),
            "episode_file_count": stats.get("episodeFileCount", 0),
        })
    return out


def list_episodes(base_url: str, api_key: str, series_id: int) -> List[Dict]:
    r = requests.get(
        f"{base_url.rstrip('/')}/api/v3/episode",
        params={"seriesId": series_id, "includeEpisodeFile": True},
        headers=_h(api_key), timeout=30,
    )
    r.raise_for_status()
    out = []
    for e in r.json() or []:
        ef = e.get("episodeFile") or {}
        out.append({
            "sonarr_episode_id": e["id"],
            "season_number": e.get("seasonNumber", 0),
            "episode_number": e.get("episodeNumber", 0),
            "title": e.get("title", ""),
            "description": e.get("overview") or "",
            "air_date": (e.get("airDate") or ""),
            "duration_minutes": int((ef.get("runTime") or 0) / 60) if ef.get("runTime") else 0,
            "has_file": bool(e.get("hasFile")),
            "file_path": ef.get("path") if ef else None,
        })
    return out