Missing config submodule in the repository

parent e9ca7272
......@@ -145,7 +145,6 @@ ehthumbs.db
Thumbs.db
# Project specific
config/
logs/
videos/
*.db
......
"""
Configuration management for MbetterClient
"""
from .settings import AppSettings, DatabaseConfig, WebConfig, QtConfig, ApiConfig
from .manager import ConfigManager
__all__ = [
'AppSettings',
'DatabaseConfig',
'WebConfig',
'QtConfig',
'ApiConfig',
'ConfigManager'
]
\ No newline at end of file
"""
Configuration management with database persistence and validation
"""
import json
import logging
from typing import Dict, Any, Optional
from pathlib import Path
from .settings import AppSettings
from ..database.models import ConfigurationModel
from ..database.manager import DatabaseManager
logger = logging.getLogger(__name__)
class ConfigManager:
"""Manages application configuration with database persistence"""
def __init__(self, db_manager: DatabaseManager):
self.db_manager = db_manager
self._settings: Optional[AppSettings] = None
self._config_cache: Dict[str, Any] = {}
def initialize(self) -> bool:
"""Initialize configuration manager"""
try:
# Load settings from database
self._settings = self.load_settings()
if self._settings is None:
# Create default settings
self._settings = AppSettings()
self.save_settings(self._settings)
# Validate settings
if not self._settings.validate():
logger.error("Configuration validation failed")
return False
# Ensure directories exist
self._settings.ensure_directories()
logger.info("Configuration manager initialized")
return True
except Exception as e:
logger.error(f"Failed to initialize configuration manager: {e}")
return False
def get_settings(self) -> AppSettings:
"""Get current application settings"""
if self._settings is None:
raise RuntimeError("Configuration manager not initialized")
return self._settings
def update_settings(self, settings: AppSettings) -> bool:
"""Update application settings"""
try:
# Validate new settings
if not settings.validate():
logger.error("Settings validation failed")
return False
# Save to database
if self.save_settings(settings):
self._settings = settings
logger.info("Settings updated successfully")
return True
else:
logger.error("Failed to save settings")
return False
except Exception as e:
logger.error(f"Failed to update settings: {e}")
return False
def load_settings(self) -> Optional[AppSettings]:
"""Load settings from database"""
try:
config_data = self.db_manager.get_configuration()
if config_data:
return AppSettings.from_dict(config_data)
return None
except Exception as e:
logger.error(f"Failed to load settings: {e}")
return None
def save_settings(self, settings: AppSettings) -> bool:
"""Save settings to database"""
try:
config_data = settings.to_dict()
return self.db_manager.save_configuration(config_data)
except Exception as e:
logger.error(f"Failed to save settings: {e}")
return False
def get_config_value(self, key: str, default: Any = None) -> Any:
"""Get a specific configuration value"""
try:
# Check cache first
if key in self._config_cache:
return self._config_cache[key]
# Load from database
value = self.db_manager.get_config_value(key, default)
self._config_cache[key] = value
return value
except Exception as e:
logger.error(f"Failed to get config value '{key}': {e}")
return default
def set_config_value(self, key: str, value: Any) -> bool:
"""Set a specific configuration value"""
try:
# Save to database
if self.db_manager.set_config_value(key, value):
# Update cache
self._config_cache[key] = value
logger.debug(f"Config value '{key}' updated")
return True
else:
logger.error(f"Failed to set config value '{key}'")
return False
except Exception as e:
logger.error(f"Failed to set config value '{key}': {e}")
return False
def delete_config_value(self, key: str) -> bool:
"""Delete a configuration value"""
try:
if self.db_manager.delete_config_value(key):
# Remove from cache
self._config_cache.pop(key, None)
logger.debug(f"Config value '{key}' deleted")
return True
else:
logger.error(f"Failed to delete config value '{key}'")
return False
except Exception as e:
logger.error(f"Failed to delete config value '{key}': {e}")
return False
def get_all_config_values(self) -> Dict[str, Any]:
"""Get all configuration values"""
try:
return self.db_manager.get_all_config_values()
except Exception as e:
logger.error(f"Failed to get all config values: {e}")
return {}
def export_config(self, file_path: str) -> bool:
"""Export configuration to JSON file"""
try:
if self._settings is None:
logger.error("No settings to export")
return False
config_data = self._settings.to_dict()
# Add individual config values
config_data["custom_values"] = self.get_all_config_values()
with open(file_path, 'w') as f:
json.dump(config_data, f, indent=2, default=str)
logger.info(f"Configuration exported to {file_path}")
return True
except Exception as e:
logger.error(f"Failed to export configuration: {e}")
return False
def import_config(self, file_path: str) -> bool:
"""Import configuration from JSON file"""
try:
if not Path(file_path).exists():
logger.error(f"Config file not found: {file_path}")
return False
with open(file_path, 'r') as f:
config_data = json.load(f)
# Extract custom values
custom_values = config_data.pop("custom_values", {})
# Create settings from data
settings = AppSettings.from_dict(config_data)
# Validate settings
if not settings.validate():
logger.error("Imported configuration is invalid")
return False
# Update settings
if not self.update_settings(settings):
logger.error("Failed to update settings")
return False
# Import custom values
for key, value in custom_values.items():
self.set_config_value(key, value)
logger.info(f"Configuration imported from {file_path}")
return True
except Exception as e:
logger.error(f"Failed to import configuration: {e}")
return False
def reset_to_defaults(self) -> bool:
"""Reset configuration to default values"""
try:
# Create default settings
default_settings = AppSettings()
# Update settings
if self.update_settings(default_settings):
# Clear custom config values
all_keys = list(self.get_all_config_values().keys())
for key in all_keys:
self.delete_config_value(key)
# Clear cache
self._config_cache.clear()
logger.info("Configuration reset to defaults")
return True
else:
logger.error("Failed to reset configuration")
return False
except Exception as e:
logger.error(f"Failed to reset configuration: {e}")
return False
def get_web_config_dict(self) -> Dict[str, Any]:
"""Get configuration data for web dashboard"""
try:
if self._settings is None:
return {}
# Return sanitized config (no sensitive data)
config = self._settings.to_dict()
# Remove sensitive keys
sensitive_keys = ["secret_key", "jwt_secret_key", "token"]
def remove_sensitive(d):
if isinstance(d, dict):
return {k: remove_sensitive(v) for k, v in d.items()
if k not in sensitive_keys}
return d
return remove_sensitive(config)
except Exception as e:
logger.error(f"Failed to get web config: {e}")
return {}
def update_from_web(self, config_data: Dict[str, Any]) -> bool:
"""Update configuration from web dashboard"""
try:
if self._settings is None:
logger.error("Configuration manager not initialized")
return False
# Create new settings with updated data
current_dict = self._settings.to_dict()
# Update with new data (preserve sensitive keys)
def update_dict(target, source):
for key, value in source.items():
if isinstance(value, dict) and key in target:
if isinstance(target[key], dict):
update_dict(target[key], value)
else:
target[key] = value
else:
target[key] = value
update_dict(current_dict, config_data)
# Create new settings object
new_settings = AppSettings.from_dict(current_dict)
# Update settings
return self.update_settings(new_settings)
except Exception as e:
logger.error(f"Failed to update config from web: {e}")
return False
def validate_configuration(self) -> bool:
"""Validate current configuration"""
try:
if self._settings is None:
logger.error("Configuration manager not initialized")
return False
return self._settings.validate()
except Exception as e:
logger.error(f"Configuration validation failed: {e}")
return False
def get_section_config(self, section: str) -> Dict[str, Any]:
"""Get configuration for a specific section"""
try:
if self._settings is None:
logger.error("Configuration manager not initialized")
return {}
config_dict = self._settings.to_dict()
# Return the specific section if it exists
if section in config_dict:
return config_dict[section]
# Handle nested sections (e.g., "api.client")
sections = section.split('.')
current = config_dict
for sec in sections:
if isinstance(current, dict) and sec in current:
current = current[sec]
else:
return {}
return current if isinstance(current, dict) else {}
except Exception as e:
logger.error(f"Failed to get section config '{section}': {e}")
return {}
def update_section(self, section: str, config_data: Dict[str, Any]) -> bool:
"""Update a specific configuration section"""
try:
if self._settings is None:
logger.error("Configuration manager not initialized")
return False
# Get current configuration as dict
current_config = self._settings.to_dict()
# Update the specific section
if section in current_config:
# Direct section update
current_config[section] = config_data
else:
# Handle nested sections (e.g., "api.client")
sections = section.split('.')
current = current_config
# Navigate to the parent of the target section
for i, sec in enumerate(sections[:-1]):
if sec not in current:
current[sec] = {}
current = current[sec]
# Update the target section
current[sections[-1]] = config_data
# Create new settings from updated config
new_settings = AppSettings.from_dict(current_config)
# Update settings
return self.update_settings(new_settings)
except Exception as e:
logger.error(f"Failed to update section '{section}': {e}")
return False
def get_all_config(self) -> Dict[str, Any]:
"""Get all configuration data for web dashboard"""
return self.get_web_config_dict()
\ No newline at end of file
"""
Application settings and configuration classes
"""
import os
import sys
from dataclasses import dataclass, field
from typing import Dict, Any, Optional
from pathlib import Path
def get_user_data_dir() -> Path:
"""Get platform-appropriate user data directory for persistent storage
Cross-platform implementation following OS conventions:
- Windows: %APPDATA%/MbetterClient (e.g., C:/Users/username/AppData/Roaming/MbetterClient)
- macOS: ~/Library/Application Support/MbetterClient
- Linux: ~/.local/share/MbetterClient (XDG Base Directory)
"""
app_name = "MbetterClient"
try:
if sys.platform.startswith("win"):
# Windows: Use APPDATA (Roaming)
appdata = os.environ.get('APPDATA')
if appdata:
base_dir = Path(appdata)
else:
# Fallback for Windows without APPDATA (rare)
base_dir = Path.home() / 'AppData' / 'Roaming'
elif sys.platform == "darwin":
# macOS: Use Application Support
base_dir = Path.home() / 'Library' / 'Application Support'
else:
# Linux/Unix: Use XDG Base Directory specification
xdg_data_home = os.environ.get('XDG_DATA_HOME')
if xdg_data_home:
base_dir = Path(xdg_data_home)
else:
base_dir = Path.home() / '.local' / 'share'
# Ensure base directory exists and is accessible
if not base_dir.exists():
base_dir.mkdir(parents=True, exist_ok=True)
# Create application directory
user_dir = base_dir / app_name
user_dir.mkdir(parents=True, exist_ok=True)
# Verify write permissions
test_file = user_dir / '.write_test'
try:
test_file.write_text('test')
test_file.unlink()
except (OSError, PermissionError) as e:
# Fall back to a writable location
print(f"Warning: Cannot write to {user_dir}, using fallback: {e}")
fallback_dir = Path.home() / f'.{app_name.lower()}'
fallback_dir.mkdir(parents=True, exist_ok=True)
return fallback_dir
return user_dir
except Exception as e:
# Ultimate fallback to home directory
print(f"Error determining user data directory: {e}")
fallback_dir = Path.home() / f'.{app_name.lower()}'
fallback_dir.mkdir(parents=True, exist_ok=True)
return fallback_dir
def get_user_config_dir() -> Path:
"""Get platform-appropriate user config directory
Cross-platform implementation:
- Windows: Same as data directory (%APPDATA%/MbetterClient)
- macOS: Same as data directory (~/Library/Application Support/MbetterClient)
- Linux: ~/.config/MbetterClient (XDG Base Directory)
"""
app_name = "MbetterClient"
try:
if sys.platform.startswith("win") or sys.platform == "darwin":
# Windows and macOS: Use same location as data directory
return get_user_data_dir()
else:
# Linux/Unix: Use XDG config directory
xdg_config_home = os.environ.get('XDG_CONFIG_HOME')
if xdg_config_home:
base_dir = Path(xdg_config_home)
else:
base_dir = Path.home() / '.config'
# Ensure base directory exists
if not base_dir.exists():
base_dir.mkdir(parents=True, exist_ok=True)
# Create application config directory
config_dir = base_dir / app_name
config_dir.mkdir(parents=True, exist_ok=True)
# Verify write permissions
test_file = config_dir / '.write_test'
try:
test_file.write_text('test')
test_file.unlink()
except (OSError, PermissionError) as e:
# Fall back to data directory
print(f"Warning: Cannot write to {config_dir}, using data directory: {e}")
return get_user_data_dir()
return config_dir
except Exception as e:
# Fallback to data directory
print(f"Error determining user config directory: {e}")
return get_user_data_dir()
def get_user_cache_dir() -> Path:
"""Get platform-appropriate user cache directory for temporary files"""
app_name = "MbetterClient"
try:
if sys.platform.startswith("win"):
# Windows: Use Local AppData for cache
local_appdata = os.environ.get('LOCALAPPDATA')
if local_appdata:
base_dir = Path(local_appdata)
else:
base_dir = Path.home() / 'AppData' / 'Local'
elif sys.platform == "darwin":
# macOS: Use Caches directory
base_dir = Path.home() / 'Library' / 'Caches'
else:
# Linux/Unix: Use XDG cache directory
xdg_cache_home = os.environ.get('XDG_CACHE_HOME')
if xdg_cache_home:
base_dir = Path(xdg_cache_home)
else:
base_dir = Path.home() / '.cache'
cache_dir = base_dir / app_name
cache_dir.mkdir(parents=True, exist_ok=True)
return cache_dir
except Exception as e:
# Fallback to data directory
print(f"Error determining user cache directory: {e}")
return get_user_data_dir() / 'cache'
def is_pyinstaller_executable() -> bool:
"""Check if running as PyInstaller executable"""
return getattr(sys, 'frozen', False) and hasattr(sys, '_MEIPASS')
@dataclass
class DatabaseConfig:
"""Database configuration settings"""
path: str = ""
backup_enabled: bool = True
backup_interval_hours: int = 24
max_backups: int = 7
auto_vacuum: bool = True
def __post_init__(self):
"""Set default database path if not specified"""
if not self.path:
# Always use user data directory for database persistence
self.path = str(get_user_data_dir() / "mbetterclient.db")
def get_absolute_path(self, base_path: Optional[Path] = None) -> Path:
"""Get absolute database path"""
# Always use absolute path for persistence
if os.path.isabs(self.path):
return Path(self.path)
else:
# For relative paths, use user data directory
return get_user_data_dir() / self.path
@dataclass
class WebConfig:
"""Web dashboard configuration"""
host: str = "127.0.0.1"
port: int = 5001
debug: bool = False
secret_key: str = "dev-secret-key-change-in-production"
jwt_secret_key: str = "dev-jwt-secret-key"
jwt_expiration_hours: int = 24
session_timeout_hours: int = 8
max_login_attempts: int = 5
rate_limit_enabled: bool = True
# SSL/HTTPS settings
enable_ssl: bool = False
ssl_cert_path: Optional[str] = None
ssl_key_path: Optional[str] = None
ssl_auto_generate: bool = True # Auto-generate self-signed certificate if paths not provided
# Offline mode settings
use_local_assets: bool = True
cdn_fallback_enabled: bool = True
@dataclass
class QtConfig:
"""PyQt video player configuration"""
fullscreen: bool = True
window_width: int = 1920
window_height: int = 1080
always_on_top: bool = False # FIXED: WindowStaysOnTopHint interferes with video rendering on Linux
background_color: str = "#000000"
# Video settings
auto_play: bool = True
loop_video: bool = False
volume: float = 1.0
mute: bool = False
# Overlay settings
overlay_enabled: bool = True
default_template: str = "news_template"
overlay_opacity: float = 0.9
use_native_overlay: bool = False # Use native Qt widgets instead of QWebEngineView
# Performance settings
hardware_acceleration: bool = True
vsync_enabled: bool = True
buffer_size: int = 50
@dataclass
class ApiConfig:
"""REST API client configuration"""
base_url: str = ""
token: str = ""
fastapi_url: str = "https://mbetter.nexlab.net/api/updates"
api_token: str = ""
api_timeout: int = 30
api_enabled: bool = True
api_interval: int = 1800 # Default 30 minutes in seconds
timeout_seconds: int = 30
retry_attempts: int = 3
retry_delay_seconds: int = 5
rustdesk_id: Optional[str] = None # RustDesk ID for periodic whoami calls
# Request intervals (for backward compatibility)
request_interval_hours: int = 0
request_interval_minutes: int = 30
request_interval_seconds: int = 0
# Health check
health_check_enabled: bool = True
health_check_interval_seconds: int = 60
# Request settings
verify_ssl: bool = True
user_agent: str = "MbetterClient/1.0"
max_response_size_mb: int = 100
# Additional API client settings
max_consecutive_failures: int = 5
def __post_init__(self):
"""Post-initialization to sync api_interval with component intervals"""
# If api_interval is provided, update the component intervals
if hasattr(self, 'api_interval') and self.api_interval > 0:
# Convert api_interval to hours, minutes, seconds
total_seconds = self.api_interval
self.request_interval_hours = total_seconds // 3600
total_seconds %= 3600
self.request_interval_minutes = total_seconds // 60
self.request_interval_seconds = total_seconds % 60
else:
# Use component intervals to set api_interval
self.api_interval = self.get_request_interval_seconds()
def get_request_interval_seconds(self) -> int:
"""Get total request interval in seconds"""
return (
self.request_interval_hours * 3600 +
self.request_interval_minutes * 60 +
self.request_interval_seconds
)
@dataclass
class ScreenCastConfig:
"""Screen capture and Chromecast configuration"""
enabled: bool = False
stream_port: int = 8000
chromecast_name: Optional[str] = None # Auto-discover if None
output_dir: Optional[str] = None # Use user data dir if None
resolution: str = "1280x720"
framerate: int = 15
auto_start_capture: bool = False
auto_start_streaming: bool = False
# FFmpeg settings
video_codec: str = "libx264"
audio_codec: str = "aac"
preset: str = "ultrafast"
tune: str = "zerolatency"
@dataclass
class GeneralConfig:
"""General application configuration"""
app_name: str = "MbetterClient"
log_level: str = "INFO"
enable_qt: bool = True
match_interval: int = 20 # Default match interval in minutes
@dataclass
class TimerConfig:
"""Timer configuration for automated game start"""
enabled: bool = False
delay_minutes: int = 4 # Default 4 minutes
@dataclass
class LoggingConfig:
"""Logging configuration"""
level: str = "INFO"
file_path: str = "logs/mbetterclient.log"
max_file_size_mb: int = 10
backup_count: int = 5
format_string: str = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
# Console logging
console_enabled: bool = True
console_level: str = "INFO"
# File logging
file_enabled: bool = True
file_level: str = "DEBUG"
# Component-specific logging
component_levels: Dict[str, str] = field(default_factory=lambda: {
"mbetterclient.qt_player": "INFO",
"mbetterclient.web_dashboard": "INFO",
"mbetterclient.api_client": "INFO",
"mbetterclient.core": "INFO"
})
@dataclass
class AppSettings:
"""Main application settings container"""
# Component configurations
database: DatabaseConfig = field(default_factory=DatabaseConfig)
web: WebConfig = field(default_factory=WebConfig)
qt: QtConfig = field(default_factory=QtConfig)
api: ApiConfig = field(default_factory=ApiConfig)
logging: LoggingConfig = field(default_factory=LoggingConfig)
screen_cast: ScreenCastConfig = field(default_factory=ScreenCastConfig)
general: GeneralConfig = field(default_factory=GeneralConfig)
timer: TimerConfig = field(default_factory=TimerConfig)
# Application settings
version: str = "1.0.0"
debug_mode: bool = False
dev_message: bool = False # Enable debug mode showing only message bus messages
enable_web: bool = True
enable_qt: bool = True
enable_api_client: bool = True
enable_screen_cast: bool = True # Enabled by default, can be disabled with --no-screen-cast
# Runtime settings (not persisted)
fullscreen: bool = True
web_host: str = "127.0.0.1"
web_port: int = 5001
database_path: Optional[str] = None
def __post_init__(self):
"""Post-initialization setup"""
# Sync runtime settings with component configs
self.qt.fullscreen = self.fullscreen
self.web.host = self.web_host
self.web.port = self.web_port
if self.database_path:
self.database.path = self.database_path
# Sync general config with main settings
self.general.app_name = self.general.app_name
self.general.log_level = self.logging.level
self.general.enable_qt = self.enable_qt
def to_dict(self) -> Dict[str, Any]:
"""Convert settings to dictionary"""
return {
"database": self.database.__dict__,
"web": self.web.__dict__,
"qt": self.qt.__dict__,
"api": self.api.__dict__,
"logging": self.logging.__dict__,
"screen_cast": self.screen_cast.__dict__,
"general": self.general.__dict__,
"timer": self.timer.__dict__,
"version": self.version,
"debug_mode": self.debug_mode,
"enable_web": self.enable_web,
"enable_qt": self.enable_qt,
"enable_api_client": self.enable_api_client,
"enable_screen_cast": self.enable_screen_cast
}
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> 'AppSettings':
"""Create settings from dictionary"""
settings = cls()
# Update component configs
if "database" in data:
settings.database = DatabaseConfig(**data["database"])
if "web" in data:
settings.web = WebConfig(**data["web"])
if "qt" in data:
settings.qt = QtConfig(**data["qt"])
if "api" in data:
settings.api = ApiConfig(**data["api"])
if "logging" in data:
settings.logging = LoggingConfig(**data["logging"])
if "screen_cast" in data:
settings.screen_cast = ScreenCastConfig(**data["screen_cast"])
if "general" in data:
settings.general = GeneralConfig(**data["general"])
if "timer" in data:
settings.timer = TimerConfig(**data["timer"])
# Update app settings
for key in ["version", "debug_mode", "enable_web", "enable_qt", "enable_api_client", "enable_screen_cast"]:
if key in data:
setattr(settings, key, data[key])
return settings
def validate(self) -> bool:
"""Validate settings"""
try:
# Validate web config
if not (1 <= self.web.port <= 65535):
raise ValueError(f"Invalid web port: {self.web.port}")
# Validate Qt config
if not (0.0 <= self.qt.volume <= 1.0):
raise ValueError(f"Invalid volume: {self.qt.volume}")
if not (0.0 <= self.qt.overlay_opacity <= 1.0):
raise ValueError(f"Invalid overlay opacity: {self.qt.overlay_opacity}")
# Validate API config
if self.api.timeout_seconds <= 0:
raise ValueError(f"Invalid timeout: {self.api.timeout_seconds}")
if self.api.api_timeout <= 0:
raise ValueError(f"Invalid API timeout: {self.api.api_timeout}")
if self.api.retry_attempts < 0:
raise ValueError(f"Invalid retry attempts: {self.api.retry_attempts}")
# Validate logging config
valid_levels = ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
if self.logging.level not in valid_levels:
raise ValueError(f"Invalid log level: {self.logging.level}")
return True
except Exception as e:
print(f"Settings validation failed: {e}")
return False
def get_project_root(self) -> Path:
"""Get project root directory (for development) or user data directory (for executable)"""
if is_pyinstaller_executable():
return get_user_data_dir()
else:
return Path(__file__).parent.parent.parent
def get_user_data_dir(self) -> Path:
"""Get user data directory for persistent storage"""
return get_user_data_dir()
def get_user_config_dir(self) -> Path:
"""Get user config directory"""
return get_user_config_dir()
def ensure_directories(self):
"""Ensure required directories exist"""
# Use user directories for persistent storage
user_data_dir = get_user_data_dir()
user_config_dir = get_user_config_dir()
# Database directory (in user data)
db_path = self.database.get_absolute_path()
db_path.parent.mkdir(parents=True, exist_ok=True)
# Logs directory (in user data)
if not os.path.isabs(self.logging.file_path):
log_path = user_data_dir / "logs" / Path(self.logging.file_path).name
else:
log_path = Path(self.logging.file_path)
log_path.parent.mkdir(parents=True, exist_ok=True)
# Update logging path to use persistent location
self.logging.file_path = str(log_path)
# Create essential user directories
(user_data_dir / "logs").mkdir(exist_ok=True)
(user_data_dir / "data").mkdir(exist_ok=True)
(user_data_dir / "uploads").mkdir(exist_ok=True)
(user_config_dir / "templates").mkdir(exist_ok=True)
# For development, also create project directories
if not is_pyinstaller_executable():
project_root = Path(__file__).parent.parent.parent
(project_root / "assets").mkdir(exist_ok=True)
(project_root / "templates").mkdir(exist_ok=True)
(project_root / "data").mkdir(exist_ok=True)
(project_root / "logs").mkdir(exist_ok=True)
\ No newline at end of file
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment