"""Thread-safe configuration file manager for runtime updates.""" import logging import os import threading from pathlib import Path from typing import Any, Optional import yaml from .config import ( CallbackConfig, LinkConfig, MediaFolderConfig, ScriptConfig, _restrict_config_perms, _write_yaml_atomic, settings, ) logger = logging.getLogger(__name__) class ConfigManager: """Thread-safe configuration file manager. All writes go through ``_save()`` which writes to ``config.yaml.tmp`` and then ``os.replace()``s it into place so a crash mid-write cannot corrupt the only persistent user data. On POSIX the file is also chmodded to 0600 so co-tenant users cannot read the API token. """ def __init__(self, config_path: Optional[Path] = None): self._lock = threading.Lock() self._config_path = config_path or self._find_config_path() logger.info(f"ConfigManager initialized with path: {self._config_path}") @staticmethod def _find_config_path() -> Path: """Find the active config file path (or the default if none exists yet).""" search_paths = [Path("config.yaml"), Path("config.yml")] if os.name == "nt": appdata = os.environ.get("APPDATA", "") if appdata: search_paths.append(Path(appdata) / "media-server" / "config.yaml") else: search_paths.append(Path.home() / ".config" / "media-server" / "config.yaml") search_paths.append(Path("/etc/media-server/config.yaml")) for search_path in search_paths: if search_path.exists(): return search_path if os.name == "nt": default_path = Path(os.environ.get("APPDATA", "")) / "media-server" / "config.yaml" else: default_path = Path.home() / ".config" / "media-server" / "config.yaml" logger.warning(f"No config file found, using default path: {default_path}") return default_path def _load(self) -> dict[str, Any]: """Read the config YAML, returning an empty dict if the file is missing.""" if not self._config_path.exists(): return {} with open(self._config_path, "r", encoding="utf-8") as f: return yaml.safe_load(f) or {} def _save(self, data: dict[str, Any]) -> None: """Atomically write the config YAML and lock down its permissions.""" self._config_path.parent.mkdir(parents=True, exist_ok=True) _write_yaml_atomic(self._config_path, data) _restrict_config_perms(self._config_path) # --- Generic per-section CRUD -------------------------------------- def _upsert( self, section: str, key: str, value: Any, *, require_absent: bool = False, require_present: bool = False, in_memory_target: dict[str, Any] | None = None, verb: str = "set", ) -> None: with self._lock: data = self._load() existing = data.get(section, {}) if require_absent and key in existing: raise ValueError(f"{section[:-1].title()} '{key}' already exists") if require_present and (not isinstance(existing, dict) or key not in existing): raise ValueError(f"{section[:-1].title()} '{key}' does not exist") if not isinstance(existing, dict): existing = {} existing[key] = value.model_dump(exclude_none=True) data[section] = existing self._save(data) if in_memory_target is not None: in_memory_target[key] = value logger.info(f"{section[:-1].title()} '{key}' {verb} in config") def _delete( self, section: str, key: str, *, in_memory_target: dict[str, Any] | None = None, ) -> None: with self._lock: data = self._load() existing = data.get(section, {}) if not isinstance(existing, dict) or key not in existing: raise ValueError(f"{section[:-1].title()} '{key}' does not exist") del existing[key] data[section] = existing self._save(data) if in_memory_target is not None and key in in_memory_target: del in_memory_target[key] logger.info(f"{section[:-1].title()} '{key}' deleted from config") # --- Scripts ------------------------------------------------------- def add_script(self, name: str, config: ScriptConfig) -> None: self._upsert( "scripts", name, config, require_absent=True, in_memory_target=settings.scripts, verb="added", ) def update_script(self, name: str, config: ScriptConfig) -> None: self._upsert( "scripts", name, config, require_present=True, in_memory_target=settings.scripts, verb="updated", ) def delete_script(self, name: str) -> None: self._delete("scripts", name, in_memory_target=settings.scripts) # --- Callbacks ----------------------------------------------------- def add_callback(self, name: str, config: CallbackConfig) -> None: self._upsert( "callbacks", name, config, require_absent=True, in_memory_target=settings.callbacks, verb="added", ) def update_callback(self, name: str, config: CallbackConfig) -> None: self._upsert( "callbacks", name, config, require_present=True, in_memory_target=settings.callbacks, verb="updated", ) def delete_callback(self, name: str) -> None: self._delete("callbacks", name, in_memory_target=settings.callbacks) # --- Media folders ------------------------------------------------- def add_media_folder(self, folder_id: str, config: MediaFolderConfig) -> None: self._upsert( "media_folders", folder_id, config, require_absent=True, in_memory_target=settings.media_folders, verb="added", ) def update_media_folder(self, folder_id: str, config: MediaFolderConfig) -> None: self._upsert( "media_folders", folder_id, config, require_present=True, in_memory_target=settings.media_folders, verb="updated", ) def delete_media_folder(self, folder_id: str) -> None: self._delete("media_folders", folder_id, in_memory_target=settings.media_folders) # --- Links --------------------------------------------------------- def add_link(self, name: str, config: LinkConfig) -> None: self._upsert( "links", name, config, require_absent=True, in_memory_target=settings.links, verb="added", ) def update_link(self, name: str, config: LinkConfig) -> None: self._upsert( "links", name, config, require_present=True, in_memory_target=settings.links, verb="updated", ) def delete_link(self, name: str) -> None: self._delete("links", name, in_memory_target=settings.links) # --- Top-level settings -------------------------------------------- def set_setting(self, key: str, value: Any) -> None: """Set a top-level config setting and persist to YAML.""" with self._lock: data = self._load() if value is None: data.pop(key, None) else: data[key] = value self._save(data) if hasattr(settings, key): setattr(settings, key, value) logger.info("Setting '%s' updated to: %s", key, value) config_manager = ConfigManager()