"""Processing Result Data Transfer Object""" from dataclasses import dataclass from typing import Optional, Dict, Any from datetime import datetime @dataclass class ProcessingResultDto: """DTO for pipeline output data Contains the results of processing audio through the STT -> Translation -> TTS pipeline. """ success: bool original_text: Optional[str] = None translated_text: Optional[str] = None audio_path: Optional[str] = None processing_time: float = 0.0 error_message: Optional[str] = None error_code: Optional[str] = None metadata: Optional[Dict[str, Any]] = None timestamp: Optional[datetime] = None def __post_init__(self): """Validate and set defaults after initialization""" self._validate() if self.metadata is None: self.metadata = {} if self.timestamp is None: self.timestamp = datetime.utcnow() def _validate(self): """Validate processing result data""" if not isinstance(self.success, bool): raise ValueError("Success must be a boolean value") if self.processing_time < 0: raise ValueError("Processing time cannot be negative") if self.success: # For successful processing, we should have some output if not self.original_text and not self.translated_text and not self.audio_path: raise ValueError("Successful processing must have at least one output (text or audio)") else: # For failed processing, we should have an error message if not self.error_message: raise ValueError("Failed processing must include an error message") # Validate error code format if provided if self.error_code: valid_error_codes = [ 'STT_ERROR', 'TRANSLATION_ERROR', 'TTS_ERROR', 'AUDIO_FORMAT_ERROR', 'VALIDATION_ERROR', 'SYSTEM_ERROR', 'TYPE_ERROR', 'TIMEOUT_ERROR', 'PERMISSION_ERROR', 'MEMORY_ERROR', 'UNKNOWN_ERROR' # Add missing error code for unexpected errors ] if self.error_code not in valid_error_codes: raise ValueError(f"Invalid error code: {self.error_code}. Valid codes: {valid_error_codes}") # Validate metadata if provided if self.metadata and not isinstance(self.metadata, dict): raise ValueError("Metadata must be a dictionary") @property def has_text_output(self) -> bool: """Check if result has text output""" return bool(self.original_text or self.translated_text) @property def has_audio_output(self) -> bool: """Check if result has audio output""" return bool(self.audio_path) @property def is_complete(self) -> bool: """Check if processing is complete (success or failure with error)""" return self.success or bool(self.error_message) def add_metadata(self, key: str, value: Any) -> None: """Add metadata entry""" if self.metadata is None: self.metadata = {} self.metadata[key] = value def get_metadata(self, key: str, default: Any = None) -> Any: """Get metadata value""" if self.metadata is None: return default return self.metadata.get(key, default) def to_dict(self) -> dict: """Convert to dictionary representation""" return { 'success': self.success, 'original_text': self.original_text, 'translated_text': self.translated_text, 'audio_path': self.audio_path, 'processing_time': self.processing_time, 'error_message': self.error_message, 'error_code': self.error_code, 'metadata': self.metadata or {}, 'timestamp': self.timestamp.isoformat() if self.timestamp else None, 'has_text_output': self.has_text_output, 'has_audio_output': self.has_audio_output, 'is_complete': self.is_complete } @classmethod def success_result(cls, original_text: str = None, translated_text: str = None, audio_path: str = None, processing_time: float = 0.0, metadata: Dict[str, Any] = None) -> 'ProcessingResultDto': """Create a successful processing result""" return cls( success=True, original_text=original_text, translated_text=translated_text, audio_path=audio_path, processing_time=processing_time, metadata=metadata ) @classmethod def error_result(cls, error_message: str, error_code: str = None, processing_time: float = 0.0, metadata: Dict[str, Any] = None) -> 'ProcessingResultDto': """Create a failed processing result""" return cls( success=False, error_message=error_message, error_code=error_code, processing_time=processing_time, metadata=metadata ) @classmethod def from_dict(cls, data: dict) -> 'ProcessingResultDto': """Create instance from dictionary""" timestamp = None if data.get('timestamp'): timestamp = datetime.fromisoformat(data['timestamp'].replace('Z', '+00:00')) return cls( success=data['success'], original_text=data.get('original_text'), translated_text=data.get('translated_text'), audio_path=data.get('audio_path'), processing_time=data.get('processing_time', 0.0), error_message=data.get('error_message'), error_code=data.get('error_code'), metadata=data.get('metadata'), timestamp=timestamp )