"""Domain validation service.""" import re from typing import List, Optional from ..entities.video import Video from ..value_objects.audio_format import AudioFormat from ..value_objects.audio_quality import AudioQuality from ..value_objects.file_size import FileSize from ..exceptions.domain_exceptions import ( ValidationError, FileSizeExceededError, InvalidVideoFormatError, InvalidAudioFormatError, InvalidExternalJobIdFormatError ) class ValidationService: """Service for validating domain rules.""" def __init__(self, max_file_size_mb: float, supported_video_formats: List[str], supported_audio_formats: List[str]): self.max_file_size_mb = max_file_size_mb self.supported_video_formats = supported_video_formats self.supported_audio_formats = supported_audio_formats def validate_video(self, video: Video) -> None: """Validate video entity against business rules.""" # Check file size if not video.size.is_within_limit(self.max_file_size_mb): raise FileSizeExceededError(video.size.megabytes, self.max_file_size_mb) # Check format extension = video.get_extension() if extension not in self.supported_video_formats: raise InvalidVideoFormatError(extension, self.supported_video_formats) def validate_extraction_request(self, video: Video, format: str, quality: str) -> None: """Validate complete extraction request.""" # Validate video self.validate_video(video) # Validate audio format try: AudioFormat(format) except InvalidAudioFormatError: raise InvalidAudioFormatError(format, self.supported_audio_formats) # Validate quality AudioQuality(quality) # Will raise if invalid def validate_external_job_id(self, external_job_id: Optional[str]) -> None: """Validate external job ID format. Args: external_job_id: External job ID to validate Raises: InvalidExternalJobIdFormatError: If external job ID format is invalid """ if external_job_id is None or external_job_id == "": return # Optional field, empty/None is valid # Check length if len(external_job_id) > 50: raise InvalidExternalJobIdFormatError( external_job_id, "Must be 50 characters or less" ) if len(external_job_id) < 1: raise InvalidExternalJobIdFormatError( external_job_id, "Cannot be empty if provided" ) # Check format: alphanumeric, underscores, and hyphens only if not re.match(r'^[a-zA-Z0-9_-]+$', external_job_id): raise InvalidExternalJobIdFormatError( external_job_id, "Must contain only alphanumeric characters, underscores, and hyphens" ) def can_process_directly(self, video: Video, threshold_mb: float) -> bool: """Check if video can be processed directly (not async).""" return not video.is_large_file(threshold_mb) def validate_time_format(self, time_str: str) -> float: """Validate and convert HH:MM:SS format to seconds. Args: time_str: Time string in HH:MM:SS format Returns: float: Time in seconds Raises: ValidationError: If format is invalid """ if not time_str: raise ValidationError("Time string cannot be empty") # Pattern for HH:MM:SS format pattern = r'^(\d{1,2}):(\d{2}):(\d{2})$' match = re.match(pattern, time_str.strip()) if not match: raise ValidationError( f"Invalid time format '{time_str}'. Expected format: HH:MM:SS (e.g., 01:23:45)" ) hours, minutes, seconds = map(int, match.groups()) # Validate ranges if minutes >= 60: raise ValidationError(f"Invalid minutes '{minutes}'. Must be 0-59") if seconds >= 60: raise ValidationError(f"Invalid seconds '{seconds}'. Must be 0-59") # Convert to total seconds total_seconds = hours * 3600 + minutes * 60 + seconds if total_seconds < 0: raise ValidationError("Time cannot be negative") return float(total_seconds) def validate_time_range(self, start_seconds: Optional[float], end_seconds: Optional[float], audio_duration: float) -> None: """Validate that time range is valid for the audio duration. Args: start_seconds: Start time in seconds (None means start from beginning) end_seconds: End time in seconds (None means end at audio end) audio_duration: Total audio duration in seconds Raises: ValidationError: If time range is invalid """ if audio_duration <= 0: raise ValidationError("Audio duration must be positive") # Validate start time if start_seconds is not None: if start_seconds < 0: raise ValidationError("Start time cannot be negative") if start_seconds >= audio_duration: raise ValidationError( f"Start time {start_seconds:.1f}s exceeds audio duration {audio_duration:.1f}s" ) # Validate end time if end_seconds is not None: if end_seconds < 0: raise ValidationError("End time cannot be negative") if end_seconds > audio_duration: raise ValidationError( f"End time {end_seconds:.1f}s exceeds audio duration {audio_duration:.1f}s" ) # Validate range relationship if start_seconds is not None and end_seconds is not None: if start_seconds >= end_seconds: raise ValidationError( f"Start time {start_seconds:.1f}s must be less than end time {end_seconds:.1f}s" ) # Check for minimum segment duration (at least 1 second) if end_seconds - start_seconds < 1.0: raise ValidationError( "Audio segment must be at least 1 second long" )