Initial commit

backend/sortarr/config.py

@@ -0,0 +1,71 @@
+from __future__ import annotations
+
+import copy
+import os
+import tomllib
+from pathlib import Path
+from typing import Any
+
+
+def _read_toml(path: Path) -> dict[str, Any]:
+    if not path.exists():
+        return {}
+    with path.open("rb") as handle:
+        return tomllib.load(handle)
+
+
+def _merge(base: dict[str, Any], override: dict[str, Any]) -> dict[str, Any]:
+    merged = copy.deepcopy(base)
+    for key, value in override.items():
+        if isinstance(value, dict) and isinstance(merged.get(key), dict):
+            merged[key] = _merge(merged[key], value)
+        else:
+            merged[key] = copy.deepcopy(value)
+    return merged
+
+
+def _bool(value: str) -> bool:
+    return value.strip().lower() in {"1", "true", "yes", "on"}
+
+
+def load_config() -> dict[str, Any]:
+    default_path = Path(os.getenv("SORTARR_DEFAULT_CONFIG", "/app/default-config/app.toml"))
+    user_path = Path(os.getenv("SORTARR_CONFIG", "/config/app.toml"))
+    config = _merge(_read_toml(default_path), _read_toml(user_path))
+
+    app = config.setdefault("app", {})
+    paths = config.setdefault("paths", {})
+
+    env_map = {
+        "SORTARR_DRY_RUN": ("app", "dry_run", _bool),
+        "SORTARR_LOG_LEVEL": ("app", "log_level", str),
+        "SORTARR_SCAN_INTERVAL_SECONDS": ("app", "scan_interval_seconds", int),
+        "SORTARR_SETTLE_SECONDS": ("app", "settle_seconds", int),
+        "SORTARR_DATA_DIR": ("paths", "data", str),
+        "SORTARR_LOG_DIR": ("paths", "logs", str),
+        "SORTARR_CACHE_DIR": ("paths", "cache", str),
+        "TMDB_API_KEY": ("metadata", "tmdb_api_key", str),
+        "TMDB_BEARER_TOKEN": ("metadata", "tmdb_bearer_token", str),
+    }
+    for env, (section, key, caster) in env_map.items():
+        if os.getenv(env) not in (None, ""):
+            config.setdefault(section, {})[key] = caster(os.environ[env])
+
+    if os.getenv("SORTARR_MIN_FREE_GB"):
+        for drive in config.get("drives", []):
+            drive["min_free_gb"] = int(os.environ["SORTARR_MIN_FREE_GB"])
+
+    Path(paths.get("data", "/data")).mkdir(parents=True, exist_ok=True)
+    Path(paths.get("logs", "/logs")).mkdir(parents=True, exist_ok=True)
+    Path(paths.get("cache", str(Path(paths.get("data", "/data")) / "cache"))).mkdir(parents=True, exist_ok=True)
+    app.setdefault("dry_run", True)
+    return config
+
+
+def public_config(config: dict[str, Any]) -> dict[str, Any]:
+    clone = copy.deepcopy(config)
+    metadata = clone.get("metadata", {})
+    for key in ("tmdb_api_key", "tmdb_bearer_token"):
+        if metadata.get(key):
+            metadata[key] = "********"
+    return clone