import os import yaml DEFAULT_CONFIG = { "recording": { "download_path": "/app/data/recordings", "max_retries": 5, "retry_delay": 5, }, "server": { "host": "0.0.0.0", "port": 8484, }, "auth": { "enabled": False, "username": "admin", "password": "stream-recorder", }, "ntfy": { "url": "", "token": "", "events": "error", }, "plik": { "url": "", "api_key": "", "ttl": "30d", }, "logging": { "level": "INFO", }, } class Config: def __init__(self, config_path: str = "/app/data/config.yml"): self.config_path = config_path self.data = {} self.load() def load(self): self.data = _deep_copy(DEFAULT_CONFIG) if os.path.exists(self.config_path): with open(self.config_path, "r") as f: user_config = yaml.safe_load(f) or {} _deep_merge(self.data, user_config) else: self.save() def save(self): os.makedirs(os.path.dirname(self.config_path), exist_ok=True) with open(self.config_path, "w") as f: yaml.dump(self.data, f, default_flow_style=False, allow_unicode=True, sort_keys=False) def get(self, *keys, default=None): val = self.data for key in keys: if isinstance(val, dict) and key in val: val = val[key] else: return default return val def _deep_copy(d): if isinstance(d, dict): return {k: _deep_copy(v) for k, v in d.items()} if isinstance(d, list): return [_deep_copy(v) for v in d] return d def _deep_merge(base, override): for key, value in override.items(): if key in base and isinstance(base[key], dict) and isinstance(value, dict): _deep_merge(base[key], value) else: base[key] = value