#!/usr/bin/env python3 """sonarr-cli — Sonarr TV series management from the terminal.""" import argparse, json, os, sys, warnings from typing import Any, Dict, List, Optional, Tuple warnings.simplefilter("ignore") import requests ENV_SERVER = os.getenv("ARR_SERVER_SONARR", "http://localhost:8989") ENV_KEY = os.getenv("ARR_KEY_SONARR", "") QUIET = False GLOBAL_FLAGS: Dict[str, Any] = {"json": False, "dry_run": False, "quiet": False, "verbose": False} def log(m): global QUIET; G = GLOBAL_FLAGS; (not QUIET and not G.get("json")) and print(m) def warn(m): print(f"Warning: {m}", file=sys.stderr) def die(m, c=1): print(f"Error: {m}", file=sys.stderr); sys.exit(c) def emit(h, d): if GLOBAL_FLAGS.get("json"): print(json.dumps(d, default=str)) else: print(h) def _preparse(argv): BOOLS = {"--json","--dry-run","--force","--quiet","--verbose"} f, fl = {}, [argv[0]] i = 1 while i < len(argv): a = argv[i] if a in BOOLS: f[a.lstrip("-").replace("-","_")] = True; i += 1 elif a in ("--help","-h"): return f, argv elif a == "--": fl.extend(argv[i:]); break else: fl.append(a); i += 1 return f, fl class SonarrClient: def __init__(self, server="", key="", dry_run=False): self.server = (server or ENV_SERVER).rstrip("/"); self.key = key or ENV_KEY; self.dry_run = dry_run def _get(self, path, params=None): url = f"{self.server}/api/v3{path}" if self.dry_run: return {"dry_run":True, "url":url, "params":params} if not self.key: die("ARR_KEY_SONARR not set. Get one from Sonarr → Settings → General → API Key.") try: r = requests.get(url, params=params, headers={"X-Api-Key": self.key, "Accept":"application/json"}, timeout=30) except ConnectionError as e: die(f"Cannot connect: {e}") if r.status_code == 401: die("Auth failed (401). Check ARR_KEY_SONARR.") if r.status_code >= 400: try: d = r.json() except: d = r.text[:200] die(f"API error ({r.status_code}): {d}") return r.json() def _post(self, path, data): url = f"{self.server}/api/v3{path}" if self.dry_run: return {"dry_run":True, "url":url, "json":data} if not self.key: die("ARR_KEY_SONARR not set.") try: r = requests.post(url, json=data, headers={"X-Api-Key":self.key,"Content-Type":"application/json"}, timeout=30) except ConnectionError as e: die(f"Cannot connect: {e}") if r.status_code >= 400: try: d = r.json() except: d = r.text[:200] die(f"API error ({r.status_code}): {d}") return r.json() def status(self): return self._get("/system/status") def series_list(self): return self._get("/series") def series(self, sid): return self._get(f"/series/{sid}") def lookup(self, term): return self._get("/series/lookup", {"term": term}) def episodes(self, series_id): return self._get("/episode", {"seriesId": series_id}) def episode_files(self, series_id): return self._get("/episodefile", {"seriesId": series_id}) def calendar(self, start="", end=""): p = {} if start: p["start"] = start if end: p["end"] = end return self._get("/calendar", p) def queue(self, page=1, limit=20): return self._get("/queue", {"page":page, "pageSize":limit}) def history(self, page=1, limit=20): return self._get("/history", {"page":page, "pageSize":limit}) def wanted_missing(self, page=1, limit=20): return self._get("/wanted/missing", {"page":page, "pageSize":limit}) def add_series(self, tvdb_id, quality_profile=4, root_folder_path="", monitored=True, search=True): return self._post("/series", {"tvdbId":tvdb_id, "qualityProfileId":quality_profile, "rootFolderPath":root_folder_path, "monitored":monitored, "addOptions":{"searchForMissingEpisodes":search}}) def cmd_status(client, args): if client.dry_run: return emit("[dry-run] Get system status", {"dry_run":True}) d = client.status() or {} emit(f"🖥️ Sonarr v{d.get('version','?')} OS: {d.get('osName','?')} DB: {d.get('databaseVersion','?')}", {"version":d.get("version"),"osName":d.get("osName")}) def cmd_series(client, args): p = argparse.ArgumentParser(prog="sonarr-cli series") p.add_argument("--limit", type=int, default=50) p.add_argument("--status", help="Filter by status (continuing, ended, upcoming)") parsed, _ = p.parse_known_args(args) if client.dry_run: return emit("[dry-run] List series", {"dry_run":True}) data = client.series_list() or [] if parsed.status: data = [s for s in data if s.get("status") == parsed.status] data = data[:parsed.limit] if not data: return emit("No series found.", {"series":[]}) lines, out = [], [] for s in data: title = s.get("title","?"); year = s.get("year",""); net = s.get("network","") status = s.get("status","?"); mon = s.get("monitored",False); seasons = len(s.get("seasons",[])) icon = "✅" if s.get("statistics",{}).get("episodeCount",0) > 0 else ("👁️" if mon else "🚫") net_str = f" [{net}]" if net else "" lines.append(f" {icon} {title:40} ({year}){net_str} [{status}] {seasons} seasons") out.append({"title":title,"year":year,"network":net,"status":status,"monitored":mon,"seasons":seasons, "id":s.get("id"),"tvdbId":s.get("tvdbId")}) emit(f"{len(data)} series:\n"+"\n".join(lines), {"series":out}) def cmd_lookup(client, args): p = argparse.ArgumentParser(prog="sonarr-cli lookup") p.add_argument("--term","-t",required=True) parsed, _ = p.parse_known_args(args) if client.dry_run: return emit(f"[dry-run] Lookup: {parsed.term}", {"dry_run":True}) data = client.lookup(parsed.term) or [] if not data: return emit("No results.", {"results":[]}) lines, out = [], [] for s in data[:20]: t = s.get("title","?"); y = s.get("year",""); net = s.get("network",""); st = s.get("status","?") tvdb = s.get("tvdbId","?"); net_str = f" [{net}]" if net else "" lines.append(f" {t:40} ({y}){net_str} [{st}] tvdb={tvdb}") out.append({"title":t,"year":y,"network":net,"status":st,"tvdbId":tvdb,"overview":s.get("overview","")[:120]}) emit(f"{len(data)} result(s):\n"+"\n".join(lines), {"results":out}) def cmd_episodes(client, args): p = argparse.ArgumentParser(prog="sonarr-cli episodes") p.add_argument("--series-id", type=int, required=True) p.add_argument("--limit", type=int, default=50) parsed, _ = p.parse_known_args(args) if client.dry_run: return emit("[dry-run] List episodes", {"dry_run":True}) data = client.episodes(parsed.series_id) or [] data = data[:parsed.limit] if not data: return emit("No episodes.", {"episodes":[]}) title = data[0].get("seriesTitle","?") lines, out = [], [] for e in data: ep = e.get("episodeNumber","?"); sn = e.get("seasonNumber","?") t = e.get("title","?"); had = e.get("hasFile",False); mon = e.get("monitored",False) air = (e.get("airDate") or "?") icon = "✅" if had else ("👁️" if mon else "🚫") lines.append(f" {icon} S{sn:02d}E{ep:02d} {t:45} airs {air}") out.append({"seriesTitle":title,"seasonNumber":sn,"episodeNumber":ep,"title":t,"hasFile":had,"monitored":mon,"airDate":air}) emit(f"{title} — {len(data)} episodes:\n"+"\n".join(lines), {"episodes":out}) def cmd_calendar(client, args): if client.dry_run: return emit("[dry-run] Get calendar", {"dry_run":True}) data = client.calendar() or [] if not data: return emit("No upcoming.", {"episodes":[]}) lines, out = [], [] for e in data[:30]: t = e.get("seriesTitle","?"); sn = e.get("seasonNumber","?"); ep = e.get("episodeNumber","?") air = (e.get("airDate") or "?"); title = e.get("title","") lines.append(f" {air} {t:40} S{sn:02d}E{ep:02d} — {title}") out.append({"seriesTitle":t,"seasonNumber":sn,"episodeNumber":ep,"airDate":air,"title":title}) emit(f"Upcoming:\n"+"\n".join(lines), {"episodes":out}) def cmd_wanted(client, args): p = argparse.ArgumentParser(prog="sonarr-cli wanted") p.add_argument("--limit", type=int, default=20) parsed, _ = p.parse_known_args(args) if client.dry_run: return emit("[dry-run] List wanted", {"dry_run":True}) data = client.wanted_missing(limit=parsed.limit) or {} records = data.get("records",[]) if not records: return emit("No missing episodes.", {"wanted":[]}) lines, out = [], [] for e in records: t = e.get("seriesTitle","?"); sn = e.get("seasonNumber","?"); ep = e.get("episodeNumber","?") title = e.get("title",""); air = (e.get("airDate") or "?") lines.append(f" {t:40} S{sn:02d}E{ep:02d} — {title} airs {air}") out.append({"seriesTitle":t,"seasonNumber":sn,"episodeNumber":ep,"title":title,"airDate":air}) total = data.get("totalRecords", len(records)) emit(f"{total} missing episode(s):\n"+"\n".join(lines), {"total":total,"wanted":out}) def main(): global GLOBAL_FLAGS, QUIET GLOBAL_FLAGS, filtered_argv = _preparse(sys.argv) if GLOBAL_FLAGS.get("quiet"): QUIET = True if GLOBAL_FLAGS.get("json"): warnings.simplefilter("ignore") parser = argparse.ArgumentParser(prog="sonarr-cli", description="Sonarr TV series management.", epilog="Set ARR_SERVER_SONARR and ARR_KEY_SONARR.") sub = parser.add_subparsers(dest="command") sub.add_parser("status") p_series = sub.add_parser("series", help="List series") p_series.add_argument("--limit", type=int, default=50) p_series.add_argument("--status", help="Filter by status") p_lookup = sub.add_parser("lookup", help="Search for series") p_lookup.add_argument("--term","-t",required=True) p_eps = sub.add_parser("episodes", help="List episodes") p_eps.add_argument("--series-id",type=int,required=True) p_eps.add_argument("--limit",type=int,default=50) sub.add_parser("calendar") sub.add_parser("wanted").add_argument("--limit",type=int,default=20) args = parser.parse_args(filtered_argv[1:]) if not args.command: parser.print_help(); sys.exit(1) client = SonarrClient(dry_run=GLOBAL_FLAGS.get("dry_run",False)) handlers = {"status":cmd_status,"series":cmd_series,"lookup":cmd_lookup,"episodes":cmd_episodes,"calendar":cmd_calendar,"wanted":cmd_wanted} h = handlers.get(args.command) if not h: parser.print_help(); sys.exit(1) r = filtered_argv[filtered_argv.index(args.command)+1:] h(client, r) if __name__ == "__main__": main()