Update utils/utils.py

utils/utils.py

@@ -1,6 +1,6 @@
 """
 Unified Utilities Module for BackgroundFX Pro
-Combines FileManager, VideoUtils, ImageUtils, and CV utilities
+Combines FileManager, VideoUtils, ImageUtils, ValidationUtils, and CV utilities
 """
 
 # Set OMP_NUM_THREADS at the very beginning to prevent libgomp errors
@@ -17,6 +17,7 @@
 from datetime import datetime
 import subprocess
 import time
+import re
 
 import cv2
 import numpy as np
@@ -109,6 +110,381 @@ class BackgroundReplacementError(Exception):
     """Custom exception for background replacement failures"""
     pass
 
+# ============================================================================
+# VALIDATION UTILS CLASS
+# ============================================================================
+
+class ValidationUtils:
+    """Validation utilities for BackgroundFX Pro application."""
+    
+    # Supported formats
+    SUPPORTED_VIDEO_FORMATS = {'.mp4', '.avi', '.mov', '.mkv', '.webm', '.flv', '.wmv', '.m4v'}
+    SUPPORTED_IMAGE_FORMATS = {'.jpg', '.jpeg', '.png', '.bmp', '.tiff', '.webp'}
+    
+    # Size limits (in bytes)
+    MAX_VIDEO_SIZE = 500 * 1024 * 1024  # 500MB
+    MAX_IMAGE_SIZE = 50 * 1024 * 1024   # 50MB
+    MIN_VIDEO_SIZE = 1024                # 1KB (to avoid empty files)
+    
+    # Video constraints
+    MAX_VIDEO_DURATION = 300  # 5 minutes in seconds
+    MIN_VIDEO_DURATION = 1    # 1 second minimum
+    MAX_RESOLUTION = (3840, 2160)  # 4K
+    MIN_RESOLUTION = (320, 240)    # Minimum reasonable resolution
+    MAX_FPS = 120
+    MIN_FPS = 10
+    
+    @staticmethod
+    def validate_video_file(file_path, check_content=False):
+        """
+        Validate video file for processing.
+        
+        Args:
+            file_path: Path to the video file
+            check_content: Whether to perform deep content validation
+            
+        Returns:
+            tuple: (is_valid, error_message)
+        """
+        from pathlib import Path
+        
+        if not file_path:
+            return False, "No file path provided"
+        
+        path = Path(file_path)
+        
+        # Check if file exists
+        if not path.exists():
+            return False, f"File not found: {file_path}"
+        
+        # Check file extension
+        if path.suffix.lower() not in ValidationUtils.SUPPORTED_VIDEO_FORMATS:
+            return False, f"Unsupported video format: {path.suffix}. Supported formats: {', '.join(ValidationUtils.SUPPORTED_VIDEO_FORMATS)}"
+        
+        # Check file size
+        file_size = path.stat().st_size
+        if file_size > ValidationUtils.MAX_VIDEO_SIZE:
+            size_mb = file_size / (1024 * 1024)
+            return False, f"Video file too large: {size_mb:.1f}MB (max: {ValidationUtils.MAX_VIDEO_SIZE / (1024 * 1024):.0f}MB)"
+        
+        if file_size < ValidationUtils.MIN_VIDEO_SIZE:
+            return False, "Video file appears to be empty or corrupted"
+        
+        # Deep content validation if requested
+        if check_content:
+            try:
+                cap = cv2.VideoCapture(str(file_path))
+                
+                if not cap.isOpened():
+                    return False, "Unable to open video file - may be corrupted"
+                
+                # Get video properties
+                fps = cap.get(cv2.CAP_PROP_FPS)
+                frame_count = cap.get(cv2.CAP_PROP_FRAME_COUNT)
+                width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
+                height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
+                
+                # Calculate duration
+                duration = frame_count / fps if fps > 0 else 0
+                
+                cap.release()
+                
+                # Validate properties
+                if duration > ValidationUtils.MAX_VIDEO_DURATION:
+                    return False, f"Video too long: {duration:.1f}s (max: {ValidationUtils.MAX_VIDEO_DURATION}s)"
+                
+                if duration < ValidationUtils.MIN_VIDEO_DURATION:
+                    return False, f"Video too short: {duration:.1f}s (min: {ValidationUtils.MIN_VIDEO_DURATION}s)"
+                
+                if width > ValidationUtils.MAX_RESOLUTION[0] or height > ValidationUtils.MAX_RESOLUTION[1]:
+                    return False, f"Video resolution too high: {width}x{height} (max: {ValidationUtils.MAX_RESOLUTION[0]}x{ValidationUtils.MAX_RESOLUTION[1]})"
+                
+                if width < ValidationUtils.MIN_RESOLUTION[0] or height < ValidationUtils.MIN_RESOLUTION[1]:
+                    return False, f"Video resolution too low: {width}x{height} (min: {ValidationUtils.MIN_RESOLUTION[0]}x{ValidationUtils.MIN_RESOLUTION[1]})"
+                
+                if fps > ValidationUtils.MAX_FPS:
+                    return False, f"Frame rate too high: {fps:.1f} fps (max: {ValidationUtils.MAX_FPS} fps)"
+                
+                if fps < ValidationUtils.MIN_FPS:
+                    return False, f"Frame rate too low: {fps:.1f} fps (min: {ValidationUtils.MIN_FPS} fps)"
+                
+            except Exception as e:
+                return False, f"Error validating video content: {str(e)}"
+        
+        return True, "Video file is valid"
+    
+    @staticmethod
+    def validate_image_file(file_path, check_content=False):
+        """
+        Validate image file for background replacement.
+        
+        Args:
+            file_path: Path to the image file
+            check_content: Whether to perform deep content validation
+            
+        Returns:
+            tuple: (is_valid, error_message)
+        """
+        from pathlib import Path
+        
+        if not file_path:
+            return False, "No file path provided"
+        
+        path = Path(file_path)
+        
+        # Check if file exists
+        if not path.exists():
+            return False, f"File not found: {file_path}"
+        
+        # Check file extension
+        if path.suffix.lower() not in ValidationUtils.SUPPORTED_IMAGE_FORMATS:
+            return False, f"Unsupported image format: {path.suffix}. Supported formats: {', '.join(ValidationUtils.SUPPORTED_IMAGE_FORMATS)}"
+        
+        # Check file size
+        file_size = path.stat().st_size
+        if file_size > ValidationUtils.MAX_IMAGE_SIZE:
+            size_mb = file_size / (1024 * 1024)
+            return False, f"Image file too large: {size_mb:.1f}MB (max: {ValidationUtils.MAX_IMAGE_SIZE / (1024 * 1024):.0f}MB)"
+        
+        # Deep content validation if requested
+        if check_content:
+            try:
+                img = cv2.imread(str(file_path))
+                if img is None:
+                    return False, "Unable to read image file - may be corrupted"
+                
+                height, width = img.shape[:2]
+                
+                # Check dimensions
+                if width > ValidationUtils.MAX_RESOLUTION[0] or height > ValidationUtils.MAX_RESOLUTION[1]:
+                    return False, f"Image resolution too high: {width}x{height} (max: {ValidationUtils.MAX_RESOLUTION[0]}x{ValidationUtils.MAX_RESOLUTION[1]})"
+                
+                if width < ValidationUtils.MIN_RESOLUTION[0] or height < ValidationUtils.MIN_RESOLUTION[1]:
+                    return False, f"Image resolution too low: {width}x{height} (min: {ValidationUtils.MIN_RESOLUTION[0]}x{ValidationUtils.MIN_RESOLUTION[1]})"
+                
+            except Exception as e:
+                return False, f"Error validating image content: {str(e)}"
+        
+        return True, "Image file is valid"
+    
+    @staticmethod
+    def validate_processing_params(params):
+        """
+        Validate processing parameters.
+        
+        Args:
+            params: Dictionary of processing parameters
+            
+        Returns:
+            tuple: (is_valid, error_message)
+        """
+        if not params:
+            return False, "No parameters provided"
+        
+        # Validate confidence threshold
+        if 'confidence_threshold' in params:
+            conf = params['confidence_threshold']
+            if not isinstance(conf, (int, float)):
+                return False, "Confidence threshold must be a number"
+            if conf < 0 or conf > 1:
+                return False, "Confidence threshold must be between 0 and 1"
+        
+        # Validate mask dilation
+        if 'mask_dilation' in params:
+            dilation = params['mask_dilation']
+            if not isinstance(dilation, int):
+                return False, "Mask dilation must be an integer"
+            if dilation < 0 or dilation > 50:
+                return False, "Mask dilation must be between 0 and 50"
+        
+        # Validate edge smoothing
+        if 'edge_smoothing' in params:
+            smooth = params['edge_smoothing']
+            if not isinstance(smooth, int):
+                return False, "Edge smoothing must be an integer"
+            if smooth < 0 or smooth > 100:
+                return False, "Edge smoothing must be between 0 and 100"
+        
+        # Validate color adjustment
+        if 'color_adjustment' in params:
+            color_adj = params['color_adjustment']
+            if not isinstance(color_adj, bool):
+                return False, "Color adjustment must be a boolean"
+        
+        # Validate output quality
+        if 'output_quality' in params:
+            quality = params['output_quality']
+            if not isinstance(quality, int):
+                return False, "Output quality must be an integer"
+            if quality < 1 or quality > 100:
+                return False, "Output quality must be between 1 and 100"
+        
+        # Validate processing method
+        if 'processing_method' in params:
+            method = params['processing_method']
+            valid_methods = {'sam2', 'matanyone', 'cv_fallback', 'auto'}
+            if method not in valid_methods:
+                return False, f"Invalid processing method. Must be one of: {', '.join(valid_methods)}"
+        
+        return True, "Parameters are valid"
+    
+    @staticmethod
+    def validate_output_path(output_path, create_dirs=False):
+        """
+        Validate output path for saving results.
+        
+        Args:
+            output_path: Path where output will be saved
+            create_dirs: Whether to create directories if they don't exist
+            
+        Returns:
+            tuple: (is_valid, error_message)
+        """
+        from pathlib import Path
+        
+        if not output_path:
+            return False, "No output path provided"
+        
+        path = Path(output_path)
+        parent_dir = path.parent
+        
+        # Check if parent directory exists
+        if not parent_dir.exists():
+            if create_dirs:
+                try:
+                    parent_dir.mkdir(parents=True, exist_ok=True)
+                except Exception as e:
+                    return False, f"Failed to create output directory: {str(e)}"
+            else:
+                return False, f"Output directory does not exist: {parent_dir}"
+        
+        # Check write permissions
+        if not os.access(parent_dir, os.W_OK):
+            return False, f"No write permission for directory: {parent_dir}"
+        
+        # Check if file already exists
+        if path.exists():
+            if not os.access(path, os.W_OK):
+                return False, f"Cannot overwrite existing file: {output_path}"
+        
+        return True, "Output path is valid"
+    
+    @staticmethod
+    def sanitize_filename(filename):
+        """
+        Sanitize filename to be safe for filesystem.
+        
+        Args:
+            filename: Original filename
+            
+        Returns:
+            str: Sanitized filename
+        """
+        from pathlib import Path
+        
+        # Get the stem and suffix separately
+        path = Path(filename)
+        stem = path.stem
+        suffix = path.suffix
+        
+        # Remove or replace invalid characters
+        # Keep only alphanumeric, dash, underscore, and dot
+        stem = re.sub(r'[^\w\-_.]', '_', stem)
+        
+        # Remove multiple underscores
+        stem = re.sub(r'_+', '_', stem)
+        
+        # Remove leading/trailing underscores
+        stem = stem.strip('_')
+        
+        # Ensure filename is not empty
+        if not stem:
+            stem = 'output'
+        
+        # Limit length (keep it reasonable for most filesystems)
+        max_length = 200
+        if len(stem) > max_length:
+            stem = stem[:max_length]
+        
+        return f"{stem}{suffix}"
+    
+    @staticmethod
+    def validate_memory_available(required_mb=1000):
+        """
+        Check if sufficient memory is available.
+        
+        Args:
+            required_mb: Required memory in megabytes
+            
+        Returns:
+            tuple: (is_sufficient, available_mb, error_message)
+        """
+        try:
+            import psutil
+            
+            mem = psutil.virtual_memory()
+            available_mb = mem.available / (1024 * 1024)
+            
+            if available_mb < required_mb:
+                return False, available_mb, f"Insufficient memory: {available_mb:.0f}MB available, {required_mb:.0f}MB required"
+            
+            return True, available_mb, f"Sufficient memory available: {available_mb:.0f}MB"
+            
+        except ImportError:
+            # If psutil not available, assume sufficient memory
+            return True, -1, "Memory check skipped (psutil not available)"
+        except Exception as e:
+            return True, -1, f"Memory check failed: {str(e)}"
+    
+    @staticmethod
+    def validate_gpu_available():
+        """
+        Check if GPU is available for processing.
+        
+        Returns:
+            tuple: (is_available, device_info)
+        """
+        try:
+            if torch.cuda.is_available():
+                device_name = torch.cuda.get_device_name(0)
+                memory_gb = torch.cuda.get_device_properties(0).total_memory / (1024**3)
+                return True, f"GPU available: {device_name} ({memory_gb:.1f}GB)"
+            else:
+                return False, "No GPU available - will use CPU"
+                
+        except ImportError:
+            return False, "PyTorch not available for GPU check"
+        except Exception as e:
+            return False, f"GPU check failed: {str(e)}"
+    
+    @staticmethod
+    def validate_url(url):
+        """
+        Validate URL format.
+        
+        Args:
+            url: URL string to validate
+            
+        Returns:
+            tuple: (is_valid, error_message)
+        """
+        if not url:
+            return False, "No URL provided"
+        
+        # Basic URL pattern
+        url_pattern = re.compile(
+            r'^https?://'  # http:// or https://
+            r'(?:(?:[A-Z0-9](?:[A-Z0-9-]{0,61}[A-Z0-9])?\.)+[A-Z]{2,6}\.?|'  # domain...
+            r'localhost|'  # localhost...
+            r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})'  # ...or ip
+            r'(?::\d+)?'  # optional port
+            r'(?:/?|[/?]\S+)$', re.IGNORECASE)
+        
+        if url_pattern.match(url):
+            return True, "Valid URL"
+        else:
+            return False, "Invalid URL format"
+
 # ============================================================================
 # FILE MANAGER CLASS
 # ============================================================================