Spaces:
Sleeping
Sleeping
| """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" | |
| ) |