"""Centralized configuration — two-layer system (ENV bootstrap + DB runtime). On first start, ENV values seed the database. After that, the database is the source of truth for all integration configs. ENV is only needed for: DB connection, ADMIN_PASSWORD, JWT_SECRET. """ from __future__ import annotations import json import logging import os from dataclasses import dataclass, field from typing import Any, Dict, List logger = logging.getLogger(__name__) @dataclass class UnraidServer: name: str host: str api_key: str = "" port: int = 80 @dataclass class Settings: # --- Bootstrap (always from ENV) --- db_host: str = "10.10.10.10" db_port: int = 5433 db_name: str = "openclaw" db_user: str = "sam" db_password: str = "sam" # --- Weather --- weather_location: str = "Leverkusen" weather_location_secondary: str = "Rab,Croatia" weather_cache_ttl: int = 1800 # --- Home Assistant --- ha_url: str = "" ha_token: str = "" ha_cache_ttl: int = 30 ha_enabled: bool = False # --- Vikunja Tasks --- vikunja_url: str = "" vikunja_token: str = "" vikunja_cache_ttl: int = 60 vikunja_enabled: bool = False vikunja_private_projects: List[int] = field(default_factory=lambda: [3, 4]) vikunja_sams_projects: List[int] = field(default_factory=lambda: [2, 5]) # --- Unraid Servers --- unraid_servers: List[UnraidServer] = field(default_factory=list) unraid_cache_ttl: int = 15 unraid_enabled: bool = False # --- News --- news_cache_ttl: int = 300 news_max_age_hours: int = 48 news_enabled: bool = True # --- MQTT --- mqtt_host: str = "" mqtt_port: int = 1883 mqtt_username: str = "" mqtt_password: str = "" mqtt_topics: List[str] = field(default_factory=lambda: ["#"]) mqtt_client_id: str = "daily-briefing" mqtt_enabled: bool = False # --- Server --- host: str = "0.0.0.0" port: int = 8080 debug: bool = False # --- WebSocket --- ws_interval: int = 15 @classmethod def from_env(cls) -> "Settings": """Load bootstrap config from environment variables.""" s = cls() s.db_host = os.getenv("DB_HOST", s.db_host) s.db_port = int(os.getenv("DB_PORT", str(s.db_port))) s.db_name = os.getenv("DB_NAME", s.db_name) s.db_user = os.getenv("DB_USER", s.db_user) s.db_password = os.getenv("DB_PASSWORD", s.db_password) s.debug = os.getenv("DEBUG", "").lower() in ("1", "true", "yes") # Legacy ENV support — used for first-run seeding s.weather_location = os.getenv("WEATHER_LOCATION", s.weather_location) s.weather_location_secondary = os.getenv("WEATHER_LOCATION_SECONDARY", s.weather_location_secondary) s.ha_url = os.getenv("HA_URL", s.ha_url) s.ha_token = os.getenv("HA_TOKEN", s.ha_token) s.ha_enabled = bool(s.ha_url) s.vikunja_url = os.getenv("VIKUNJA_URL", s.vikunja_url) s.vikunja_token = os.getenv("VIKUNJA_TOKEN", s.vikunja_token) s.vikunja_enabled = bool(s.vikunja_url) s.mqtt_host = os.getenv("MQTT_HOST", s.mqtt_host) s.mqtt_port = int(os.getenv("MQTT_PORT", str(s.mqtt_port))) s.mqtt_username = os.getenv("MQTT_USERNAME", s.mqtt_username) s.mqtt_password = os.getenv("MQTT_PASSWORD", s.mqtt_password) s.mqtt_client_id = os.getenv("MQTT_CLIENT_ID", s.mqtt_client_id) s.mqtt_enabled = bool(s.mqtt_host) # Parse MQTT_TOPICS raw_topics = os.getenv("MQTT_TOPICS", "") if raw_topics: try: s.mqtt_topics = json.loads(raw_topics) except (json.JSONDecodeError, TypeError): s.mqtt_topics = [t.strip() for t in raw_topics.split(",") if t.strip()] # Parse UNRAID_SERVERS JSON raw = os.getenv("UNRAID_SERVERS", "[]") try: servers_data = json.loads(raw) s.unraid_servers = [ UnraidServer( name=srv.get("name", f"Server {i+1}"), host=srv.get("host", ""), api_key=srv.get("api_key", ""), port=int(srv.get("port", 80)), ) for i, srv in enumerate(servers_data) if srv.get("host") ] s.unraid_enabled = len(s.unraid_servers) > 0 except (json.JSONDecodeError, TypeError): s.unraid_servers = [] return s async def load_from_db(self) -> None: """Override fields with values from the database.""" try: from server.services import settings_service # Load integrations integrations = await settings_service.get_integrations() for integ in integrations: cfg = integ.get("config", {}) enabled = integ.get("enabled", True) itype = integ["type"] if itype == "weather": self.weather_location = cfg.get("location", self.weather_location) self.weather_location_secondary = cfg.get("location_secondary", self.weather_location_secondary) elif itype == "news": self.news_max_age_hours = int(cfg.get("max_age_hours", self.news_max_age_hours)) self.news_enabled = enabled elif itype == "ha": self.ha_url = cfg.get("url", self.ha_url) self.ha_token = cfg.get("token", self.ha_token) self.ha_enabled = enabled elif itype == "vikunja": self.vikunja_url = cfg.get("url", self.vikunja_url) self.vikunja_token = cfg.get("token", self.vikunja_token) self.vikunja_private_projects = cfg.get("private_projects", self.vikunja_private_projects) self.vikunja_sams_projects = cfg.get("sams_projects", self.vikunja_sams_projects) self.vikunja_enabled = enabled elif itype == "unraid": servers = cfg.get("servers", []) self.unraid_servers = [ UnraidServer( name=s.get("name", ""), host=s.get("host", ""), api_key=s.get("api_key", ""), port=int(s.get("port", 80)), ) for s in servers if s.get("host") ] self.unraid_enabled = enabled elif itype == "mqtt": self.mqtt_host = cfg.get("host", self.mqtt_host) self.mqtt_port = int(cfg.get("port", self.mqtt_port)) self.mqtt_username = cfg.get("username", self.mqtt_username) self.mqtt_password = cfg.get("password", self.mqtt_password) self.mqtt_client_id = cfg.get("client_id", self.mqtt_client_id) self.mqtt_topics = cfg.get("topics", self.mqtt_topics) self.mqtt_enabled = enabled # Load app_settings (cache TTLs, etc.) all_settings = await settings_service.get_all_settings() for key, data in all_settings.items(): val = data["value"] if key == "weather_cache_ttl": self.weather_cache_ttl = int(val) elif key == "ha_cache_ttl": self.ha_cache_ttl = int(val) elif key == "vikunja_cache_ttl": self.vikunja_cache_ttl = int(val) elif key == "unraid_cache_ttl": self.unraid_cache_ttl = int(val) elif key == "news_cache_ttl": self.news_cache_ttl = int(val) elif key == "ws_interval": self.ws_interval = int(val) logger.info("Settings loaded from database (%d integrations)", len(integrations)) except Exception: logger.exception("Failed to load settings from DB — using ENV defaults") # --- Module-level singleton --- settings = Settings.from_env() def get_settings() -> Settings: """Return the current settings. Used by all routers.""" return settings async def reload_settings() -> None: """Reload settings from DB. Called after admin changes.""" global settings s = Settings.from_env() await s.load_from_db() settings = s logger.info("Settings reloaded from database")