"""Thread-safe configuration file manager for runtime script updates.""" import logging import os import threading from pathlib import Path from typing import Optional import yaml from .config import ScriptConfig, settings logger = logging.getLogger(__name__) class ConfigManager: """Thread-safe configuration file manager.""" def __init__(self, config_path: Optional[Path] = None): """Initialize the config manager. Args: config_path: Path to config file. If None, will search standard locations. """ self._lock = threading.Lock() self._config_path = config_path or self._find_config_path() logger.info(f"ConfigManager initialized with path: {self._config_path}") def _find_config_path(self) -> Path: """Find the active config file path. Returns: Path to the config file. Raises: FileNotFoundError: If no config file is found. """ # Same search logic as Settings.load_from_yaml() search_paths = [ Path("config.yaml"), Path("config.yml"), ] # Add platform-specific config directory if os.name == "nt": # Windows appdata = os.environ.get("APPDATA", "") if appdata: search_paths.append(Path(appdata) / "media-server" / "config.yaml") else: # Linux/Unix/macOS 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 not found, use the default location 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 add_script(self, name: str, config: ScriptConfig) -> None: """Add a new script to config. Args: name: Script name (must be unique). config: Script configuration. Raises: ValueError: If script already exists. IOError: If config file cannot be written. """ with self._lock: # Read YAML if not self._config_path.exists(): data = {} else: with open(self._config_path, "r", encoding="utf-8") as f: data = yaml.safe_load(f) or {} # Check if script already exists if "scripts" in data and name in data["scripts"]: raise ValueError(f"Script '{name}' already exists") # Add script if "scripts" not in data: data["scripts"] = {} data["scripts"][name] = config.model_dump(exclude_none=True) # Write YAML self._config_path.parent.mkdir(parents=True, exist_ok=True) with open(self._config_path, "w", encoding="utf-8") as f: yaml.dump(data, f, default_flow_style=False, sort_keys=False) # Update in-memory settings settings.scripts[name] = config logger.info(f"Script '{name}' added to config") def update_script(self, name: str, config: ScriptConfig) -> None: """Update an existing script. Args: name: Script name. config: New script configuration. Raises: ValueError: If script does not exist. IOError: If config file cannot be written. """ with self._lock: # Read YAML if not self._config_path.exists(): raise ValueError(f"Config file not found: {self._config_path}") with open(self._config_path, "r", encoding="utf-8") as f: data = yaml.safe_load(f) or {} # Check if script exists if "scripts" not in data or name not in data["scripts"]: raise ValueError(f"Script '{name}' does not exist") # Update script data["scripts"][name] = config.model_dump(exclude_none=True) # Write YAML with open(self._config_path, "w", encoding="utf-8") as f: yaml.dump(data, f, default_flow_style=False, sort_keys=False) # Update in-memory settings settings.scripts[name] = config logger.info(f"Script '{name}' updated in config") def delete_script(self, name: str) -> None: """Delete a script from config. Args: name: Script name. Raises: ValueError: If script does not exist. IOError: If config file cannot be written. """ with self._lock: # Read YAML if not self._config_path.exists(): raise ValueError(f"Config file not found: {self._config_path}") with open(self._config_path, "r", encoding="utf-8") as f: data = yaml.safe_load(f) or {} # Check if script exists if "scripts" not in data or name not in data["scripts"]: raise ValueError(f"Script '{name}' does not exist") # Delete script del data["scripts"][name] # Write YAML with open(self._config_path, "w", encoding="utf-8") as f: yaml.dump(data, f, default_flow_style=False, sort_keys=False) # Update in-memory settings if name in settings.scripts: del settings.scripts[name] logger.info(f"Script '{name}' deleted from config") # Global config manager instance config_manager = ConfigManager()