magnus919_agent-skills/prowlarr-cli/scripts/prowlarr-cli

#!/usr/bin/env python3
"""prowlarr-cli — Prowlarr indexer management from the terminal."""

import argparse, json, os, sys, warnings
from typing import Any, Dict, List
warnings.simplefilter("ignore")
import requests

ENV_SERVER = os.getenv("ARR_SERVER_PROWLARR", "http://localhost:9696")
ENV_KEY = os.getenv("ARR_KEY_PROWLARR", "")

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 ProwlarrClient:
    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/v1{path}"
        if self.dry_run: return {"dry_run":True, "url":url, "params":params}
        if not self.key: die("ARR_KEY_PROWLARR not set. Get one from Prowlarr → 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_PROWLARR.")
        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/v1{path}"
        if self.dry_run: return {"dry_run":True, "url":url, "json":data}
        if not self.key: die("ARR_KEY_PROWLARR not set.")
        try:
            r = requests.post(url, json=data, headers={"X-Api-Key":self.key,"Content-Type":"application/json"}, timeout=60)
        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 indexers(self): return self._get("/indexer")
    def indexer(self, idx_id): return self._get(f"/indexer/{idx_id}")
    def indexer_stats(self): return self._get("/indexerstats")
    def indexer_status(self): return self._get("/indexerstatus")
    def applications(self): return self._get("/applications")
    def download_clients(self): return self._get("/downloadclient")
    def history(self, page=1, limit=20): return self._get("/history", {"page":page, "pageSize":limit})
    def health(self): return self._get("/health")
    def tags(self): return self._get("/tag")
    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"🔍 Prowlarr v{d.get('version','?')}  OS: {d.get('osName','?')}  DB: {d.get('databaseVersion','?')}",
         {"version":d.get("version"),"osName":d.get("osName")})

def cmd_indexers(client, args):
    p = argparse.ArgumentParser(prog="prowlarr-cli indexers")
    p.add_argument("--limit", type=int, default=50)
    parsed, _ = p.parse_known_args(args)
    if client.dry_run: return emit("[dry-run] List indexers", {"dry_run":True})
    data = client.indexers() or []
    data = data[:parsed.limit]
    if not data: return emit("No indexers configured.", {"indexers":[]})
    lines, out = [], []
    for idx in data:
        name = idx.get("name","?"); protocol = idx.get("protocol","?")
        enabled = idx.get("enable",False); priority = idx.get("priority",0)
        status = idx.get("status","?")
        icon = "✅" if enabled else "🚫"
        lines.append(f"  {icon} {name:35}  {protocol:7}  pri={priority}  [{status}]")
        out.append({"name":name,"protocol":protocol,"enable":enabled,"priority":priority,
                     "status":status,"id":idx.get("id"),"implementation":idx.get("implementationName")})
    emit(f"{len(data)} indexer(s):\n"+"\n".join(lines), {"indexers":out})

def cmd_indexer(client, args):
    p = argparse.ArgumentParser(prog="prowlarr-cli indexer")
    p.add_argument("id", type=int, help="Indexer ID")
    parsed, _ = p.parse_known_args(args)
    if client.dry_run: return emit(f"[dry-run] Get indexer {parsed.id}", {"dry_run":True})
    idx = client.indexer(parsed.id) or {}
    lines = [
        f"  {idx.get('name','?')} (id={idx.get('id','?')})",
        f"  Implementation: {idx.get('implementationName','?')}  Protocol: {idx.get('protocol','?')}",
        f"  Enabled: {idx.get('enable',False)}  Priority: {idx.get('priority',0)}",
        f"  Status: {idx.get('status','?')}",
        f"  Language: {idx.get('language','?')}",
    ]
    emit("\n".join(lines), idx)

def cmd_indexer_stats(client, args):
    if client.dry_run: return emit("[dry-run] Get indexer statistics", {"dry_run":True})
    data = client.indexer_stats() or {}
    lines = [f"  Total queries: {data.get('totalQueryCount',0)}"]
    for idx in data.get("indexers",[]):
        name = idx.get("indexerName","?")
        q = idx.get("queryCount",0); g = idx.get("grabCount",0)
        avg = idx.get("averageResponseTime",0)
        lines.append(f"  {name:35}  {q:5} queries  {g:5} grabs  {avg}ms avg")
    emit("\n".join(lines) if len(lines)>1 else "No statistics available.", data)

def cmd_indexer_status(client, args):
    if client.dry_run: return emit("[dry-run] Get indexer status", {"dry_run":True})
    data = client.indexer_status() or []
    if not data: return emit("All indexers OK.", {"indexerStatus":[]})
    lines = [f"  {s.get('indexerName','?'):35}  [{s.get('status','?')}]  {s.get('disabledTillUser','?') or 'ok'}" for s in data]
    emit(f"{len(data)} indexer(s) with status:\n"+"\n".join(lines), {"indexerStatus":data})

def cmd_applications(client, args):
    if client.dry_run: return emit("[dry-run] List applications", {"dry_run":True})
    data = client.applications() or []
    if not data: return emit("No applications connected.", {"applications":[]})
    lines, out = [], []
    for app in data:
        name = app.get("name","?"); sync = app.get("syncLevel","?")
        enabled = app.get("enabled",False)
        icon = "✅" if enabled else "🚫"
        lines.append(f"  {icon} {name:30}  sync={sync}")
        out.append({"name":name,"syncLevel":sync,"enabled":enabled,"id":app.get("id")})
    emit(f"{len(data)} application(s):\n"+"\n".join(lines), {"applications":out})

def cmd_download_clients(client, args):
    if client.dry_run: return emit("[dry-run] List download clients", {"dry_run":True})
    data = client.download_clients() or []
    if not data: return emit("No download clients configured.", {"downloadClients":[]})
    lines, out = [], []
    for dc in data:
        name = dc.get("name","?"); enabled = dc.get("enable",False)
        protocol = dc.get("protocol","?")
        icon = "✅" if enabled else "🚫"
        lines.append(f"  {icon} {name:30}  {protocol:5}")
        out.append({"name":name,"enable":enabled,"protocol":protocol,"id":dc.get("id")})
    emit(f"{len(data)} download client(s):\n"+"\n".join(lines), {"downloadClients":out})

def cmd_history(client, args):
    p = argparse.ArgumentParser(prog="prowlarr-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("[dry-run] List history", {"dry_run":True})
    data = client.history(limit=parsed.limit) or {}
    records = data.get("records",[])
    if parsed.event_type:
        records = [r for r in records if r.get("eventType") == parsed.event_type]
    if not records: return emit("No history.", {"history":[]})
    lines, out = [], []
    for e in records[:parsed.limit]:
        idx_name = e.get("indexer",{}).get("name","?") if isinstance(e.get("indexer"),dict) else e.get("indexerName","?")
        evt = e.get("eventType","?"); title = e.get("title","?")
        date = (e.get("date") or "")[:10]
        lines.append(f"  {date}  {title:40}  {idx_name:25}  {evt}")
        out.append({"date":date,"title":title,"indexer":idx_name,"eventType":evt,"id":e.get("id")})
    emit(f"History:\n"+"\n".join(lines), {"history":out})

def cmd_health(client, args):
    if client.dry_run: return emit("[dry-run] Get health", {"dry_run":True})
    data = client.health() or []
    if not data: return emit("All healthy.", {"health":[]})
    lines = [f"  {h.get('type','?'):25} — {h.get('message','?')}" for h in data]
    emit(f"{len(data)} warning(s):\n"+"\n".join(lines), {"health":data})

def cmd_tags(client, args):
    if client.dry_run: return emit("[dry-run] List tags", {"dry_run":True})
    data = client.tags() or []
    if not data: return emit("No tags.", {"tags":[]})
    lines = [f"  {t.get('id')}: {t.get('label','?')}" for t in data]
    emit(f"{len(data)} tag(s):\n"+"\n".join(lines), {"tags":data})

def cmd_test_all(client, args):
    if client.dry_run: return emit("[dry-run] Test all indexers", {"dry_run":True})
    data = client.command("TestAllIndexers")
    emit(f"Testing triggered: command #{data.get('id')} — {data.get('status','?')}",
         {"id":data.get("id"), "status":data.get("status"), "name":data.get("name")})

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="prowlarr-cli", description="Prowlarr indexer management.",
        epilog="Set ARR_SERVER_PROWLARR and ARR_KEY_PROWLARR.")
    sub = parser.add_subparsers(dest="command")
    sub.add_parser("status")
    sub.add_parser("indexers", help="List indexers").add_argument("--limit", type=int, default=50)
    sub.add_parser("indexer", help="Get indexer details").add_argument("id", type=int)
    sub.add_parser("indexer-stats", help="Indexer query/grab statistics")
    sub.add_parser("indexer-status", help="Indexer health status")
    sub.add_parser("applications", help="List connected *arr applications")
    sub.add_parser("download-clients", help="List download clients")
    sub.add_parser("history", help="Search history").add_argument("--limit", type=int, default=20)
    sub.add_parser("health", help="Health warnings")
    sub.add_parser("tags", help="List tags")
    sub.add_parser("test-all", help="Test all indexers")
    args = parser.parse_args(filtered_argv[1:])
    if not args.command: parser.print_help(); sys.exit(1)
    client = ProwlarrClient(dry_run=GLOBAL_FLAGS.get("dry_run",False))
    handlers = {
        "status":cmd_status, "indexers":cmd_indexers, "indexer":cmd_indexer,
        "indexer-stats":cmd_indexer_stats, "indexer-status":cmd_indexer_status,
        "applications":cmd_applications, "download-clients":cmd_download_clients,
        "history":cmd_history, "health":cmd_health, "tags":cmd_tags,
        "test-all":cmd_test_all,
    }
    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()