"""Configuration management for WebDrop Bridge application.""" import json import logging import os import sys from dataclasses import dataclass, field from pathlib import Path from typing import List from dotenv import load_dotenv logger = logging.getLogger(__name__) DEFAULT_BRAND_ID = "webdrop_bridge" DEFAULT_CONFIG_DIR_NAME = "webdrop_bridge" DEFAULT_UPDATE_BASE_URL = "https://git.him-tools.de" DEFAULT_UPDATE_REPO = "HIM-public/webdrop-bridge" DEFAULT_UPDATE_CHANNEL = "stable" DEFAULT_UPDATE_MANIFEST_NAME = "release-manifest.json" class ConfigurationError(Exception): """Raised when configuration is invalid.""" pass @dataclass class URLMapping: """Maps an Azure Blob Storage URL prefix to a local drive path.""" url_prefix: str local_path: str def __post_init__(self): """Validate mapping configuration.""" if not self.url_prefix.startswith(("http://", "https://")): raise ConfigurationError( f"URL prefix must start with http:// or https://: {self.url_prefix}" ) # Ensure URL prefix ends with / if not self.url_prefix.endswith("/"): self.url_prefix += "/" # Normalize local path self.local_path = str(Path(self.local_path)) @dataclass class Config: """Application configuration loaded from environment variables or JSON file. Attributes: app_name: Application display name app_version: Application version (semantic versioning) log_level: Logging level (DEBUG, INFO, WARNING, ERROR) log_file: Optional log file path allowed_roots: List of whitelisted root directories for file access allowed_urls: List of whitelisted URL domains/patterns (empty = no restriction) webapp_url: URL to load in embedded web application (default: https://dev.agravity.io/) url_mappings: List of Azure URL to local path mappings check_file_exists: Whether to validate that files exist before drag auto_check_updates: Whether to automatically check for updates update_check_interval_hours: Hours between update checks window_width: Initial window width in pixels window_height: Initial window height in pixels window_title: Main window title (default: "{app_name} v{app_version}") enable_logging: Whether to write logs to file enable_checkout: Whether to check asset checkout status and show checkout dialog on drag. Disabled by default as checkout support is optional. brand_id: Stable brand identifier used for packaging and update selection config_dir_name: AppData/config directory name for this branded variant update_base_url: Base Forgejo URL used for release checks update_repo: Forgejo repository containing shared releases update_channel: Update channel name used by release manifest selection update_manifest_name: Asset name of the shared release manifest Raises: ConfigurationError: If configuration values are invalid """ app_name: str app_version: str log_level: str log_file: Path | None allowed_roots: List[Path] allowed_urls: List[str] webapp_url: str url_mappings: List[URLMapping] = field(default_factory=list) check_file_exists: bool = True auto_check_updates: bool = True update_check_interval_hours: int = 24 window_width: int = 1024 window_height: int = 768 window_title: str = "" enable_logging: bool = True enable_checkout: bool = False language: str = "auto" brand_id: str = DEFAULT_BRAND_ID config_dir_name: str = DEFAULT_CONFIG_DIR_NAME update_base_url: str = DEFAULT_UPDATE_BASE_URL update_repo: str = DEFAULT_UPDATE_REPO update_channel: str = DEFAULT_UPDATE_CHANNEL update_manifest_name: str = DEFAULT_UPDATE_MANIFEST_NAME @classmethod def from_file(cls, config_path: Path) -> "Config": """Load configuration from JSON file. Args: config_path: Path to configuration file Returns: Config: Configured instance from JSON file Raises: ConfigurationError: If configuration file is invalid """ if not config_path.exists(): raise ConfigurationError(f"Configuration file not found: {config_path}") try: with open(config_path, "r", encoding="utf-8") as f: data = json.load(f) except (json.JSONDecodeError, IOError) as e: raise ConfigurationError(f"Failed to load configuration: {e}") from e # Get version from package from webdrop_bridge import __version__ # Parse URL mappings mappings = [ URLMapping(url_prefix=m["url_prefix"], local_path=m["local_path"]) for m in data.get("url_mappings", []) ] # Parse allowed roots allowed_roots = [Path(p).resolve() for p in data.get("allowed_roots", [])] # Validate allowed roots exist for root in allowed_roots: if not root.exists(): logger.warning(f"Allowed root does not exist: {root}") elif not root.is_dir(): raise ConfigurationError(f"Allowed root is not a directory: {root}") brand_id = data.get("brand_id", DEFAULT_BRAND_ID) config_dir_name = data.get("config_dir_name", cls._slugify_config_dir_name(brand_id)) # Get log file path log_file = None if data.get("enable_logging", True): log_file_str = data.get("log_file", None) if log_file_str: log_file = Path(log_file_str) # If relative path, resolve relative to app data directory instead of cwd if not log_file.is_absolute(): log_file = Config.get_default_log_dir(config_dir_name) / log_file else: # Use default log path in app data log_file = Config.get_default_log_path(config_dir_name) app_name = data.get("app_name", "WebDrop Bridge") stored_window_title = data.get("window_title", "") # Regenerate default window titles on version upgrade # If the stored title matches the pattern "{app_name} v{version}", regenerate it # with the current version. This ensures the title updates automatically on upgrades. import re version_pattern = re.compile(rf"^{re.escape(app_name)}\s+v[\d.]+$") if stored_window_title and version_pattern.match(stored_window_title): # Detected a default-pattern title with old version, regenerate window_title = f"{app_name} v{__version__}" elif stored_window_title: # Custom window title, keep it as-is window_title = stored_window_title else: # No window title specified, use default window_title = f"{app_name} v{__version__}" return cls( app_name=app_name, app_version=__version__, log_level=data.get("log_level", "INFO").upper(), log_file=log_file, allowed_roots=allowed_roots, allowed_urls=data.get("allowed_urls", []), webapp_url=data.get("webapp_url", "https://dev.agravity.io/"), url_mappings=mappings, check_file_exists=data.get("check_file_exists", True), auto_check_updates=data.get("auto_check_updates", True), update_check_interval_hours=data.get("update_check_interval_hours", 24), window_width=data.get("window_width", 1024), window_height=data.get("window_height", 768), window_title=window_title, enable_logging=data.get("enable_logging", True), enable_checkout=data.get("enable_checkout", False), language=data.get("language", "auto"), brand_id=brand_id, config_dir_name=config_dir_name, update_base_url=data.get("update_base_url", DEFAULT_UPDATE_BASE_URL), update_repo=data.get("update_repo", DEFAULT_UPDATE_REPO), update_channel=data.get("update_channel", DEFAULT_UPDATE_CHANNEL), update_manifest_name=data.get("update_manifest_name", DEFAULT_UPDATE_MANIFEST_NAME), ) @classmethod def from_env(cls, env_file: str | None = None) -> "Config": """Load configuration from environment variables. Looks for a .env file in the current working directory and loads environment variables from it. All unset variables use sensible defaults. Args: env_file: Optional path to .env file to load. If None, loads default .env Returns: Config: Configured instance with environment values or defaults Raises: ConfigurationError: If any configured value is invalid """ # Load .env file if it exists load_dotenv(env_file) # Extract and validate configuration values app_name = os.getenv("APP_NAME", "WebDrop Bridge") # Version always comes from __init__.py for consistency from webdrop_bridge import __version__ app_version = __version__ brand_id = os.getenv("BRAND_ID", DEFAULT_BRAND_ID) config_dir_name = os.getenv("APP_CONFIG_DIR_NAME", cls._slugify_config_dir_name(brand_id)) log_level = os.getenv("LOG_LEVEL", "INFO").upper() log_file_str = os.getenv("LOG_FILE", None) allowed_roots_str = os.getenv("ALLOWED_ROOTS", "Z:/,C:/Users/Public") allowed_urls_str = os.getenv("ALLOWED_URLS", "") webapp_url = os.getenv("WEBAPP_URL", "https://dev.agravity.io/") window_width = int(os.getenv("WINDOW_WIDTH", "1024")) window_height = int(os.getenv("WINDOW_HEIGHT", "768")) # Window title defaults to app_name + version if not specified default_title = f"{app_name} v{app_version}" window_title = os.getenv("WINDOW_TITLE", default_title) enable_logging = os.getenv("ENABLE_LOGGING", "true").lower() == "true" enable_checkout = os.getenv("ENABLE_CHECKOUT", "false").lower() == "true" language = os.getenv("LANGUAGE", "auto") update_base_url = os.getenv("UPDATE_BASE_URL", DEFAULT_UPDATE_BASE_URL) update_repo = os.getenv("UPDATE_REPO", DEFAULT_UPDATE_REPO) update_channel = os.getenv("UPDATE_CHANNEL", DEFAULT_UPDATE_CHANNEL) update_manifest_name = os.getenv("UPDATE_MANIFEST_NAME", DEFAULT_UPDATE_MANIFEST_NAME) # Validate log level valid_levels = {"DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"} if log_level not in valid_levels: raise ConfigurationError( f"Invalid LOG_LEVEL: {log_level}. " f"Must be one of: {', '.join(valid_levels)}" ) # Validate and parse allowed roots try: allowed_roots = [] for p in allowed_roots_str.split(","): root_path = Path(p.strip()).resolve() if not root_path.exists(): logger.warning(f"Allowed root does not exist: {p.strip()}") elif not root_path.is_dir(): raise ConfigurationError(f"Allowed root '{p.strip()}' is not a directory") else: allowed_roots.append(root_path) except ConfigurationError: raise except (ValueError, OSError) as e: raise ConfigurationError( f"Invalid ALLOWED_ROOTS: {allowed_roots_str}. Error: {e}" ) from e # Validate window dimensions if window_width <= 0 or window_height <= 0: raise ConfigurationError( f"Window dimensions must be positive: " f"{window_width}x{window_height}" ) # Create log file path if logging enabled log_file = None if enable_logging: if log_file_str: log_file = Path(log_file_str) # If relative path, resolve relative to app data directory instead of cwd if not log_file.is_absolute(): log_file = Config.get_default_log_dir(config_dir_name) / log_file else: # Use default log path in app data log_file = Config.get_default_log_path(config_dir_name) # Validate webapp URL is not empty if not webapp_url: raise ConfigurationError("WEBAPP_URL cannot be empty") # Parse allowed URLs (empty string = no restriction) allowed_urls = ( [url.strip() for url in allowed_urls_str.split(",") if url.strip()] if allowed_urls_str else [] ) # Parse URL mappings (Azure Blob Storage → Local Paths) # Format: url_prefix1=local_path1;url_prefix2=local_path2 url_mappings_str = os.getenv("URL_MAPPINGS", "") url_mappings = [] if url_mappings_str: try: for mapping in url_mappings_str.split(";"): mapping = mapping.strip() if not mapping: continue if "=" not in mapping: raise ConfigurationError( f"Invalid URL mapping format: {mapping}. Expected 'url=path'" ) url_prefix, local_path_str = mapping.split("=", 1) url_mappings.append( URLMapping(url_prefix=url_prefix.strip(), local_path=local_path_str.strip()) ) except (ValueError, OSError) as e: raise ConfigurationError( f"Invalid URL_MAPPINGS: {url_mappings_str}. Error: {e}" ) from e return cls( app_name=app_name, app_version=app_version, log_level=log_level, log_file=log_file, allowed_roots=allowed_roots, allowed_urls=allowed_urls, webapp_url=webapp_url, url_mappings=url_mappings, window_width=window_width, window_height=window_height, window_title=window_title, enable_logging=enable_logging, enable_checkout=enable_checkout, language=language, brand_id=brand_id, config_dir_name=config_dir_name, update_base_url=update_base_url, update_repo=update_repo, update_channel=update_channel, update_manifest_name=update_manifest_name, ) def to_file(self, config_path: Path) -> None: """Save configuration to JSON file. Args: config_path: Path to save configuration to Creates parent directories if they don't exist. """ data = { "app_name": self.app_name, "webapp_url": self.webapp_url, "url_mappings": [ {"url_prefix": m.url_prefix, "local_path": m.local_path} for m in self.url_mappings ], "allowed_roots": [str(p) for p in self.allowed_roots], "allowed_urls": self.allowed_urls, "check_file_exists": self.check_file_exists, "auto_check_updates": self.auto_check_updates, "update_check_interval_hours": self.update_check_interval_hours, "log_level": self.log_level, "log_file": str(self.log_file) if self.log_file else None, "window_width": self.window_width, "window_height": self.window_height, "window_title": self.window_title, "enable_logging": self.enable_logging, "enable_checkout": self.enable_checkout, "language": self.language, "brand_id": self.brand_id, "config_dir_name": self.config_dir_name, "update_base_url": self.update_base_url, "update_repo": self.update_repo, "update_channel": self.update_channel, "update_manifest_name": self.update_manifest_name, } config_path.parent.mkdir(parents=True, exist_ok=True) with open(config_path, "w", encoding="utf-8") as f: json.dump(data, f, indent=2) @staticmethod def load_bootstrap_env(env_file: str | None = None) -> Path | None: """Load a bootstrap .env before configuration path lookup. This lets branded builds decide their config directory before the main config file is loaded. Args: env_file: Optional explicit .env path Returns: Path to the loaded .env file, or None if nothing was loaded """ candidate_paths: list[Path] = [] if env_file: candidate_paths.append(Path(env_file).resolve()) else: if getattr(sys, "frozen", False): candidate_paths.append(Path(sys.executable).resolve().parent / ".env") candidate_paths.append(Path.cwd() / ".env") candidate_paths.append(Path(__file__).resolve().parents[2] / ".env") for path in candidate_paths: if path.exists(): load_dotenv(path, override=False) logger.debug(f"Loaded bootstrap environment from {path}") return path return None @staticmethod def _slugify_config_dir_name(value: str) -> str: """Convert brand-like identifiers into a filesystem-safe directory name.""" sanitized = "".join(c.lower() if c.isalnum() else "_" for c in value).strip("_") return sanitized or DEFAULT_CONFIG_DIR_NAME @staticmethod def get_default_config_dir_name() -> str: """Get the default config directory name from environment or fallback.""" return os.getenv("APP_CONFIG_DIR_NAME", DEFAULT_CONFIG_DIR_NAME) @staticmethod def get_default_config_path(config_dir_name: str | None = None) -> Path: """Get the default configuration file path. Returns: Path to default config file in user's AppData/Roaming """ import platform if platform.system() == "Windows": base = Path.home() / "AppData" / "Roaming" else: base = Path.home() / ".config" return base / (config_dir_name or Config.get_default_config_dir_name()) / "config.json" @staticmethod def get_default_log_dir(config_dir_name: str | None = None) -> Path: """Get the default directory for log files. Always uses user's AppData directory to ensure permissions work correctly in both development and installed scenarios. Returns: Path to default logs directory in user's AppData/Roaming """ import platform if platform.system() == "Windows": base = Path.home() / "AppData" / "Roaming" else: base = Path.home() / ".local" / "share" return base / (config_dir_name or Config.get_default_config_dir_name()) / "logs" @staticmethod def get_default_log_path(config_dir_name: str | None = None) -> Path: """Get the default log file path. Returns: Path to default log file in user's AppData/Roaming/webdrop_bridge/logs """ dir_name = config_dir_name or Config.get_default_config_dir_name() return Config.get_default_log_dir(dir_name) / f"{dir_name}.log" def get_config_path(self) -> Path: """Get the default config file path for this configured brand.""" return self.get_default_config_path(self.config_dir_name) def get_cache_dir(self) -> Path: """Get the update/cache directory for this configured brand.""" return self.get_default_config_path(self.config_dir_name).parent / "cache" def __repr__(self) -> str: """Return developer-friendly representation.""" return ( f"Config(app={self.app_name} v{self.app_version}, " f"brand={self.brand_id}, " f"log_level={self.log_level}, " f"allowed_roots={len(self.allowed_roots)} dirs, " f"window={self.window_width}x{self.window_height})" )