"""Configuration management for WebDrop Bridge application.""" import json import logging import os from dataclasses import dataclass, field from pathlib import Path from typing import List from dotenv import load_dotenv logger = logging.getLogger(__name__) 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. 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 @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}") # 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() / log_file else: # Use default log path in app data log_file = Config.get_default_log_path() 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), ) @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__ 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" # 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() / log_file else: # Use default log path in app data log_file = Config.get_default_log_path() # 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, ) 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, } 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 get_default_config_path() -> 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 / "webdrop_bridge" / "config.json" @staticmethod def get_default_log_dir() -> 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 / "webdrop_bridge" / "logs" @staticmethod def get_default_log_path() -> Path: """Get the default log file path. Returns: Path to default log file in user's AppData/Roaming/webdrop_bridge/logs """ return Config.get_default_log_dir() / "webdrop_bridge.log" def __repr__(self) -> str: """Return developer-friendly representation.""" return ( f"Config(app={self.app_name} v{self.app_version}, " f"log_level={self.log_level}, " f"allowed_roots={len(self.allowed_roots)} dirs, " f"window={self.window_width}x{self.window_height})" )