"""Dynamic provider configuration management. This module provides runtime configuration management for anime providers, allowing dynamic updates without application restart. """ import json import logging from dataclasses import asdict, dataclass from pathlib import Path from typing import Any, Dict, List, Optional logger = logging.getLogger(__name__) @dataclass class ProviderSettings: """Configuration settings for a single provider.""" name: str enabled: bool = True priority: int = 0 timeout_seconds: int = 30 max_retries: int = 3 retry_delay_seconds: float = 1.0 max_concurrent_downloads: int = 3 bandwidth_limit_mbps: Optional[float] = None custom_headers: Optional[Dict[str, str]] = None custom_params: Optional[Dict[str, Any]] = None def to_dict(self) -> Dict[str, Any]: """Convert settings to dictionary.""" return { k: v for k, v in asdict(self).items() if v is not None } @classmethod def from_dict(cls, data: Dict[str, Any]) -> "ProviderSettings": """Create settings from dictionary.""" return cls(**{k: v for k, v in data.items() if hasattr(cls, k)}) class ProviderConfigManager: """Manages dynamic configuration for anime providers.""" def __init__(self, config_file: Optional[Path] = None): """Initialize provider configuration manager. Args: config_file: Path to configuration file (optional). """ self._config_file = config_file self._provider_settings: Dict[str, ProviderSettings] = {} self._global_settings: Dict[str, Any] = { "default_timeout": 30, "default_max_retries": 3, "default_retry_delay": 1.0, "enable_health_monitoring": True, "enable_failover": True, } # Load configuration if file exists if config_file and config_file.exists(): self.load_config() logger.info("Provider configuration manager initialized") def get_provider_settings( self, provider_name: str ) -> Optional[ProviderSettings]: """Get settings for a specific provider. Args: provider_name: Name of the provider. Returns: Provider settings or None if not configured. """ return self._provider_settings.get(provider_name) def set_provider_settings( self, provider_name: str, settings: ProviderSettings ) -> None: """Set settings for a specific provider. Args: provider_name: Name of the provider. settings: Provider settings to apply. """ self._provider_settings[provider_name] = settings logger.info(f"Updated settings for provider: {provider_name}") def update_provider_settings( self, provider_name: str, **kwargs ) -> bool: """Update specific provider settings. Args: provider_name: Name of the provider. **kwargs: Settings to update. Returns: True if updated, False if provider not found. """ if provider_name not in self._provider_settings: # Create new settings self._provider_settings[provider_name] = ProviderSettings( name=provider_name, **kwargs ) logger.info(f"Created new settings for provider: {provider_name}") # noqa: E501 return True settings = self._provider_settings[provider_name] # Update settings for key, value in kwargs.items(): if hasattr(settings, key): setattr(settings, key, value) logger.info( f"Updated settings for provider {provider_name}: {kwargs}" ) return True def get_all_provider_settings(self) -> Dict[str, ProviderSettings]: """Get settings for all configured providers. Returns: Dictionary mapping provider names to their settings. """ return self._provider_settings.copy() def get_enabled_providers(self) -> List[str]: """Get list of enabled providers. Returns: List of enabled provider names. """ return [ name for name, settings in self._provider_settings.items() if settings.enabled ] def enable_provider(self, provider_name: str) -> bool: """Enable a provider. Args: provider_name: Name of the provider. Returns: True if enabled, False if not found. """ if provider_name in self._provider_settings: self._provider_settings[provider_name].enabled = True logger.info(f"Enabled provider: {provider_name}") return True return False def disable_provider(self, provider_name: str) -> bool: """Disable a provider. Args: provider_name: Name of the provider. Returns: True if disabled, False if not found. """ if provider_name in self._provider_settings: self._provider_settings[provider_name].enabled = False logger.info(f"Disabled provider: {provider_name}") return True return False def set_provider_priority( self, provider_name: str, priority: int ) -> bool: """Set priority for a provider. Lower priority values = higher priority. Args: provider_name: Name of the provider. priority: Priority value (lower = higher priority). Returns: True if updated, False if not found. """ if provider_name in self._provider_settings: self._provider_settings[provider_name].priority = priority logger.info( f"Set priority for {provider_name} to {priority}" ) return True return False def get_providers_by_priority(self) -> List[str]: """Get providers sorted by priority. Returns: List of provider names sorted by priority (low to high). """ sorted_providers = sorted( self._provider_settings.items(), key=lambda x: x[1].priority, ) return [name for name, _ in sorted_providers] def get_global_setting(self, key: str) -> Optional[Any]: """Get a global setting value. Args: key: Setting key. Returns: Setting value or None if not found. """ return self._global_settings.get(key) def set_global_setting(self, key: str, value: Any) -> None: """Set a global setting value. Args: key: Setting key. value: Setting value. """ self._global_settings[key] = value logger.info(f"Updated global setting {key}: {value}") def get_all_global_settings(self) -> Dict[str, Any]: """Get all global settings. Returns: Dictionary of global settings. """ return self._global_settings.copy() def load_config(self, file_path: Optional[Path] = None) -> bool: """Load configuration from file. Args: file_path: Path to configuration file (uses default if None). Returns: True if loaded successfully, False otherwise. """ config_path = file_path or self._config_file if not config_path or not config_path.exists(): logger.warning( f"Configuration file not found: {config_path}" ) return False try: with open(config_path, "r", encoding="utf-8") as f: data = json.load(f) # Load provider settings if "providers" in data: for name, settings_data in data["providers"].items(): self._provider_settings[name] = ( ProviderSettings.from_dict(settings_data) ) # Load global settings if "global" in data: self._global_settings.update(data["global"]) logger.info( f"Loaded configuration from {config_path} " f"({len(self._provider_settings)} providers)" ) return True except Exception as e: logger.error( f"Failed to load configuration from {config_path}: {e}", exc_info=True, ) return False def save_config(self, file_path: Optional[Path] = None) -> bool: """Save configuration to file. Args: file_path: Path to save to (uses default if None). Returns: True if saved successfully, False otherwise. """ config_path = file_path or self._config_file if not config_path: logger.error("No configuration file path specified") return False try: # Ensure parent directory exists config_path.parent.mkdir(parents=True, exist_ok=True) data = { "providers": { name: settings.to_dict() for name, settings in self._provider_settings.items() }, "global": self._global_settings, } with open(config_path, "w", encoding="utf-8") as f: json.dump(data, f, indent=2) logger.info(f"Saved configuration to {config_path}") return True except Exception as e: logger.error( f"Failed to save configuration to {config_path}: {e}", exc_info=True, ) return False def reset_to_defaults(self) -> None: """Reset all settings to defaults.""" self._provider_settings.clear() self._global_settings = { "default_timeout": 30, "default_max_retries": 3, "default_retry_delay": 1.0, "enable_health_monitoring": True, "enable_failover": True, } logger.info("Reset configuration to defaults") # Global configuration manager instance _config_manager: Optional[ProviderConfigManager] = None def get_config_manager( config_file: Optional[Path] = None, ) -> ProviderConfigManager: """Get or create global provider configuration manager. Args: config_file: Configuration file path (used on first call). Returns: Global ProviderConfigManager instance. """ global _config_manager if _config_manager is None: _config_manager = ProviderConfigManager(config_file=config_file) return _config_manager