"""Enhanced error handling and recovery mechanisms.""" import logging import traceback from enum import Enum from pathlib import Path from typing import Dict, List, Optional, Any from contextlib import contextmanager logger = logging.getLogger(__name__) class ErrorSeverity(Enum): """Error severity levels for classification.""" LOW = "low" # Warnings, non-critical issues MEDIUM = "medium" # Recoverable errors HIGH = "high" # Critical errors requiring user intervention FATAL = "fatal" # Unrecoverable errors class CaptionToolError(Exception): """Base exception class for caption tool errors.""" def __init__(self, message: str, severity: ErrorSeverity = ErrorSeverity.MEDIUM, suggestions: Optional[List[str]] = None): super().__init__(message) self.severity = severity self.suggestions = suggestions or [] def get_user_message(self) -> str: """Get user-friendly error message with suggestions.""" msg = f"āŒ {self.severity.value.upper()}: {str(self)}" if self.suggestions: msg += "\n\nšŸ’” Suggestions:" for i, suggestion in enumerate(self.suggestions, 1): msg += f"\n {i}. {suggestion}" return msg class AudioValidationError(CaptionToolError): """Errors related to audio file validation.""" pass class ScriptValidationError(CaptionToolError): """Errors related to script file validation.""" pass class AlignmentError(CaptionToolError): """Errors during the alignment process.""" pass class ModelError(CaptionToolError): """Errors related to model loading/downloading.""" pass class ErrorRecovery: """Error recovery and retry mechanisms.""" @staticmethod @contextmanager def retry_on_failure(max_retries: int = 3, delay: float = 1.0, exceptions: tuple = (Exception,)): """Retry operation with exponential backoff.""" import time for attempt in range(max_retries + 1): try: yield attempt break except exceptions as e: if attempt == max_retries: raise wait_time = delay * (2 ** attempt) logger.warning(f"Attempt {attempt + 1} failed: {e}. " f"Retrying in {wait_time}s...") time.sleep(wait_time) @staticmethod def diagnose_alignment_failure(audio_path: Path, script_path: Path) -> List[str]: """Diagnose common alignment failure causes.""" suggestions = [] # Check file sizes audio_size = audio_path.stat().st_size script_size = script_path.stat().st_size if audio_size < 1024: # Very small audio file suggestions.append("Audio file seems too small - ensure it contains speech") if script_size < 10: # Very small script suggestions.append("Script file seems too short - ensure it contains text") # Check script content try: with open(script_path, 'r', encoding='utf-8') as f: content = f.read().strip() if len(content.split()) < 5: suggestions.append("Script contains very few words - alignment may be unreliable") if not any('\u0600' <= c <= '\u06FF' for c in content): suggestions.append("Script contains no Arabic text - ensure language setting is correct") except Exception: suggestions.append("Cannot read script file - check encoding (should be UTF-8)") # Audio duration check try: import subprocess cmd = ['ffprobe', '-v', 'quiet', '-show_entries', 'format=duration', '-of', 'default=noprint_wrappers=1:nokey=1', str(audio_path)] result = subprocess.run(cmd, capture_output=True, text=True, timeout=5) duration = float(result.stdout.strip()) if duration < 1.0: suggestions.append("Audio is very short - ensure it contains sufficient speech") elif duration > 300: # 5 minutes suggestions.append("Audio is very long - consider splitting into smaller segments") except Exception: suggestions.append("Cannot determine audio duration - ensure file is valid") return suggestions @staticmethod def suggest_recovery_actions(error: Exception, context: Dict[str, Any]) -> List[str]: """Suggest recovery actions based on error type and context.""" suggestions = [] error_str = str(error).lower() if "memory" in error_str or "out of memory" in error_str: suggestions.extend([ "Free up system memory by closing other applications", "Try processing smaller audio segments", "Use sentence-level alignment instead of word-level", "Restart the script to clear memory" ]) elif "network" in error_str or "connection" in error_str or "download" in error_str: suggestions.extend([ "Check your internet connection", "Try again in a few minutes (server may be busy)", "Use a VPN if in a restricted network", "Clear the model cache directory and retry" ]) elif "permission" in error_str or "access" in error_str: suggestions.extend([ "Check file permissions for input/output directories", "Run as administrator if necessary", "Ensure output directory is writable" ]) elif "format" in error_str or "codec" in error_str: suggestions.extend([ "Convert audio to a supported format (MP3, WAV, M4A)", "Ensure audio has speech content (not just music/silence)", "Check if audio file is corrupted" ]) elif "alignment failed" in error_str: audio_path = context.get('audio_path') script_path = context.get('script_path') if audio_path and script_path: suggestions.extend( ErrorRecovery.diagnose_alignment_failure(audio_path, script_path) ) return suggestions class ErrorLogger: """Enhanced error logging with context.""" def __init__(self, log_file: Optional[Path] = None): self.log_file = log_file or Path("caption_tool_errors.log") def log_error(self, error: Exception, context: Dict[str, Any] = None): """Log error with full context and stack trace.""" context = context or {} error_info = { "error_type": type(error).__name__, "error_message": str(error), "context": context, "stack_trace": traceback.format_exc() } # Log to file try: with open(self.log_file, 'a', encoding='utf-8') as f: import json import datetime log_entry = { "timestamp": datetime.datetime.now().isoformat(), **error_info } f.write(json.dumps(log_entry, ensure_ascii=False, indent=2) + "\n\n") except Exception as e: logger.error(f"Failed to write error log: {e}") # Log to console logger.error(f"Error: {error_info['error_type']}: {error_info['error_message']}") if context: logger.error(f"Context: {context}") def handle_graceful_shutdown(error: Exception, context: Dict[str, Any] = None) -> str: """Handle graceful shutdown with user-friendly error reporting.""" context = context or {} # Log the error error_logger = ErrorLogger() error_logger.log_error(error, context) # Determine error type and provide appropriate response if isinstance(error, CaptionToolError): return error.get_user_message() # For other exceptions, create a generic CaptionToolError suggestions = ErrorRecovery.suggest_recovery_actions(error, context) if "memory" in str(error).lower(): severity = ErrorSeverity.HIGH elif "network" in str(error).lower() or "download" in str(error).lower(): severity = ErrorSeverity.MEDIUM else: severity = ErrorSeverity.HIGH wrapped_error = CaptionToolError( message=str(error), severity=severity, suggestions=suggestions ) return wrapped_error.get_user_message()