"""Configuration Application Service for settings management.""" import logging import os import json from typing import Dict, List, Any, Optional, Union from pathlib import Path from dataclasses import asdict from ..error_handling.error_mapper import ErrorMapper from ..error_handling.structured_logger import StructuredLogger, LogContext, get_structured_logger from ...infrastructure.config.app_config import AppConfig, TTSConfig, STTConfig, TranslationConfig, ProcessingConfig, LoggingConfig from ...infrastructure.config.dependency_container import DependencyContainer from ...domain.exceptions import DomainException logger = get_structured_logger(__name__) class ConfigurationException(DomainException): """Exception raised for configuration-related errors.""" pass class ConfigurationApplicationService: """Application service for managing application configuration and settings.""" def __init__( self, container: DependencyContainer, config: Optional[AppConfig] = None ): """ Initialize the configuration application service. Args: container: Dependency injection container config: Application configuration (optional, will be resolved from container) """ self._container = container self._config = config or container.resolve(AppConfig) # Initialize error handling self._error_mapper = ErrorMapper() logger.info("ConfigurationApplicationService initialized") def get_current_configuration(self) -> Dict[str, Any]: """ Get the current application configuration. Returns: Dict[str, Any]: Current configuration as dictionary """ try: return { 'tts': self._config.get_tts_config(), 'stt': self._config.get_stt_config(), 'translation': self._config.get_translation_config(), 'processing': self._config.get_processing_config(), 'logging': self._config.get_logging_config() } except Exception as e: logger.error(f"Failed to get current configuration: {e}") raise ConfigurationException(f"Failed to retrieve configuration: {str(e)}") def get_tts_configuration(self) -> Dict[str, Any]: """ Get TTS-specific configuration. Returns: Dict[str, Any]: TTS configuration """ try: return self._config.get_tts_config() except Exception as e: logger.error(f"Failed to get TTS configuration: {e}") raise ConfigurationException(f"Failed to retrieve TTS configuration: {str(e)}") def get_stt_configuration(self) -> Dict[str, Any]: """ Get STT-specific configuration. Returns: Dict[str, Any]: STT configuration """ try: return self._config.get_stt_config() except Exception as e: logger.error(f"Failed to get STT configuration: {e}") raise ConfigurationException(f"Failed to retrieve STT configuration: {str(e)}") def get_translation_configuration(self) -> Dict[str, Any]: """ Get translation-specific configuration. Returns: Dict[str, Any]: Translation configuration """ try: return self._config.get_translation_config() except Exception as e: logger.error(f"Failed to get translation configuration: {e}") raise ConfigurationException(f"Failed to retrieve translation configuration: {str(e)}") def get_processing_configuration(self) -> Dict[str, Any]: """ Get processing-specific configuration. Returns: Dict[str, Any]: Processing configuration """ try: return self._config.get_processing_config() except Exception as e: logger.error(f"Failed to get processing configuration: {e}") raise ConfigurationException(f"Failed to retrieve processing configuration: {str(e)}") def get_logging_configuration(self) -> Dict[str, Any]: """ Get logging-specific configuration. Returns: Dict[str, Any]: Logging configuration """ try: return self._config.get_logging_config() except Exception as e: logger.error(f"Failed to get logging configuration: {e}") raise ConfigurationException(f"Failed to retrieve logging configuration: {str(e)}") def update_tts_configuration(self, updates: Dict[str, Any]) -> Dict[str, Any]: """ Update TTS configuration. Args: updates: Configuration updates to apply Returns: Dict[str, Any]: Updated TTS configuration Raises: ConfigurationException: If update fails or validation fails """ try: # Validate updates self._validate_tts_updates(updates) # Apply updates to current config current_config = self._config.get_tts_config() for key, value in updates.items(): if key in current_config: # Update the actual config object if hasattr(self._config.tts, key): setattr(self._config.tts, key, value) logger.info(f"Updated TTS config: {key} = {value}") else: logger.warning(f"Unknown TTS configuration key: {key}") # Return updated configuration updated_config = self._config.get_tts_config() logger.info(f"TTS configuration updated: {list(updates.keys())}") return updated_config except Exception as e: logger.error(f"Failed to update TTS configuration: {e}") raise ConfigurationException(f"Failed to update TTS configuration: {str(e)}") def update_stt_configuration(self, updates: Dict[str, Any]) -> Dict[str, Any]: """ Update STT configuration. Args: updates: Configuration updates to apply Returns: Dict[str, Any]: Updated STT configuration Raises: ConfigurationException: If update fails or validation fails """ try: # Validate updates self._validate_stt_updates(updates) # Apply updates to current config current_config = self._config.get_stt_config() for key, value in updates.items(): if key in current_config: # Update the actual config object if hasattr(self._config.stt, key): setattr(self._config.stt, key, value) logger.info(f"Updated STT config: {key} = {value}") else: logger.warning(f"Unknown STT configuration key: {key}") # Return updated configuration updated_config = self._config.get_stt_config() logger.info(f"STT configuration updated: {list(updates.keys())}") return updated_config except Exception as e: logger.error(f"Failed to update STT configuration: {e}") raise ConfigurationException(f"Failed to update STT configuration: {str(e)}") def update_translation_configuration(self, updates: Dict[str, Any]) -> Dict[str, Any]: """ Update translation configuration. Args: updates: Configuration updates to apply Returns: Dict[str, Any]: Updated translation configuration Raises: ConfigurationException: If update fails or validation fails """ try: # Validate updates self._validate_translation_updates(updates) # Apply updates to current config current_config = self._config.get_translation_config() for key, value in updates.items(): if key in current_config: # Update the actual config object if hasattr(self._config.translation, key): setattr(self._config.translation, key, value) logger.info(f"Updated translation config: {key} = {value}") else: logger.warning(f"Unknown translation configuration key: {key}") # Return updated configuration updated_config = self._config.get_translation_config() logger.info(f"Translation configuration updated: {list(updates.keys())}") return updated_config except Exception as e: logger.error(f"Failed to update translation configuration: {e}") raise ConfigurationException(f"Failed to update translation configuration: {str(e)}") def update_processing_configuration(self, updates: Dict[str, Any]) -> Dict[str, Any]: """ Update processing configuration. Args: updates: Configuration updates to apply Returns: Dict[str, Any]: Updated processing configuration Raises: ConfigurationException: If update fails or validation fails """ try: # Validate updates self._validate_processing_updates(updates) # Apply updates to current config current_config = self._config.get_processing_config() for key, value in updates.items(): if key in current_config: # Update the actual config object if hasattr(self._config.processing, key): setattr(self._config.processing, key, value) logger.info(f"Updated processing config: {key} = {value}") else: logger.warning(f"Unknown processing configuration key: {key}") # Return updated configuration updated_config = self._config.get_processing_config() logger.info(f"Processing configuration updated: {list(updates.keys())}") return updated_config except Exception as e: logger.error(f"Failed to update processing configuration: {e}") raise ConfigurationException(f"Failed to update processing configuration: {str(e)}") def _validate_tts_updates(self, updates: Dict[str, Any]) -> None: """ Validate TTS configuration updates. Args: updates: Updates to validate Raises: ConfigurationException: If validation fails """ valid_providers = ['chatterbox', 'dummy'] valid_languages = ['en', 'es', 'fr', 'de', 'it', 'pt', 'ru', 'ja', 'ko', 'zh'] for key, value in updates.items(): if key == 'preferred_providers': if not isinstance(value, list): raise ConfigurationException("preferred_providers must be a list") for provider in value: if provider not in valid_providers: raise ConfigurationException(f"Invalid TTS provider: {provider}") elif key == 'default_speed': if not isinstance(value, (int, float)) or not (0.1 <= value <= 3.0): raise ConfigurationException("default_speed must be between 0.1 and 3.0") elif key == 'default_language': if value not in valid_languages: raise ConfigurationException(f"Invalid language: {value}") elif key == 'enable_streaming': if not isinstance(value, bool): raise ConfigurationException("enable_streaming must be a boolean") elif key == 'max_text_length': if not isinstance(value, int) or value <= 0: raise ConfigurationException("max_text_length must be a positive integer") def _validate_stt_updates(self, updates: Dict[str, Any]) -> None: """ Validate STT configuration updates. Args: updates: Updates to validate Raises: ConfigurationException: If validation fails """ valid_providers = ['whisper', 'parakeet'] for key, value in updates.items(): if key == 'preferred_providers': if not isinstance(value, list): raise ConfigurationException("preferred_providers must be a list") for provider in value: if provider not in valid_providers: raise ConfigurationException(f"Invalid STT provider: {provider}") elif key == 'default_model': if value not in valid_providers: raise ConfigurationException(f"Invalid STT model: {value}") elif key == 'chunk_length_s': if not isinstance(value, int) or value <= 0: raise ConfigurationException("chunk_length_s must be a positive integer") elif key == 'batch_size': if not isinstance(value, int) or value <= 0: raise ConfigurationException("batch_size must be a positive integer") elif key == 'enable_vad': if not isinstance(value, bool): raise ConfigurationException("enable_vad must be a boolean") def _validate_translation_updates(self, updates: Dict[str, Any]) -> None: """ Validate translation configuration updates. Args: updates: Updates to validate Raises: ConfigurationException: If validation fails """ for key, value in updates.items(): if key == 'default_provider': if not isinstance(value, str) or not value: raise ConfigurationException("default_provider must be a non-empty string") elif key == 'model_name': if not isinstance(value, str) or not value: raise ConfigurationException("model_name must be a non-empty string") elif key == 'max_chunk_length': if not isinstance(value, int) or value <= 0: raise ConfigurationException("max_chunk_length must be a positive integer") elif key == 'batch_size': if not isinstance(value, int) or value <= 0: raise ConfigurationException("batch_size must be a positive integer") elif key == 'cache_translations': if not isinstance(value, bool): raise ConfigurationException("cache_translations must be a boolean") def _validate_processing_updates(self, updates: Dict[str, Any]) -> None: """ Validate processing configuration updates. Args: updates: Updates to validate Raises: ConfigurationException: If validation fails """ for key, value in updates.items(): if key == 'temp_dir': if not isinstance(value, str) or not value: raise ConfigurationException("temp_dir must be a non-empty string") # Try to create directory to validate path try: Path(value).mkdir(parents=True, exist_ok=True) except Exception as e: raise ConfigurationException(f"Invalid temp_dir path: {e}") elif key == 'cleanup_temp_files': if not isinstance(value, bool): raise ConfigurationException("cleanup_temp_files must be a boolean") elif key == 'max_file_size_mb': if not isinstance(value, int) or value <= 0: raise ConfigurationException("max_file_size_mb must be a positive integer") elif key == 'supported_audio_formats': if not isinstance(value, list): raise ConfigurationException("supported_audio_formats must be a list") valid_formats = ['wav', 'mp3', 'flac', 'ogg', 'm4a'] for fmt in value: if fmt not in valid_formats: raise ConfigurationException(f"Invalid audio format: {fmt}") elif key == 'processing_timeout_seconds': if not isinstance(value, int) or value <= 0: raise ConfigurationException("processing_timeout_seconds must be a positive integer") def save_configuration_to_file(self, file_path: str) -> None: """ Save current configuration to file. Args: file_path: Path to save configuration file Raises: ConfigurationException: If save fails """ try: self._config.save_configuration(file_path) logger.info(f"Configuration saved to {file_path}") except Exception as e: logger.error(f"Failed to save configuration to {file_path}: {e}") raise ConfigurationException(f"Failed to save configuration: {str(e)}") def load_configuration_from_file(self, file_path: str) -> Dict[str, Any]: """ Load configuration from file. Args: file_path: Path to configuration file Returns: Dict[str, Any]: Loaded configuration Raises: ConfigurationException: If load fails """ try: if not os.path.exists(file_path): raise ConfigurationException(f"Configuration file not found: {file_path}") # Create new config instance with the file new_config = AppConfig(config_file=file_path) # Update current config self._config = new_config # Update container with new config self._container.register_singleton(AppConfig, new_config) logger.info(f"Configuration loaded from {file_path}") return self.get_current_configuration() except Exception as e: logger.error(f"Failed to load configuration from {file_path}: {e}") raise ConfigurationException(f"Failed to load configuration: {str(e)}") def reload_configuration(self) -> Dict[str, Any]: """ Reload configuration from original sources. Returns: Dict[str, Any]: Reloaded configuration Raises: ConfigurationException: If reload fails """ try: self._config.reload_configuration() logger.info("Configuration reloaded successfully") return self.get_current_configuration() except Exception as e: logger.error(f"Failed to reload configuration: {e}") raise ConfigurationException(f"Failed to reload configuration: {str(e)}") def get_provider_availability(self) -> Dict[str, Dict[str, bool]]: """ Get availability status of all providers. Returns: Dict[str, Dict[str, bool]]: Provider availability by category """ try: availability = { 'tts': {}, 'stt': {}, 'translation': {} } # Check TTS providers tts_factory = self._container.resolve(type(self._container._get_tts_factory())) for provider in ['chatterbox', 'dummy']: try: tts_factory.create_provider(provider) availability['tts'][provider] = True except Exception: availability['tts'][provider] = False # Check STT providers stt_factory = self._container.resolve(type(self._container._get_stt_factory())) for provider in ['whisper', 'parakeet']: try: stt_factory.create_provider(provider) availability['stt'][provider] = True except Exception: availability['stt'][provider] = False # Check translation providers translation_factory = self._container.resolve(type(self._container._get_translation_factory())) try: translation_factory.get_default_provider() availability['translation']['nllb'] = True except Exception: availability['translation']['nllb'] = False return availability except Exception as e: logger.error(f"Failed to check provider availability: {e}") raise ConfigurationException(f"Failed to check provider availability: {str(e)}") def get_system_info(self) -> Dict[str, Any]: """ Get system information and configuration summary. Returns: Dict[str, Any]: System information """ try: return { 'config_file': self._config.config_file, 'temp_directory': self._config.processing.temp_dir, 'log_level': self._config.logging.level, 'provider_availability': self.get_provider_availability(), 'supported_languages': [ 'en', 'es', 'fr', 'de', 'it', 'pt', 'ru', 'ja', 'ko', 'zh', 'ar', 'hi', 'tr', 'pl', 'nl', 'sv', 'da', 'no', 'fi' ], 'supported_audio_formats': self._config.processing.supported_audio_formats, 'max_file_size_mb': self._config.processing.max_file_size_mb, 'processing_timeout_seconds': self._config.processing.processing_timeout_seconds } except Exception as e: logger.error(f"Failed to get system info: {e}") raise ConfigurationException(f"Failed to get system info: {str(e)}") def validate_configuration(self) -> Dict[str, List[str]]: """ Validate current configuration and return any issues. Returns: Dict[str, List[str]]: Validation issues by category """ issues = { 'tts': [], 'stt': [], 'translation': [], 'processing': [], 'logging': [] } try: # Validate TTS configuration tts_config = self._config.get_tts_config() if not (0.1 <= tts_config['default_speed'] <= 3.0): issues['tts'].append(f"Invalid default_speed: {tts_config['default_speed']}") if tts_config['max_text_length'] <= 0: issues['tts'].append(f"Invalid max_text_length: {tts_config['max_text_length']}") # Validate STT configuration stt_config = self._config.get_stt_config() if stt_config['chunk_length_s'] <= 0: issues['stt'].append(f"Invalid chunk_length_s: {stt_config['chunk_length_s']}") if stt_config['batch_size'] <= 0: issues['stt'].append(f"Invalid batch_size: {stt_config['batch_size']}") # Validate processing configuration processing_config = self._config.get_processing_config() if not os.path.exists(processing_config['temp_dir']): issues['processing'].append(f"Temp directory does not exist: {processing_config['temp_dir']}") if processing_config['max_file_size_mb'] <= 0: issues['processing'].append(f"Invalid max_file_size_mb: {processing_config['max_file_size_mb']}") # Validate logging configuration logging_config = self._config.get_logging_config() valid_levels = ['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] if logging_config['level'].upper() not in valid_levels: issues['logging'].append(f"Invalid log level: {logging_config['level']}") except Exception as e: issues['general'] = [f"Configuration validation error: {str(e)}"] return issues def reset_to_defaults(self) -> Dict[str, Any]: """ Reset configuration to default values. Returns: Dict[str, Any]: Reset configuration Raises: ConfigurationException: If reset fails """ try: # Create new config with defaults default_config = AppConfig() # Update current config self._config = default_config # Update container with new config self._container.register_singleton(AppConfig, default_config) logger.info("Configuration reset to defaults") return self.get_current_configuration() except Exception as e: logger.error(f"Failed to reset configuration: {e}") raise ConfigurationException(f"Failed to reset configuration: {str(e)}") def cleanup(self) -> None: """Cleanup configuration service resources.""" logger.info("Cleaning up ConfigurationApplicationService") # No specific cleanup needed for configuration service logger.info("ConfigurationApplicationService cleanup completed") def __enter__(self): """Context manager entry.""" return self def __exit__(self, exc_type, exc_val, exc_tb): """Context manager exit with cleanup.""" self.cleanup()