alexei.dolgolyov
99dbbb1019
- Add LinkConfig model and links field to settings - Add CRUD API endpoints for links (list/create/update/delete) - Add Links management tab in WebUI with add/edit/delete dialogs - Add live icon preview in Link and Script dialog forms - Show MDI icons inline in Quick Actions cards, Scripts table, Links table - Add broadcast_links_changed WebSocket event for live updates - Add EN/RU translations for all links management strings Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
457 lines
16 KiB
Python
457 lines
16 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, LinkConfig, MediaFolderConfig, 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")
|
|
|
|
def add_media_folder(self, folder_id: str, config: MediaFolderConfig) -> None:
|
|
"""Add a new media folder to config.
|
|
|
|
Args:
|
|
folder_id: Folder ID (must be unique).
|
|
config: Media folder configuration.
|
|
|
|
Raises:
|
|
ValueError: If folder 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 folder already exists
|
|
if "media_folders" in data and folder_id in data["media_folders"]:
|
|
raise ValueError(f"Media folder '{folder_id}' already exists")
|
|
|
|
# Add folder
|
|
if "media_folders" not in data:
|
|
data["media_folders"] = {}
|
|
data["media_folders"][folder_id] = 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.media_folders[folder_id] = config
|
|
|
|
logger.info(f"Media folder '{folder_id}' added to config")
|
|
|
|
def update_media_folder(self, folder_id: str, config: MediaFolderConfig) -> None:
|
|
"""Update an existing media folder.
|
|
|
|
Args:
|
|
folder_id: Folder ID.
|
|
config: New media folder configuration.
|
|
|
|
Raises:
|
|
ValueError: If folder 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 folder exists
|
|
if "media_folders" not in data or folder_id not in data["media_folders"]:
|
|
raise ValueError(f"Media folder '{folder_id}' does not exist")
|
|
|
|
# Update folder
|
|
data["media_folders"][folder_id] = 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.media_folders[folder_id] = config
|
|
|
|
logger.info(f"Media folder '{folder_id}' updated in config")
|
|
|
|
def delete_media_folder(self, folder_id: str) -> None:
|
|
"""Delete a media folder from config.
|
|
|
|
Args:
|
|
folder_id: Folder ID.
|
|
|
|
Raises:
|
|
ValueError: If folder 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 folder exists
|
|
if "media_folders" not in data or folder_id not in data["media_folders"]:
|
|
raise ValueError(f"Media folder '{folder_id}' does not exist")
|
|
|
|
# Delete folder
|
|
del data["media_folders"][folder_id]
|
|
|
|
# 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 folder_id in settings.media_folders:
|
|
del settings.media_folders[folder_id]
|
|
|
|
logger.info(f"Media folder '{folder_id}' deleted from config")
|
|
|
|
def add_link(self, name: str, config: LinkConfig) -> None:
|
|
"""Add a new link to config."""
|
|
with self._lock:
|
|
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 {}
|
|
|
|
if "links" in data and name in data["links"]:
|
|
raise ValueError(f"Link '{name}' already exists")
|
|
|
|
if "links" not in data:
|
|
data["links"] = {}
|
|
data["links"][name] = config.model_dump(exclude_none=True)
|
|
|
|
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)
|
|
|
|
settings.links[name] = config
|
|
logger.info(f"Link '{name}' added to config")
|
|
|
|
def update_link(self, name: str, config: LinkConfig) -> None:
|
|
"""Update an existing link."""
|
|
with self._lock:
|
|
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 {}
|
|
|
|
if "links" not in data or name not in data["links"]:
|
|
raise ValueError(f"Link '{name}' does not exist")
|
|
|
|
data["links"][name] = config.model_dump(exclude_none=True)
|
|
|
|
with open(self._config_path, "w", encoding="utf-8") as f:
|
|
yaml.dump(data, f, default_flow_style=False, sort_keys=False)
|
|
|
|
settings.links[name] = config
|
|
logger.info(f"Link '{name}' updated in config")
|
|
|
|
def delete_link(self, name: str) -> None:
|
|
"""Delete a link from config."""
|
|
with self._lock:
|
|
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 {}
|
|
|
|
if "links" not in data or name not in data["links"]:
|
|
raise ValueError(f"Link '{name}' does not exist")
|
|
|
|
del data["links"][name]
|
|
|
|
with open(self._config_path, "w", encoding="utf-8") as f:
|
|
yaml.dump(data, f, default_flow_style=False, sort_keys=False)
|
|
|
|
if name in settings.links:
|
|
del settings.links[name]
|
|
logger.info(f"Link '{name}' deleted from config")
|
|
|
|
|
|
# Global config manager instance
|
|
config_manager = ConfigManager()
|