claudi
88dc358894
- Renamed `initiate_drag` to `handle_drag` in MainWindow and updated related tests. - Improved drag handling logic to utilize a bridge for starting file drags. - Updated `_on_drag_started` and `_on_drag_failed` methods to match new signatures. - Modified test cases to reflect changes in drag handling and assertions. Enhance path validation and logging - Updated `PathValidator` to log warnings for nonexistent roots instead of raising errors. - Adjusted tests to verify the new behavior of skipping nonexistent roots. Update web application UI and functionality - Changed displayed text for drag items to reflect local paths and Azure Blob Storage URLs. - Added debug logging for drag operations in the web application. - Improved instructions for testing drag and drop functionality. Add configuration documentation and example files - Created `CONFIG_README.md` to provide detailed configuration instructions for WebDrop Bridge. - Added `config.example.json` and `config_test.json` for reference and testing purposes. Implement URL conversion logic - Introduced `URLConverter` class to handle conversion of Azure Blob Storage URLs to local paths. - Added unit tests for URL conversion to ensure correct functionality. Develop download interceptor script - Created `download_interceptor.js` to intercept download-related actions in the web application. - Implemented logging for fetch calls, XMLHttpRequests, and Blob URL creations. Add download test page and related tests - Created `test_download.html` for testing various download scenarios. - Implemented `test_download.py` to verify download path resolution and file construction. - Added `test_url_mappings.py` to ensure URL mappings are loaded correctly. Add unit tests for URL converter - Created `test_url_converter.py` to validate URL conversion logic and mapping behavior.
337 lines
12 KiB
Python
337 lines
12 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://wps.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
|
|
|
|
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
|
|
|
|
@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", "logs/webdrop_bridge.log")
|
|
log_file = Path(log_file_str).resolve()
|
|
|
|
app_name = data.get("app_name", "WebDrop Bridge")
|
|
window_title = data.get("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://wps.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),
|
|
)
|
|
|
|
@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 comes from __init__.py (lazy import to avoid circular imports)
|
|
if not os.getenv("APP_VERSION"):
|
|
from webdrop_bridge import __version__
|
|
app_version = __version__
|
|
else:
|
|
app_version = os.getenv("APP_VERSION")
|
|
log_level = os.getenv("LOG_LEVEL", "INFO").upper()
|
|
log_file_str = os.getenv("LOG_FILE", "logs/webdrop_bridge.log")
|
|
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://wps.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"
|
|
|
|
# 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:
|
|
log_file = Path(log_file_str).resolve()
|
|
|
|
# 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,
|
|
)
|
|
|
|
def to_file(self, config_path: Path) -> None:
|
|
"""Save configuration to JSON file.
|
|
|
|
Args:
|
|
config_path: Path to save configuration
|
|
"""
|
|
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,
|
|
}
|
|
|
|
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
|
|
"""
|
|
import platform
|
|
if platform.system() == "Windows":
|
|
base = Path.home() / "AppData" / "Roaming"
|
|
else:
|
|
base = Path.home() / ".config"
|
|
return base / "webdrop_bridge" / "config.json"
|
|
|
|
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})"
|
|
)
|