"""Audio Processing Application Service for pipeline orchestration.""" import logging import os import tempfile import time import uuid from pathlib import Path from typing import Optional, Dict, Any from contextlib import contextmanager from ..dtos.audio_upload_dto import AudioUploadDto from ..dtos.processing_request_dto import ProcessingRequestDto from ..dtos.processing_result_dto import ProcessingResultDto from ..error_handling.error_mapper import ErrorMapper from ..error_handling.structured_logger import StructuredLogger, LogContext, get_structured_logger from ..error_handling.recovery_manager import RecoveryManager, RetryConfig, CircuitBreakerConfig from ...domain.interfaces.speech_recognition import ISpeechRecognitionService from ...domain.interfaces.translation import ITranslationService from ...domain.interfaces.speech_synthesis import ISpeechSynthesisService from ...domain.models.audio_content import AudioContent from ...domain.models.text_content import TextContent from ...domain.models.translation_request import TranslationRequest from ...domain.models.speech_synthesis_request import SpeechSynthesisRequest from ...domain.models.voice_settings import VoiceSettings from ...domain.exceptions import ( DomainException, AudioProcessingException, SpeechRecognitionException, TranslationFailedException, SpeechSynthesisException ) from ...infrastructure.config.app_config import AppConfig from ...infrastructure.config.dependency_container import DependencyContainer logger = get_structured_logger(__name__) class AudioProcessingApplicationService: """Application service for orchestrating the complete audio processing pipeline.""" def __init__( self, container: DependencyContainer, config: Optional[AppConfig] = None ): """ Initialize the audio processing application service. Args: container: Dependency injection container config: Application configuration (optional, will be resolved from container) """ try: logger.info("Initializing AudioProcessingApplicationService...") self._container = container self._config = config or container.resolve(AppConfig) self._temp_files: Dict[str, str] = {} # Track temporary files for cleanup # Initialize error handling components self._error_mapper = ErrorMapper() self._recovery_manager = RecoveryManager() # Skip complex logging setup for now to avoid issues # self._setup_logging() logger.info("AudioProcessingApplicationService initialized successfully") except Exception as e: print(f"Error: Failed to initialize AudioProcessingApplicationService: {e}") raise def _setup_logging(self) -> None: """Setup logging configuration.""" try: log_config = self._config.get_logging_config() # Configure logger level logger.setLevel(getattr(logging, log_config['level'].upper(), logging.INFO)) # Add file handler if enabled if log_config.get('enable_file_logging', False): file_handler = logging.FileHandler(log_config['log_file_path']) file_handler.setLevel(logger.level) formatter = logging.Formatter(log_config['format']) file_handler.setFormatter(formatter) logger.addHandler(file_handler) except Exception as e: logger.warning(f"Failed to setup logging configuration: {e}") def process_audio_pipeline(self, request: ProcessingRequestDto) -> ProcessingResultDto: """ Process audio through the complete pipeline: STT -> Translation -> TTS. Args: request: Processing request containing audio and parameters Returns: ProcessingResultDto: Result of the complete processing pipeline """ # Generate correlation ID and start operation logging correlation_id = logger.log_operation_start( "audio_processing_pipeline", extra={ 'asr_model': request.asr_model, 'target_language': request.target_language, 'voice': request.voice, 'file_name': request.audio.filename, 'file_size': request.audio.size } ) start_time = time.time() context = LogContext( correlation_id=correlation_id, operation="audio_processing_pipeline", component="AudioProcessingApplicationService" ) try: # Validate request self._validate_request(request) # Create temporary working directory with self._create_temp_directory(correlation_id) as temp_dir: # Step 1: Convert uploaded audio to domain model audio_content = self._convert_upload_to_audio_content(request.audio, temp_dir) # Step 2: Speech-to-Text with retry and fallback original_text = self._perform_speech_recognition_with_recovery( audio_content, request.asr_model, correlation_id ) # Step 3: Translation (if needed) with retry translated_text = original_text if request.requires_translation: translated_text = self._perform_translation_with_recovery( original_text, request.source_language, request.target_language, correlation_id ) # Step 4: Text-to-Speech with fallback providers output_audio_path = self._perform_speech_synthesis_with_recovery( translated_text, request.voice, request.speed, request.target_language, temp_dir, correlation_id ) # Calculate processing time processing_time = time.time() - start_time # Create successful result result = ProcessingResultDto.success_result( original_text=original_text.text, translated_text=translated_text.text if translated_text != original_text else None, audio_path=output_audio_path, processing_time=processing_time, metadata={ 'correlation_id': correlation_id, 'asr_model': request.asr_model, 'target_language': request.target_language, 'voice': request.voice, 'speed': request.speed, 'translation_required': request.requires_translation } ) # Log successful completion logger.log_operation_end( "audio_processing_pipeline", correlation_id, success=True, duration=processing_time, context=context, extra={ 'original_text_length': len(original_text.text), 'translated_text_length': len(translated_text.text) if translated_text != original_text else 0, 'output_file': output_audio_path } ) return result except DomainException as e: processing_time = time.time() - start_time # Map exception to user-friendly error error_context = { 'file_name': request.audio.filename, 'file_size': request.audio.size, 'operation': 'audio_processing_pipeline', 'correlation_id': correlation_id } error_mapping = self._error_mapper.map_exception(e, error_context) logger.error( f"Domain error in audio processing pipeline: {error_mapping.user_message}", context=context, exception=e, extra={ 'error_code': error_mapping.error_code, 'error_category': error_mapping.category.value, 'error_severity': error_mapping.severity.value, 'recovery_suggestions': error_mapping.recovery_suggestions } ) # Log operation failure logger.log_operation_end( "audio_processing_pipeline", correlation_id, success=False, duration=processing_time, context=context ) return ProcessingResultDto.error_result( error_message=error_mapping.user_message, error_code=error_mapping.error_code, processing_time=processing_time, metadata={ 'correlation_id': correlation_id, 'error_category': error_mapping.category.value, 'error_severity': error_mapping.severity.value, 'recovery_suggestions': error_mapping.recovery_suggestions, 'technical_details': error_mapping.technical_details } ) except Exception as e: processing_time = time.time() - start_time # Map unexpected exception error_context = { 'file_name': request.audio.filename, 'operation': 'audio_processing_pipeline', 'correlation_id': correlation_id } error_mapping = self._error_mapper.map_exception(e, error_context) logger.critical( f"Unexpected error in audio processing pipeline: {error_mapping.user_message}", context=context, exception=e, extra={ 'error_code': error_mapping.error_code, 'error_category': error_mapping.category.value, 'error_severity': error_mapping.severity.value } ) # Log operation failure logger.log_operation_end( "audio_processing_pipeline", correlation_id, success=False, duration=processing_time, context=context ) return ProcessingResultDto.error_result( error_message=error_mapping.user_message, error_code=error_mapping.error_code, processing_time=processing_time, metadata={ 'correlation_id': correlation_id, 'error_category': error_mapping.category.value, 'error_severity': error_mapping.severity.value, 'technical_details': error_mapping.technical_details } ) finally: # Cleanup temporary files self._cleanup_temp_files() def _validate_request(self, request: ProcessingRequestDto) -> None: """ Validate processing request. Args: request: Processing request to validate Raises: ValueError: If request is invalid """ if not isinstance(request, ProcessingRequestDto): raise ValueError("Request must be a ProcessingRequestDto instance") # Additional validation beyond DTO validation processing_config = self._config.get_processing_config() # Check file size limits max_size_bytes = processing_config['max_file_size_mb'] * 1024 * 1024 if request.audio.size > max_size_bytes: raise ValueError( f"Audio file too large: {request.audio.size} bytes. " f"Maximum allowed: {max_size_bytes} bytes" ) # Check supported audio formats supported_formats = processing_config['supported_audio_formats'] file_ext = request.audio.file_extension.lstrip('.') if file_ext not in supported_formats: raise ValueError( f"Unsupported audio format: {file_ext}. " f"Supported formats: {supported_formats}" ) @contextmanager def _create_temp_directory(self, correlation_id: str): """ Create temporary directory for processing. Args: correlation_id: Correlation ID for tracking Yields: str: Path to temporary directory """ processing_config = self._config.get_processing_config() base_temp_dir = processing_config['temp_dir'] # Create unique temp directory temp_dir = os.path.join(base_temp_dir, f"processing_{correlation_id}") try: os.makedirs(temp_dir, exist_ok=True) logger.info(f"Created temporary directory: {temp_dir}") yield temp_dir finally: # Cleanup temp directory if configured if processing_config.get('cleanup_temp_files', True): try: import shutil shutil.rmtree(temp_dir, ignore_errors=True) logger.info(f"Cleaned up temporary directory: {temp_dir}") except Exception as e: logger.warning(f"Failed to cleanup temp directory {temp_dir}: {e}") def _convert_upload_to_audio_content( self, upload: AudioUploadDto, temp_dir: str ) -> AudioContent: """ Convert uploaded audio to domain AudioContent. Args: upload: Audio upload DTO temp_dir: Temporary directory for file operations Returns: AudioContent: Domain audio content model Raises: AudioProcessingException: If conversion fails """ try: # Save uploaded content to temporary file temp_file_path = os.path.join(temp_dir, f"input_{upload.filename}") with open(temp_file_path, 'wb') as f: f.write(upload.content) # Track temp file for cleanup self._temp_files[temp_file_path] = temp_file_path # Determine audio format from file extension audio_format = upload.file_extension.lstrip('.').lower() # Create AudioContent (simplified - in real implementation would extract metadata) # For now, set a minimal positive duration to pass validation # In a real implementation, you would extract actual duration from the audio file audio_content = AudioContent( data=upload.content, format=audio_format, sample_rate=16000, # Default, would be extracted from actual file duration=1.0 # Set minimal positive duration to pass validation ) logger.info(f"Converted upload to AudioContent: {upload.filename}") return audio_content except Exception as e: logger.error(f"Failed to convert upload to AudioContent: {e}") raise AudioProcessingException(f"Failed to process uploaded audio: {str(e)}") def _perform_speech_recognition( self, audio: AudioContent, model: str, correlation_id: str ) -> TextContent: """ Perform speech-to-text recognition. Args: audio: Audio content to transcribe model: STT model to use correlation_id: Correlation ID for tracking Returns: TextContent: Transcribed text Raises: SpeechRecognitionException: If STT fails """ try: logger.info(f"Starting STT with model: {model} [correlation_id={correlation_id}]") # Get STT provider from container stt_provider = self._container.get_stt_provider(model) # Perform transcription text_content = stt_provider.transcribe(audio, model) logger.info( f"STT completed successfully [correlation_id={correlation_id}, " f"text_length={len(text_content.text)}]" ) return text_content except Exception as e: logger.error(f"STT failed: {e} [correlation_id={correlation_id}]") raise SpeechRecognitionException(f"Speech recognition failed: {str(e)}") def _perform_translation( self, text: TextContent, source_language: Optional[str], target_language: str, correlation_id: str ) -> TextContent: """ Perform text translation. Args: text: Text to translate source_language: Source language (optional, auto-detect if None) target_language: Target language correlation_id: Correlation ID for tracking Returns: TextContent: Translated text Raises: TranslationFailedException: If translation fails """ try: logger.info( f"Starting translation: {source_language or 'auto'} -> {target_language} " f"[correlation_id={correlation_id}]" ) # Get translation provider from container translation_provider = self._container.get_translation_provider() # Create translation request translation_request = TranslationRequest( source_text=text, # text is already a TextContent object target_language=target_language, source_language=source_language ) # Perform translation translated_text = translation_provider.translate(translation_request) logger.info( f"Translation completed successfully [correlation_id={correlation_id}, " f"source_length={len(text.text)}, target_length={len(translated_text.text)}]" ) return translated_text except Exception as e: logger.error(f"Translation failed: {e} [correlation_id={correlation_id}]") raise TranslationFailedException(f"Translation failed: {str(e)}") def _perform_speech_synthesis( self, text: TextContent, voice: str, speed: float, language: str, temp_dir: str, correlation_id: str ) -> str: """ Perform text-to-speech synthesis. Args: text: Text to synthesize voice: Voice to use speed: Speech speed language: Target language temp_dir: Temporary directory for output correlation_id: Correlation ID for tracking Returns: str: Path to generated audio file Raises: SpeechSynthesisException: If TTS fails """ try: logger.info( f"Starting TTS with voice: {voice}, speed: {speed}, language: {language} " f"[correlation_id={correlation_id}]" ) logger.info(f"Text to synthesize length: {len(text.text)} characters") # Get TTS provider from container logger.info(f"Getting TTS provider for voice: {voice}") tts_provider = self._container.get_tts_provider(voice) logger.info(f"TTS provider obtained: {tts_provider.__class__.__name__}") # Create voice settings logger.info("Creating voice settings") voice_settings = VoiceSettings( voice_id=voice, speed=speed, language=language ) logger.info(f"Voice settings created: {voice_settings}") # Create synthesis request logger.info("Creating synthesis request") synthesis_request = SpeechSynthesisRequest( text_content=text, # text is already a TextContent object voice_settings=voice_settings ) logger.info("Synthesis request created successfully") # Perform synthesis logger.info("Starting TTS synthesis") audio_content = tts_provider.synthesize(synthesis_request) logger.info(f"TTS synthesis completed, audio format: {audio_content.format}, data length: {len(audio_content.data)}") # Save output to file output_filename = f"output_{correlation_id}.{audio_content.format}" output_path = os.path.join(temp_dir, output_filename) logger.info(f"Saving audio to: {output_path}") with open(output_path, 'wb') as f: f.write(audio_content.data) # Track temp file for cleanup self._temp_files[output_path] = output_path logger.info( f"TTS completed successfully [correlation_id={correlation_id}, " f"output_file={output_path}]" ) return output_path except Exception as e: logger.error(f"TTS failed: {e} [correlation_id={correlation_id}]", exception=e) raise SpeechSynthesisException(f"Speech synthesis failed: {str(e)}") def _get_error_code_from_exception(self, exception: Exception) -> str: """ Get error code from exception type. Args: exception: Exception instance Returns: str: Error code """ if isinstance(exception, SpeechRecognitionException): return 'STT_ERROR' elif isinstance(exception, TranslationFailedException): return 'TRANSLATION_ERROR' elif isinstance(exception, SpeechSynthesisException): return 'TTS_ERROR' elif isinstance(exception, ValueError): return 'VALIDATION_ERROR' else: return 'SYSTEM_ERROR' def _cleanup_temp_files(self) -> None: """Cleanup tracked temporary files.""" for file_path in list(self._temp_files.keys()): try: if os.path.exists(file_path): os.remove(file_path) logger.info(f"Cleaned up temp file: {file_path}") except Exception as e: logger.warning(f"Failed to cleanup temp file {file_path}: {e}") finally: # Remove from tracking regardless of success self._temp_files.pop(file_path, None) def get_processing_status(self, correlation_id: str) -> Dict[str, Any]: """ Get processing status for a correlation ID. Args: correlation_id: Correlation ID to check Returns: Dict[str, Any]: Processing status information """ # This would be implemented with actual status tracking # For now, return basic info return { 'correlation_id': correlation_id, 'status': 'unknown', 'message': 'Status tracking not implemented' } def get_supported_configurations(self) -> Dict[str, Any]: """ Get supported configurations for the processing pipeline. Returns: Dict[str, Any]: Supported configurations """ return { 'asr_models': ['parakeet', 'whisper-small', 'whisper-medium', 'whisper-large'], 'voices': ['chatterbox'], 'languages': [ 'en', 'es', 'fr', 'de', 'it', 'pt', 'ru', 'ja', 'ko', 'zh', 'ar', 'hi', 'tr', 'pl', 'nl', 'sv', 'da', 'no', 'fi' ], 'audio_formats': self._config.get_processing_config()['supported_audio_formats'], 'max_file_size_mb': self._config.get_processing_config()['max_file_size_mb'], 'speed_range': {'min': 0.5, 'max': 2.0} } def cleanup(self) -> None: """Cleanup application service resources.""" logger.info("Cleaning up AudioProcessingApplicationService") # Cleanup temporary files self._cleanup_temp_files() logger.info("AudioProcessingApplicationService 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() def _perform_speech_recognition_with_recovery( self, audio: AudioContent, model: str, correlation_id: str ) -> TextContent: """ Perform speech-to-text recognition with retry and fallback. Args: audio: Audio content to transcribe model: STT model to use correlation_id: Correlation ID for tracking Returns: TextContent: Transcribed text Raises: SpeechRecognitionException: If all attempts fail """ context = LogContext( correlation_id=correlation_id, operation="speech_recognition", component="AudioProcessingApplicationService" ) # Configure retry for STT retry_config = RetryConfig( max_attempts=2, base_delay=1.0, retryable_exceptions=[SpeechRecognitionException, ConnectionError, TimeoutError] ) def stt_operation(*args, **kwargs): return self._perform_speech_recognition(audio, model, correlation_id) try: # Try with retry return self._recovery_manager.retry_with_backoff( stt_operation, retry_config, correlation_id ) except Exception as e: # Try fallback models if primary fails stt_config = self._config.get_stt_config() fallback_models = [m for m in stt_config['preferred_providers'] if m != model] if fallback_models: logger.warning( f"STT model {model} failed, trying fallbacks: {fallback_models}", context=context, exception=e ) fallback_funcs = [ lambda *args, m=fallback_model, **kwargs: self._perform_speech_recognition(audio, m, correlation_id) for fallback_model in fallback_models ] return self._recovery_manager.execute_with_fallback( stt_operation, fallback_funcs, correlation_id ) else: raise def _perform_translation_with_recovery( self, text: TextContent, source_language: Optional[str], target_language: str, correlation_id: str ) -> TextContent: """ Perform text translation with retry. Args: text: Text to translate source_language: Source language (optional, auto-detect if None) target_language: Target language correlation_id: Correlation ID for tracking Returns: TextContent: Translated text Raises: TranslationFailedException: If all attempts fail """ # Configure retry for translation retry_config = RetryConfig( max_attempts=3, base_delay=1.0, exponential_backoff=True, retryable_exceptions=[TranslationFailedException, ConnectionError, TimeoutError] ) def translation_operation(*args, **kwargs): return self._perform_translation(text, source_language, target_language, correlation_id) return self._recovery_manager.retry_with_backoff( translation_operation, retry_config, correlation_id ) def _perform_speech_synthesis_with_recovery( self, text: TextContent, voice: str, speed: float, language: str, temp_dir: str, correlation_id: str ) -> str: """ Perform text-to-speech synthesis with fallback providers. Args: text: Text to synthesize voice: Voice to use speed: Speech speed language: Target language temp_dir: Temporary directory for output correlation_id: Correlation ID for tracking Returns: str: Path to generated audio file Raises: SpeechSynthesisException: If all providers fail """ context = LogContext( correlation_id=correlation_id, operation="speech_synthesis", component="AudioProcessingApplicationService" ) logger.info(f"Starting TTS synthesis with recovery [correlation_id={correlation_id}]") logger.info(f"Parameters: voice={voice}, speed={speed}, language={language}") logger.info(f"Text type: {type(text)}, Text content type: {type(text.text) if hasattr(text, 'text') else 'N/A'}") def tts_operation(*args, **kwargs): logger.info(f"Executing TTS operation [correlation_id={correlation_id}]") try: result = self._perform_speech_synthesis(text, voice, speed, language, temp_dir, correlation_id) logger.info(f"TTS operation completed successfully [correlation_id={correlation_id}]") return result except Exception as e: logger.error(f"TTS operation failed: {str(e)} [correlation_id={correlation_id}]") raise try: # Try with circuit breaker protection logger.info(f"Attempting TTS with circuit breaker [correlation_id={correlation_id}]") return self._recovery_manager.execute_with_circuit_breaker( tts_operation, f"tts_{voice}", CircuitBreakerConfig(failure_threshold=3, recovery_timeout=30.0), correlation_id ) except Exception as e: logger.error(f"Primary TTS failed, trying fallbacks: {str(e)} [correlation_id={correlation_id}]", context=context, exception=e) # Try fallback TTS providers tts_config = self._config.get_tts_config() fallback_voices = [v for v in tts_config['preferred_providers'] if v != voice] if fallback_voices: logger.warning( f"TTS voice {voice} failed, trying fallbacks: {fallback_voices}", context=context, exception=e ) fallback_funcs = [ lambda *args, v=fallback_voice, **kwargs: self._perform_speech_synthesis( text, v, speed, language, temp_dir, correlation_id ) for fallback_voice in fallback_voices ] return self._recovery_manager.execute_with_fallback( tts_operation, fallback_funcs, correlation_id ) else: logger.error(f"No fallback voices available [correlation_id={correlation_id}]") raise