magnus919_agent-skills/arr-cli/scripts/sonarr-cli

#!/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,
                   series_type="Standard", season_folder=True):
        return self._post("/series", {"tvdbId":tvdb_id, "qualityProfileId":quality_profile,
                                      "rootFolderPath":root_folder_path, "monitored":monitored,
                                      "seriesType":series_type, "seasonFolder":season_folder,
                                      "addOptions":{"searchForMissingEpisodes":search}})
    def quality_profiles(self): return self._get("/qualityprofile")
    def quality_profile(self, pid): return self._get(f"/qualityprofile/{pid}")
    def root_folders(self): return self._get("/rootfolder")
    def command(self, name, **kwargs):
        return self._post("/command", {"name":name, **kwargs})

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 cmd_queue(client, args):
    p = argparse.ArgumentParser(prog="sonarr-cli queue")
    p.add_argument("--limit", type=int, default=20)
    parsed, _ = p.parse_known_args(args)
    if client.dry_run: return emit(f"[dry-run] List queue", {"dry_run":True})
    data = client.queue(limit=parsed.limit) or []
    if not data: return emit("Queue is empty.", {"queue":[]})
    lines, out = [], []
    for item in data:
        title = item.get("title","?"); status = item.get("status","?")
        pct = item.get("sizeleft",0) and item.get("size",1) and f"{100-item.get('sizeleft',0)*100//item.get('size',1):2d}%" or "?"
        lines.append(f"  {title:45} [{status:10}] {pct}")
        out.append({"title":title,"status":status,"progress":pct})
    emit(f"{len(data)} in queue:\n"+"\n".join(lines), {"queue":out})

def cmd_history(client, args):
    p = argparse.ArgumentParser(prog="sonarr-cli history")
    p.add_argument("--limit", type=int, default=20)
    p.add_argument("--event-type", help="Filter by event type")
    parsed, _ = p.parse_known_args(args)
    if client.dry_run: return emit(f"[dry-run] List history", {"dry_run":True})
    data = client.history(limit=parsed.limit) or {}
    records = data.get("records",[]) if isinstance(data,dict) else data[:parsed.limit]
    if not records: return emit("No history.", {"history":[]})
    lines, out = [], []
    for e in records[:parsed.limit]:
        title = e.get("series",{}).get("title","?") if isinstance(e.get("series"),dict) else e.get("seriesTitle","?")
        evt = e.get("eventType","?"); date = (e.get("date") or "")[:10]
        lines.append(f"  {date}  {title:45}  {evt}")
        out.append({"date":date,"title":title,"eventType":evt})
    emit(f"History:\n"+"\n".join(lines), {"history":out})

def cmd_add(client, args):
    p = argparse.ArgumentParser(prog="sonarr-cli add")
    p.add_argument("--tvdb-id", type=int, required=True, help="TVDb ID")
    p.add_argument("--root", required=True, help="Root folder path (e.g. /tv)")
    p.add_argument("--quality-profile", type=int, default=4, help="Quality profile ID (default: 4)")
    p.add_argument("--series-type", default="Standard", choices=["Standard","Daily","Anime"], help="Series type")
    p.add_argument("--unmonitored", action="store_true", help="Add unmonitored")
    p.add_argument("--search", action="store_true", help="Search for episodes after adding")
    p.add_argument("--no-season-folder", action="store_true", help="Flat episode storage")
    parsed, _ = p.parse_known_args(args)
    if client.dry_run:
        return emit(f"[dry-run] Add series tvdb={parsed.tvdb_id} type={parsed.series_type}",
                    {"dry_run":True, "tvdbId":parsed.tvdb_id, "qualityProfileId":parsed.quality_profile,
                     "rootFolderPath":parsed.root, "monitored":not parsed.unmonitored,
                     "seriesType":parsed.series_type, "seasonFolder":not parsed.no_season_folder})
    data = client.add_series(parsed.tvdb_id, quality_profile=parsed.quality_profile,
                             root_folder_path=parsed.root, monitored=not parsed.unmonitored,
                             search=parsed.search, series_type=parsed.series_type,
                             season_folder=not parsed.no_season_folder)
    emit(f"Added: #{data.get('id')} {data.get('title')} ({data.get('year')})",
         {"id":data.get("id"), "title":data.get("title"), "year":data.get("year")})

def cmd_quality_profile(client, args):
    p = argparse.ArgumentParser(prog="sonarr-cli quality-profile")
    p.add_argument("id", nargs="?", type=int, help="Profile ID (omit to list all)")
    parsed, _ = p.parse_known_args(args)
    if client.dry_run: return emit(f"[dry-run] Quality profile: {parsed.id or 'all'}", {"dry_run":True})
    if parsed.id:
        qp = client.quality_profile(parsed.id) or {}
        items = qp.get("items",[])
        lines = [f"  {qp.get('name','?')} (id={qp.get('id','?')})  cutoff={qp.get('cutoff',{}).get('name','?')}"]
        for item in items:
            name = item.get("quality",{}).get("name","?")
            allowed = "✓" if item.get("allowed") else "✗"
            lines.append(f"    {allowed} {name}")
        emit("\n".join(lines), qp)
    else:
        profiles = client.quality_profiles() or []
        lines = [f"  {p.get('id')}: {p.get('name','?')}" for p in profiles]
        emit(f"{len(profiles)} quality profile(s):\n"+"\n".join(lines), {"profiles":profiles})

def cmd_episode_files(client, args):
    p = argparse.ArgumentParser(prog="sonarr-cli episode-file")
    p.add_argument("--series-id", type=int, required=True, help="Sonarr series ID")
    parsed, _ = p.parse_known_args(args)
    if client.dry_run: return emit(f"[dry-run] List episode files for series {parsed.series_id}", {"dry_run":True})
    data = client.episode_files(parsed.series_id) or []
    if not data: return emit("No episode files found.", {"episodeFiles":[]})
    lines, out = [], []
    for f in data:
        sn = f.get("seasonNumber","?"); ep = f.get("episodeNumbers",[None])[0]
        qual = f.get("quality",{}).get("quality",{}).get("name","?")
        size = f.get("size",0)//1048576
        lines.append(f"  S{sn:02d}E{ep:02d} — {qual:20} {size}MB")
        out.append({"seasonNumber":sn,"episodeNumbers":f.get("episodeNumbers"),"quality":qual,"size":size})
    emit(f"{len(data)} episode file(s):\n"+"\n".join(lines), {"episodeFiles":out})

def cmd_search(client, args):
    p = argparse.ArgumentParser(prog="sonarr-cli search")
    p.add_argument("--series-id", type=int, required=True, help="Sonarr series ID")
    parsed, _ = p.parse_known_args(args)
    if client.dry_run:
        return emit(f"[dry-run] Search series id={parsed.series_id}", {"dry_run":True, "name":"SeriesSearch", "seriesIds":[parsed.series_id]})
    data = client.command("SeriesSearch", seriesIds=[parsed.series_id])
    emit(f"Search triggered: command #{data.get('id')} — {data.get('status','?')}",
         {"id":data.get("id"), "status":data.get("status"), "name":data.get("name")})

def cmd_root_folder(client, args):
    if client.dry_run: return emit("[dry-run] List root folders", {"dry_run":True})
    data = client.root_folders() or []
    if not data: return emit("No root folders configured.", {"rootFolders":[]})
    lines = [f"  {r.get('path','?')}  ({r.get('freeSpace',0)//1073741824}GB free)" for r in data]
    emit(f"{len(data)} root folder(s):\n"+"\n".join(lines), {"rootFolders":data})

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)
    sub.add_parser("queue")
    sub.add_parser("history")
    p_add = sub.add_parser("add", help="Add a series to library")
    p_add.add_argument("--tvdb-id", type=int, required=True)
    p_add.add_argument("--root", required=True)
    p_add.add_argument("--quality-profile", type=int, default=4)
    p_add.add_argument("--series-type", default="Standard", choices=["Standard","Daily","Anime"])
    p_add.add_argument("--unmonitored", action="store_true")
    p_add.add_argument("--search", action="store_true")
    p_add.add_argument("--no-season-folder", action="store_true")
    sub.add_parser("quality-profile", help="List or inspect quality profiles").add_argument("id", nargs="?", type=int)
    p_ef = sub.add_parser("episode-file", help="List downloaded episode files")
    p_ef.add_argument("--series-id", type=int, required=True)
    sub.add_parser("search", help="Trigger automatic search").add_argument("--series-id", type=int, required=True)
    sub.add_parser("root-folder", help="List configured root folders")
    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,"queue":cmd_queue,"history":cmd_history,
                "add":cmd_add,"quality-profile":cmd_quality_profile,"episode-file":cmd_episode_files,
                "search":cmd_search,"root-folder":cmd_root_folder}
    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()