alexei.dolgolyov
a0af855846
- Add callback CRUD endpoints (create, update, delete, list) - Add callback management UI with all 11 callback events support - Add light/dark theme switcher with localStorage persistence - Improve button styling (wider buttons, simplified text) - Extend ConfigManager with callback operations Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
285 lines
9.5 KiB
Python
285 lines
9.5 KiB
Python
"""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 CallbackConfig, 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")
|
|
|
|
def add_callback(self, name: str, config: CallbackConfig) -> None:
|
|
"""Add a new callback to config.
|
|
|
|
Args:
|
|
name: Callback name (must be unique).
|
|
config: Callback configuration.
|
|
|
|
Raises:
|
|
ValueError: If callback 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 callback already exists
|
|
if "callbacks" in data and name in data["callbacks"]:
|
|
raise ValueError(f"Callback '{name}' already exists")
|
|
|
|
# Add callback
|
|
if "callbacks" not in data:
|
|
data["callbacks"] = {}
|
|
data["callbacks"][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.callbacks[name] = config
|
|
|
|
logger.info(f"Callback '{name}' added to config")
|
|
|
|
def update_callback(self, name: str, config: CallbackConfig) -> None:
|
|
"""Update an existing callback.
|
|
|
|
Args:
|
|
name: Callback name.
|
|
config: New callback configuration.
|
|
|
|
Raises:
|
|
ValueError: If callback 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 callback exists
|
|
if "callbacks" not in data or name not in data["callbacks"]:
|
|
raise ValueError(f"Callback '{name}' does not exist")
|
|
|
|
# Update callback
|
|
data["callbacks"][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.callbacks[name] = config
|
|
|
|
logger.info(f"Callback '{name}' updated in config")
|
|
|
|
def delete_callback(self, name: str) -> None:
|
|
"""Delete a callback from config.
|
|
|
|
Args:
|
|
name: Callback name.
|
|
|
|
Raises:
|
|
ValueError: If callback 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 callback exists
|
|
if "callbacks" not in data or name not in data["callbacks"]:
|
|
raise ValueError(f"Callback '{name}' does not exist")
|
|
|
|
# Delete callback
|
|
del data["callbacks"][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.callbacks:
|
|
del settings.callbacks[name]
|
|
|
|
logger.info(f"Callback '{name}' deleted from config")
|
|
|
|
|
|
# Global config manager instance
|
|
config_manager = ConfigManager()
|