Lukas
fecdb38a90
- Implemented ProviderHealthMonitor for real-time tracking - Monitors availability, response times, success rates - Automatic marking unavailable after failures - Background health check loop - Added ProviderFailover for automatic provider switching - Configurable retry attempts with exponential backoff - Integration with health monitoring - Smart provider selection - Created MonitoredProviderWrapper for performance tracking - Transparent monitoring for any provider - Automatic metric recording - No changes needed to existing providers - Implemented ProviderConfigManager for dynamic configuration - Runtime updates without restart - Per-provider settings (timeout, retries, bandwidth) - JSON-based persistence - Added Provider Management API (15+ endpoints) - Health monitoring endpoints - Configuration management - Failover control - Comprehensive testing (34 tests, 100% pass rate) - Health monitoring tests - Failover scenario tests - Configuration management tests - Documentation updates - Updated infrastructure.md - Updated instructions.md - Created PROVIDER_ENHANCEMENT_SUMMARY.md Total: ~2,593 lines of code, 34 passing tests
352 lines
10 KiB
Python
352 lines
10 KiB
Python
"""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
|