#!/usr/bin/env python3 """Configure safe baseline ARR app settings from local environment values.""" from __future__ import annotations import json import os import sys import time import urllib.error import urllib.request from pathlib import Path SCRIPT_DIR = Path(__file__).resolve().parent ENV_FILE = Path(os.environ.get("ARR_ENV_FILE", SCRIPT_DIR / ".env")) def load_env(path: Path) -> None: if not path.exists(): return for raw_line in path.read_text(encoding="utf-8").splitlines(): line = raw_line.strip() if not line or line.startswith("#") or "=" not in line: continue key, value = line.split("=", 1) os.environ.setdefault(key.strip(), value.strip().strip('"').strip("'")) def request_json(method: str, url: str, api_key: str, payload: dict | None = None) -> object: data = None headers = {"X-Api-Key": api_key, "Accept": "application/json"} if payload is not None: data = json.dumps(payload).encode("utf-8") headers["Content-Type"] = "application/json" request = urllib.request.Request(url, data=data, headers=headers, method=method) with urllib.request.urlopen(request, timeout=15) as response: body = response.read() if not body: return {} return json.loads(body.decode("utf-8")) def wait_for_app(name: str, base_url: str, api_key: str) -> None: for attempt in range(1, 31): try: request_json("GET", f"{base_url}/api/v3/system/status", api_key) return except Exception as exc: # noqa: BLE001 if attempt == 30: raise RuntimeError(f"{name} did not become reachable at {base_url}") from exc time.sleep(2) def ensure_root_folder(name: str, base_url: str, api_key: str, path: str) -> None: existing = request_json("GET", f"{base_url}/api/v3/rootfolder", api_key) if isinstance(existing, list) and any(item.get("path") == path for item in existing if isinstance(item, dict)): print(f"{name}: root folder already exists: {path}") return request_json("POST", f"{base_url}/api/v3/rootfolder", api_key, {"path": path}) print(f"{name}: added root folder: {path}") def schema_fields(base_url: str, api_key: str, endpoint: str, implementation: str) -> list[dict]: schemas = request_json("GET", f"{base_url}/api/v3/{endpoint}/schema", api_key) if not isinstance(schemas, list): return [] for schema in schemas: if isinstance(schema, dict) and schema.get("implementation") == implementation: fields = schema.get("fields") return fields if isinstance(fields, list) else [] return [] def fill_fields(fields: list[dict], values: dict[str, object]) -> list[dict]: hydrated = [] for field in fields: if not isinstance(field, dict): continue item = dict(field) name = item.get("name") if name in values: item["value"] = values[name] hydrated.append(item) return hydrated def ensure_qbittorrent(name: str, base_url: str, api_key: str, category: str) -> None: clients = request_json("GET", f"{base_url}/api/v3/downloadclient", api_key) if isinstance(clients, list) and any( isinstance(item, dict) and item.get("implementation") == "QBittorrent" for item in clients ): print(f"{name}: qBittorrent download client already exists") return fields = schema_fields(base_url, api_key, "downloadclient", "QBittorrent") if not fields: raise RuntimeError(f"{name}: could not find qBittorrent download client schema") port = int(os.environ.get("QBITTORRENT_PORT", "8080")) values = { "host": "qbittorrent", "port": port, "urlBase": "", "username": os.environ.get("QBITTORRENT_USERNAME", "admin"), "password": os.environ.get("QBITTORRENT_PASSWORD", ""), "category": category, "movieCategory": category, "tvCategory": category, "recentMoviePriority": 0, "olderMoviePriority": 0, "recentTvPriority": 0, "olderTvPriority": 0, "addPaused": False, "useSsl": False, } payload = { "enable": True, "protocol": "torrent", "priority": 1, "removeCompletedDownloads": True, "removeFailedDownloads": True, "name": "qBittorrent", "implementation": "QBittorrent", "configContract": "QBittorrentSettings", "fields": fill_fields(fields, values), "tags": [], } request_json("POST", f"{base_url}/api/v3/downloadclient", api_key, payload) print(f"{name}: added qBittorrent download client") def apply_import_lists(name: str, base_url: str, api_key: str, path_env: str) -> None: list_path = os.environ.get(path_env, "") if not list_path: return path = Path(list_path) if not path.exists(): raise RuntimeError(f"{name}: {path_env} points to missing file {path}") desired = json.loads(path.read_text(encoding="utf-8")) if not isinstance(desired, list): raise RuntimeError(f"{name}: {path} must contain a JSON array") existing = request_json("GET", f"{base_url}/api/v3/importlist", api_key) existing_names = set() if isinstance(existing, list): existing_names = {item.get("name") for item in existing if isinstance(item, dict)} for item in desired: if not isinstance(item, dict) or not item.get("name"): raise RuntimeError(f"{name}: every import-list entry must be an object with a name") if item["name"] in existing_names: print(f"{name}: import list already exists: {item['name']}") continue request_json("POST", f"{base_url}/api/v3/importlist", api_key, item) print(f"{name}: added import list: {item['name']}") def required_env(name: str) -> str: value = os.environ.get(name, "") if not value: raise RuntimeError(f"Set {name} in {ENV_FILE} before running this script") return value def main() -> int: load_env(ENV_FILE) radarr_key = required_env("RADARR_API_KEY") sonarr_key = required_env("SONARR_API_KEY") radarr_url = os.environ.get("RADARR_URL", "http://127.0.0.1:7878") sonarr_url = os.environ.get("SONARR_URL", "http://127.0.0.1:8989") wait_for_app("Radarr", radarr_url, radarr_key) wait_for_app("Sonarr", sonarr_url, sonarr_key) ensure_root_folder("Radarr", radarr_url, radarr_key, os.environ.get("RADARR_ROOT_FOLDER", "/data/media/movies")) ensure_root_folder("Sonarr", sonarr_url, sonarr_key, os.environ.get("SONARR_ROOT_FOLDER", "/data/media/tv")) ensure_qbittorrent("Radarr", radarr_url, radarr_key, "radarr") ensure_qbittorrent("Sonarr", sonarr_url, sonarr_key, "sonarr") apply_import_lists("Radarr", radarr_url, radarr_key, "RADARR_IMPORT_LISTS_JSON") apply_import_lists("Sonarr", sonarr_url, sonarr_key, "SONARR_IMPORT_LISTS_JSON") return 0 if __name__ == "__main__": try: raise SystemExit(main()) except (RuntimeError, urllib.error.URLError, urllib.error.HTTPError) as error: print(f"error: {error}", file=sys.stderr) raise SystemExit(1)