"""Konfigurationsmanagement - Singleton fuer Settings und Presets""" import os import logging import yaml from pathlib import Path from typing import Optional from logging.handlers import TimedRotatingFileHandler, RotatingFileHandler class Config: """Laedt und verwaltet settings.yaml und presets.yaml""" _instance: Optional['Config'] = None def __new__(cls) -> 'Config': if cls._instance is None: cls._instance = super().__new__(cls) cls._instance._initialized = False return cls._instance def __init__(self) -> None: if self._initialized: return self._initialized = True self._base_path = Path(__file__).parent self._cfg_path = self._base_path / "cfg" self._log_path = self._base_path.parent / "logs" self._data_path = self._base_path.parent / "data" # Verzeichnisse sicherstellen self._log_path.mkdir(parents=True, exist_ok=True) self._data_path.mkdir(parents=True, exist_ok=True) self.settings: dict = {} self.presets: dict = {} self._load_settings() self._load_presets() self._apply_env_overrides() def _load_settings(self) -> None: """Laedt settings.yaml""" settings_file = self._cfg_path / "settings.yaml" try: with open(settings_file, "r", encoding="utf-8") as f: self.settings = yaml.safe_load(f) or {} logging.info(f"Settings geladen: {settings_file}") except FileNotFoundError: logging.error(f"Settings nicht gefunden: {settings_file}") self.settings = {} def _load_presets(self) -> None: """Laedt presets.yaml""" presets_file = self._cfg_path / "presets.yaml" try: with open(presets_file, "r", encoding="utf-8") as f: self.presets = yaml.safe_load(f) or {} logging.info(f"Presets geladen: {presets_file}") except FileNotFoundError: logging.error(f"Presets nicht gefunden: {presets_file}") self.presets = {} def _apply_env_overrides(self) -> None: """Umgebungsvariablen ueberschreiben Settings""" env_mode = os.environ.get("VIDEO_KONVERTER_MODE") if env_mode and env_mode in ("cpu", "gpu", "auto"): self.settings.setdefault("encoding", {})["mode"] = env_mode logging.info(f"Encoding-Modus per Umgebungsvariable: {env_mode}") def save_settings(self) -> None: """Schreibt aktuelle Settings zurueck in settings.yaml""" settings_file = self._cfg_path / "settings.yaml" try: with open(settings_file, "w", encoding="utf-8") as f: yaml.dump(self.settings, f, default_flow_style=False, indent=2, allow_unicode=True) logging.info("Settings gespeichert") except Exception as e: logging.error(f"Settings speichern fehlgeschlagen: {e}") def save_presets(self) -> None: """Schreibt Presets zurueck in presets.yaml""" presets_file = self._cfg_path / "presets.yaml" try: with open(presets_file, "w", encoding="utf-8") as f: yaml.dump(self.presets, f, default_flow_style=False, indent=2, allow_unicode=True) logging.info("Presets gespeichert") except Exception as e: logging.error(f"Presets speichern fehlgeschlagen: {e}") def setup_logging(self) -> None: """Konfiguriert Logging mit Rotation""" log_cfg = self.settings.get("logging", {}) log_level = log_cfg.get("level", "INFO") log_file = log_cfg.get("file", "server.log") log_mode = log_cfg.get("rotation", "time") backup_count = log_cfg.get("backup_count", 7) log_path = self._log_path / log_file handlers = [logging.StreamHandler()] if log_mode == "time": file_handler = TimedRotatingFileHandler( str(log_path), when="midnight", interval=1, backupCount=backup_count, encoding="utf-8" ) else: max_bytes = log_cfg.get("max_size_mb", 10) * 1024 * 1024 file_handler = RotatingFileHandler( str(log_path), maxBytes=max_bytes, backupCount=backup_count, encoding="utf-8" ) handlers.append(file_handler) # force=True weil Config.__init__ logging aufruft bevor setup_logging() logging.basicConfig( level=getattr(logging, log_level, logging.INFO), format="%(asctime)s - %(levelname)s - %(message)s", handlers=handlers, force=True, ) # --- Properties fuer haeufig benoetigte Werte --- @property def encoding_mode(self) -> str: return self.settings.get("encoding", {}).get("mode", "cpu") @property def gpu_device(self) -> str: return self.settings.get("encoding", {}).get("gpu_device", "/dev/dri/renderD128") @property def max_parallel_jobs(self) -> int: return self.settings.get("encoding", {}).get("max_parallel_jobs", 1) @property def target_container(self) -> str: return self.settings.get("files", {}).get("target_container", "webm") @property def default_preset_name(self) -> str: return self.settings.get("encoding", {}).get("default_preset", "cpu_av1") @property def default_preset(self) -> dict: name = self.default_preset_name return self.presets.get(name, {}) @property def data_path(self) -> Path: return self._data_path @property def audio_config(self) -> dict: return self.settings.get("audio", {}) @property def subtitle_config(self) -> dict: return self.settings.get("subtitle", {}) @property def files_config(self) -> dict: return self.settings.get("files", {}) @property def cleanup_config(self) -> dict: return self.settings.get("cleanup", {}) @property def server_config(self) -> dict: return self.settings.get("server", {})