AI-Talent-Force/dev_caio
0
1"""2ShortSmith v2 - Helper Utilities3 4Common utility functions for file handling, validation, and data manipulation.5"""6 7import os8import shutil9import tempfile10import uuid11from pathlib import Path12from typing import Optional, List, Tuple, Union13from dataclasses import dataclass14 15from utils.logger import get_logger16 17logger = get_logger("utils.helpers")18 19# Supported file formats20SUPPORTED_VIDEO_FORMATS = {".mp4", ".avi", ".mov", ".mkv", ".webm", ".flv", ".wmv", ".m4v"}21SUPPORTED_IMAGE_FORMATS = {".jpg", ".jpeg", ".png", ".webp", ".bmp", ".gif"}22SUPPORTED_AUDIO_FORMATS = {".mp3", ".wav", ".aac", ".flac", ".ogg", ".m4a"}23 24 25@dataclass26class ValidationResult:27 """Result of file validation."""28 is_valid: bool29 error_message: Optional[str] = None30 file_path: Optional[Path] = None31 file_size: int = 032 33 34class FileValidationError(Exception):35 """Exception raised for file validation errors."""36 pass37 38 39class VideoProcessingError(Exception):40 """Exception raised for video processing errors."""41 pass42 43 44class ModelLoadError(Exception):45 """Exception raised when model loading fails."""46 pass47 48 49class InferenceError(Exception):50 """Exception raised during model inference."""51 pass52 53 54def validate_video_file(55 file_path: Union[str, Path],56 max_size_mb: float = 500.0,57 check_exists: bool = True,58) -> ValidationResult:59 """60 Validate a video file for processing.61 62 Args:63 file_path: Path to the video file64 max_size_mb: Maximum allowed file size in megabytes65 check_exists: Whether to check if file exists66 67 Returns:68 ValidationResult with validation status and details69 70 Raises:71 FileValidationError: If validation fails and raise_on_error is True72 """73 try:74 path = Path(file_path)75 76 # Check existence77 if check_exists and not path.exists():78 return ValidationResult(79 is_valid=False,80 error_message=f"Video file not found: {path}"81 )82 83 # Check extension84 if path.suffix.lower() not in SUPPORTED_VIDEO_FORMATS:85 return ValidationResult(86 is_valid=False,87 error_message=f"Unsupported video format: {path.suffix}. "88 f"Supported: {', '.join(SUPPORTED_VIDEO_FORMATS)}"89 )90 91 # Check file size92 if check_exists:93 file_size = path.stat().st_size94 size_mb = file_size / (1024 * 1024)95 96 if size_mb > max_size_mb:97 return ValidationResult(98 is_valid=False,99 error_message=f"Video file too large: {size_mb:.1f}MB (max: {max_size_mb}MB)",100 file_size=file_size101 )102 else:103 file_size = 0104 105 logger.debug(f"Video file validated: {path}")106 return ValidationResult(107 is_valid=True,108 file_path=path,109 file_size=file_size110 )111 112 except Exception as e:113 logger.error(f"Error validating video file {file_path}: {e}")114 return ValidationResult(115 is_valid=False,116 error_message=f"Validation error: {str(e)}"117 )118 119 120def validate_image_file(121 file_path: Union[str, Path],122 max_size_mb: float = 10.0,123 check_exists: bool = True,124) -> ValidationResult:125 """126 Validate an image file (e.g., reference image for person detection).127 128 Args:129 file_path: Path to the image file130 max_size_mb: Maximum allowed file size in megabytes131 check_exists: Whether to check if file exists132 133 Returns:134 ValidationResult with validation status and details135 """136 try:137 path = Path(file_path)138 139 # Check existence140 if check_exists and not path.exists():141 return ValidationResult(142 is_valid=False,143 error_message=f"Image file not found: {path}"144 )145 146 # Check extension147 if path.suffix.lower() not in SUPPORTED_IMAGE_FORMATS:148 return ValidationResult(149 is_valid=False,150 error_message=f"Unsupported image format: {path.suffix}. "151 f"Supported: {', '.join(SUPPORTED_IMAGE_FORMATS)}"152 )153 154 # Check file size155 if check_exists:156 file_size = path.stat().st_size157 size_mb = file_size / (1024 * 1024)158 159 if size_mb > max_size_mb:160 return ValidationResult(161 is_valid=False,162 error_message=f"Image file too large: {size_mb:.1f}MB (max: {max_size_mb}MB)",163 file_size=file_size164 )165 else:166 file_size = 0167 168 logger.debug(f"Image file validated: {path}")169 return ValidationResult(170 is_valid=True,171 file_path=path,172 file_size=file_size173 )174 175 except Exception as e:176 logger.error(f"Error validating image file {file_path}: {e}")177 return ValidationResult(178 is_valid=False,179 error_message=f"Validation error: {str(e)}"180 )181 182 183def get_temp_dir(prefix: str = "shortsmith_") -> Path:184 """185 Create a temporary directory for processing.186 187 Args:188 prefix: Prefix for the temp directory name189 190 Returns:191 Path to the created temporary directory192 193 Raises:194 OSError: If directory creation fails195 """196 try:197 # Use system temp dir or custom if configured198 base_temp = tempfile.gettempdir()199 unique_id = str(uuid.uuid4())[:8]200 temp_dir = Path(base_temp) / f"{prefix}{unique_id}"201 temp_dir.mkdir(parents=True, exist_ok=True)202 203 logger.debug(f"Created temp directory: {temp_dir}")204 return temp_dir205 206 except Exception as e:207 logger.error(f"Failed to create temp directory: {e}")208 raise OSError(f"Could not create temporary directory: {e}") from e209 210 211def cleanup_temp_files(212 temp_dir: Union[str, Path],213 ignore_errors: bool = True214) -> bool:215 """216 Clean up temporary files and directories.217 218 Args:219 temp_dir: Path to the temporary directory to clean220 ignore_errors: Whether to ignore cleanup errors221 222 Returns:223 True if cleanup was successful, False otherwise224 """225 try:226 path = Path(temp_dir)227 if path.exists():228 shutil.rmtree(path, ignore_errors=ignore_errors)229 logger.debug(f"Cleaned up temp directory: {path}")230 return True231 232 except Exception as e:233 logger.warning(f"Failed to cleanup temp directory {temp_dir}: {e}")234 return False235 236 237def format_duration(seconds: float) -> str:238 """239 Format duration in seconds to human-readable string.240 241 Args:242 seconds: Duration in seconds243 244 Returns:245 Formatted string (e.g., "1:23:45" or "5:30")246 """247 if seconds < 0:248 return "0:00"249 250 hours = int(seconds // 3600)251 minutes = int((seconds % 3600) // 60)252 secs = int(seconds % 60)253 254 if hours > 0:255 return f"{hours}:{minutes:02d}:{secs:02d}"256 else:257 return f"{minutes}:{secs:02d}"258 259 260def format_timestamp(seconds: float, include_ms: bool = False) -> str:261 """262 Format timestamp for display.263 264 Args:265 seconds: Timestamp in seconds266 include_ms: Whether to include milliseconds267 268 Returns:269 Formatted timestamp string270 """271 hours = int(seconds // 3600)272 minutes = int((seconds % 3600) // 60)273 secs = seconds % 60274 275 if include_ms:276 if hours > 0:277 return f"{hours}:{minutes:02d}:{secs:06.3f}"278 else:279 return f"{minutes}:{secs:06.3f}"280 else:281 secs = int(secs)282 if hours > 0:283 return f"{hours}:{minutes:02d}:{secs:02d}"284 else:285 return f"{minutes}:{secs:02d}"286 287 288def safe_divide(289 numerator: float,290 denominator: float,291 default: float = 0.0292) -> float:293 """294 Safely divide two numbers, returning default if denominator is zero.295 296 Args:297 numerator: The numerator298 denominator: The denominator299 default: Value to return if denominator is zero300 301 Returns:302 Result of division or default value303 """304 if denominator == 0:305 return default306 return numerator / denominator307 308 309def clamp(310 value: float,311 min_value: float,312 max_value: float313) -> float:314 """315 Clamp a value to a specified range.316 317 Args:318 value: The value to clamp319 min_value: Minimum allowed value320 max_value: Maximum allowed value321 322 Returns:323 Clamped value324 """325 return max(min_value, min(value, max_value))326 327 328def normalize_scores(scores: List[float]) -> List[float]:329 """330 Normalize a list of scores to [0, 1] range.331 332 Args:333 scores: List of raw scores334 335 Returns:336 Normalized scores337 """338 if not scores:339 return []340 341 min_score = min(scores)342 max_score = max(scores)343 score_range = max_score - min_score344 345 if score_range == 0:346 return [0.5] * len(scores)347 348 return [(s - min_score) / score_range for s in scores]349 350 351def batch_list(items: List, batch_size: int) -> List[List]:352 """353 Split a list into batches of specified size.354 355 Args:356 items: List to split357 batch_size: Size of each batch358 359 Returns:360 List of batches361 """362 return [items[i:i + batch_size] for i in range(0, len(items), batch_size)]363 364 365def merge_overlapping_segments(366 segments: List[Tuple[float, float]],367 min_gap: float = 0.0368) -> List[Tuple[float, float]]:369 """370 Merge overlapping or closely spaced time segments.371 372 Args:373 segments: List of (start, end) tuples374 min_gap: Minimum gap to keep segments separate375 376 Returns:377 List of merged segments378 """379 if not segments:380 return []381 382 # Sort by start time383 sorted_segments = sorted(segments, key=lambda x: x[0])384 merged = [sorted_segments[0]]385 386 for start, end in sorted_segments[1:]:387 last_start, last_end = merged[-1]388 389 # Check if segments overlap or are close enough390 if start <= last_end + min_gap:391 # Merge by extending the end392 merged[-1] = (last_start, max(last_end, end))393 else:394 merged.append((start, end))395 396 return merged397 398 399def ensure_dir(path: Union[str, Path]) -> Path:400 """401 Ensure a directory exists, creating it if necessary.402 403 Args:404 path: Path to the directory405 406 Returns:407 Path object for the directory408 """409 path = Path(path)410 path.mkdir(parents=True, exist_ok=True)411 return path412 413 414def get_unique_filename(415 directory: Union[str, Path],416 base_name: str,417 extension: str418) -> Path:419 """420 Generate a unique filename in the given directory.421 422 Args:423 directory: Directory for the file424 base_name: Base name for the file425 extension: File extension (with or without dot)426 427 Returns:428 Path to a unique file429 """430 directory = Path(directory)431 extension = extension if extension.startswith(".") else f".{extension}"432 433 # Try base name first434 candidate = directory / f"{base_name}{extension}"435 if not candidate.exists():436 return candidate437 438 # Add counter439 counter = 1440 while True:441 candidate = directory / f"{base_name}_{counter}{extension}"442 if not candidate.exists():443 return candidate444 counter += 1445 446 447# Export all public functions448__all__ = [449 "SUPPORTED_VIDEO_FORMATS",450 "SUPPORTED_IMAGE_FORMATS",451 "SUPPORTED_AUDIO_FORMATS",452 "ValidationResult",453 "FileValidationError",454 "VideoProcessingError",455 "ModelLoadError",456 "InferenceError",457 "validate_video_file",458 "validate_image_file",459 "get_temp_dir",460 "cleanup_temp_files",461 "format_duration",462 "format_timestamp",463 "safe_divide",464 "clamp",465 "normalize_scores",466 "batch_list",467 "merge_overlapping_segments",468 "ensure_dir",469 "get_unique_filename",470]471