390 lines
15 KiB
Python
390 lines
15 KiB
Python
"""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})"
|
|
)
|