"""Configuration management for BrainRotGuard.""" import logging import os import re from dataclasses import dataclass, field from pathlib import Path from typing import Any from urllib.parse import urlparse import yaml logger = logging.getLogger(__name__) def expand_env_vars(value: Any) -> Any: """Recursively expand environment variables in strings, dicts, and lists. Supports both ${VAR} and $VAR patterns. """ if isinstance(value, str): # Expand ${VAR} pattern pattern = re.compile(r'\$\{([^}]+)\}') result = pattern.sub(lambda m: os.environ.get(m.group(1), ''), value) # Expand $VAR pattern pattern = re.compile(r'\$([A-Za-z_][A-Za-z0-9_]*)') result = pattern.sub(lambda m: os.environ.get(m.group(1), ''), result) return result elif isinstance(value, dict): return {k: expand_env_vars(v) for k, v in value.items()} elif isinstance(value, list): return [expand_env_vars(item) for item in value] else: return value VALID_LOG_LEVELS = {"debug", "info", "warning", "error"} @dataclass class AppConfig: """Application-wide configuration.""" locale: str = "en" time_format: str = "locale" log_level: str = "info" def __post_init__(self): self.log_level = self.log_level.lower() if self.log_level not in VALID_LOG_LEVELS: logger.warning(f"Invalid log_level '{self.log_level}', defaulting to 'info'") self.log_level = "info" @dataclass class WebConfig: """Web server configuration.""" host: str = "0.0.0.0" port: int = 8080 poll_interval: int = 3000 # ms between status checks on pending page pin: str = "" # empty = no auth required session_secret: str = "" # auto-generated if not set base_url: str = "" # e.g. http://10.0.0.1:8080 — used for links in Telegram messages embed_host: str = "" # origin the player iframe is loaded from # youtube-nocookie keeps www.youtube.com out of the DNS allowlist entirely, but # it also drops the signed-in YouTube session that suppresses the bot check. # Env-switchable so a bot-check regression can be reverted without a redeploy. EMBED_HOSTS = ( "https://www.youtube-nocookie.com", "https://www.youtube.com", ) def __post_init__(self): if not self.base_url: self.base_url = os.environ.get("BRG_BASE_URL", "") if not self.embed_host: self.embed_host = os.environ.get("BRG_EMBED_HOST", "") if self.embed_host not in self.EMBED_HOSTS: self.embed_host = self.EMBED_HOSTS[0] @dataclass class TelegramConfig: """Telegram bot configuration.""" bot_token: str = "" admin_chat_id: str = "" api_url: str = "" # custom Bot API server root, e.g. https://tg-relay.example.com def __post_init__(self): if not self.api_url: self.api_url = os.environ.get("BRG_TELEGRAM_API_URL", "") self.api_url = str(self.api_url).strip().rstrip("/") if self.api_url: parts = urlparse(self.api_url) if parts.scheme not in ("http", "https") or not parts.hostname: raise ValueError("telegram.api_url / BRG_TELEGRAM_API_URL must be an http(s) URL with a host") if parts.scheme == "http": logger.warning("telegram.api_url uses plain http — the bot token is sent unencrypted") @dataclass class YouTubeConfig: """YouTube API configuration.""" search_max_results: int = 50 channel_cache_results: int = 200 # max videos per channel in cache channel_cache_ttl: int = 1800 # seconds between channel cache refreshes ydl_timeout: int = 30 # seconds — max wall-clock time for a single yt-dlp operation shorts_enabled: bool = False # enable Shorts row on homepage @dataclass class DatabaseConfig: """Database configuration.""" path: str = "db/videos.db" @dataclass class WatchLimitsConfig: """Watch time limits configuration.""" daily_limit_minutes: int = 0 timezone: str = "America/New_York" notify_on_limit: bool = True @dataclass class Config: """Main configuration container.""" app: AppConfig = field(default_factory=AppConfig) web: WebConfig = field(default_factory=WebConfig) telegram: TelegramConfig = field(default_factory=TelegramConfig) youtube: YouTubeConfig = field(default_factory=YouTubeConfig) database: DatabaseConfig = field(default_factory=DatabaseConfig) watch_limits: WatchLimitsConfig = field(default_factory=WatchLimitsConfig) @classmethod def from_yaml(cls, path: Path | str) -> "Config": """Load configuration from YAML file with environment variable expansion.""" path = Path(path) with open(path, "r") as f: raw_config = yaml.safe_load(f) # Expand environment variables expanded_config = expand_env_vars(raw_config) # Construct Config from expanded data app_data = expanded_config.get("app", {}) web_data = expanded_config.get("web", {}) telegram_data = expanded_config.get("telegram", {}) youtube_data = expanded_config.get("youtube", {}) database_data = expanded_config.get("database", {}) watch_limits_data = expanded_config.get("watch_limits", {}) return cls( app=AppConfig(**app_data), web=WebConfig(**web_data), telegram=TelegramConfig(**telegram_data), youtube=YouTubeConfig(**youtube_data), database=DatabaseConfig(**database_data), watch_limits=WatchLimitsConfig(**watch_limits_data), ) @classmethod def from_env(cls) -> "Config": """Load configuration directly from environment variables.""" return cls( app=AppConfig( locale=os.environ.get("BRG_LOCALE", "en"), time_format=os.environ.get("BRG_TIME_FORMAT", "locale"), log_level=os.environ.get("BRG_LOG_LEVEL", "info"), ), web=WebConfig( host=os.environ.get("BRG_WEB_HOST", "0.0.0.0"), port=int(os.environ.get("BRG_WEB_PORT", "8080")), poll_interval=int(os.environ.get("BRG_POLL_INTERVAL", "3000")), pin=os.environ.get("BRG_PIN", ""), session_secret=os.environ.get("BRG_SESSION_SECRET", ""), base_url=os.environ.get("BRG_BASE_URL", ""), embed_host=os.environ.get("BRG_EMBED_HOST", ""), ), telegram=TelegramConfig( bot_token=os.environ.get("BRG_BOT_TOKEN", ""), admin_chat_id=os.environ.get("BRG_ADMIN_CHAT_ID", ""), api_url=os.environ.get("BRG_TELEGRAM_API_URL", ""), ), youtube=YouTubeConfig( search_max_results=int(os.environ.get("BRG_YOUTUBE_MAX_RESULTS", "50")), channel_cache_results=int(os.environ.get("BRG_CHANNEL_CACHE_RESULTS", "200")), channel_cache_ttl=int(os.environ.get("BRG_CHANNEL_CACHE_TTL", "1800")), ydl_timeout=int(os.environ.get("BRG_YDL_TIMEOUT", "30")), shorts_enabled=os.environ.get("BRG_SHORTS_ENABLED", "false").lower() == "true", ), database=DatabaseConfig( path=os.environ.get("BRG_DB_PATH", "db/videos.db"), ), watch_limits=WatchLimitsConfig( daily_limit_minutes=int(os.environ.get("BRG_DAILY_LIMIT_MINUTES", "0")), timezone=os.environ.get("BRG_TIMEZONE", "America/New_York"), notify_on_limit=os.environ.get("BRG_NOTIFY_ON_LIMIT", "true").lower() == "true", ), ) def load_config(config_path: str | None = None) -> Config: """Load configuration from file or environment. Tries in order: 1. Provided config_path 2. Default paths: config.yaml, config.yml 3. Environment variables (fallback) """ config: Config | None = None if config_path: path = Path(config_path) if path.exists(): config = Config.from_yaml(path) else: raise FileNotFoundError(f"Config file not found: {config_path}") else: # Try default paths for default_path in ["config.yaml", "config.yml"]: path = Path(default_path) if path.exists(): config = Config.from_yaml(path) break if config is None: # Fallback to environment variables config = Config.from_env() from i18n import normalize_locale, normalize_time_format config.app.locale = normalize_locale(config.app.locale) config.app.time_format = normalize_time_format(config.app.time_format) # Validate admin_chat_id admin_id = config.telegram.admin_chat_id if not admin_id: logger.warning("telegram.admin_chat_id is empty — bot commands will be unauthorized") elif not admin_id.lstrip("-").isdigit(): logger.warning("telegram.admin_chat_id %r is not numeric — admin checks will fail", admin_id) # Validate timezone at startup tz = config.watch_limits.timezone if tz: try: from zoneinfo import ZoneInfo ZoneInfo(tz) except Exception: logger.warning("Invalid timezone %r in config, falling back to UTC", tz) config.watch_limits.timezone = "" return config