my-homelab-configs/infra/arr-stack/configure-arr.py

#!/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)