claudi
eab1009d8c
- Updated `brand_config.py` to include `WEBDROP_UPDATE_CHANNEL` in the environment variables. - Enhanced `build_macos.sh` to create a bundled `.env` file with brand-specific defaults, including the update channel. - Implemented a method in `build_windows.py` to create a bundled `.env` file for Windows builds, incorporating brand-specific runtime defaults. - Modified `config.py` to ensure the application can locate the `.env` file in various installation scenarios. - Added unit tests in `test_config.py` to verify the loading of the bootstrap `.env` from the PyInstaller runtime directory. - Generated new WiX object and script files for the Windows installer, including application shortcuts and registry entries.
502 lines
20 KiB
Python
502 lines
20 KiB
Python
"""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):
|
|
exe_dir = Path(sys.executable).resolve().parent
|
|
# One-folder fallback: some packagers place data files in _internal.
|
|
candidate_paths.append(exe_dir / ".env")
|
|
candidate_paths.append(exe_dir / "_internal" / ".env")
|
|
|
|
# PyInstaller runtime extraction directory (one-file and one-folder).
|
|
meipass = getattr(sys, "_MEIPASS", None)
|
|
if meipass:
|
|
candidate_paths.append(Path(meipass).resolve() / ".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})"
|
|
)
|