""" OpenCV-based image processing pipeline for smart photo handling. Handles: - EXIF orientation detection and auto-rotation - Smart cropping using OpenCV contour detection - Text orientation detection using Hough lines - Resize and compression to 1200px - Thumbnail generation (200px square) - Fallback to Pillow for basic processing if OpenCV fails """ import io import logging from typing import Dict, Optional, Tuple from PIL import Image from PIL.ExifTags import TAGS import piexif import cv2 import numpy as np logger = logging.getLogger(__name__) def strip_exif_orientation(file_bytes: bytes) -> bytes: """ Remove EXIF orientation metadata from image bytes. Returns image bytes with orientation tag removed (or set to 1 = normal). This ensures both Gemini and our backend analyze the same raw image. Args: file_bytes: Raw image file bytes Returns: Image bytes with EXIF orientation stripped """ try: image = Image.open(io.BytesIO(file_bytes)) # Try to get and remove EXIF orientation try: exif_dict = piexif.load(image.info.get('exif', b'')) if piexif.ImageIFD.Orientation in exif_dict['0th']: del exif_dict['0th'][piexif.ImageIFD.Orientation] exif_bytes = piexif.dump(exif_dict) else: exif_bytes = None except: exif_bytes = None # Save image without orientation output = io.BytesIO() if exif_bytes: image.save(output, format='JPEG', quality=85, exif=exif_bytes) else: image.save(output, format='JPEG', quality=85) return output.getvalue() except Exception as e: logger.warning(f"Failed to strip EXIF orientation: {e}, returning original bytes") return file_bytes class ImageProcessor: """Service for processing uploaded images with smart features.""" MAX_FILE_SIZE = 10 * 1024 * 1024 # 10MB LONG_SIDE = 1200 THUMBNAIL_SIZE = 200 JPEG_QUALITY = 85 # Algorithm parameters (Issue 4: DRY Violation - Magic Numbers) CANNY_CROP_THRESHOLDS = (100, 200) CANNY_TEXT_THRESHOLDS = (50, 150) HOUGH_THRESHOLD = 100 CROP_PADDING_FACTOR = 0.1 ANGLE_UPSIDE_DOWN_THRESHOLD = 80 # degrees ANGLE_SIDEWAYS_THRESHOLD = 45 # degrees def __init__(self): """Initialize the image processor.""" self.logger = logger def process_photo( self, file_bytes: bytes, crop_bounds: Optional[Dict] = None, rotation_degrees: float = 0 ) -> Dict: """ Process a photo with EXIF rotation, smart cropping, and compression. Args: file_bytes: Raw image file bytes crop_bounds: Optional manual crop bounds {x, y, width, height} rotation_degrees: Optional manual rotation in degrees (applied after crop) Returns: { 'status': 'success' | 'error', 'cropped_image_bytes': bytes or None, 'thumbnail_bytes': bytes or None, 'original_size': (width, height), 'crop_size': (width, height) or None, 'text_angle': float or None, 'metadata': { 'exif_orientation': int, 'crop_method': 'manual' | 'opencv' | 'pillow' | 'none', 'file_size_bytes': int } } """ try: # Validate file size if len(file_bytes) > self.MAX_FILE_SIZE: return { 'status': 'error', 'error': f'File too large: {len(file_bytes)} > {self.MAX_FILE_SIZE}', 'cropped_image_bytes': None, 'thumbnail_bytes': None, } # Open image with PIL (without applying EXIF yet, so crop_bounds match AI analysis) image = Image.open(io.BytesIO(file_bytes)) original_size = image.size msg = f"[PROCESS] Input: {len(file_bytes)} bytes, size={original_size}, crop_bounds={crop_bounds}, rotation={rotation_degrees}°" self.logger.info(msg) print(f">>> {msg}") # Explicit print for visibility # Extract EXIF orientation (but don't apply it) exif_orientation = self._extract_exif_orientation(image) if exif_orientation and exif_orientation > 1: self.logger.info(f"[PROCESS] Note: Image has EXIF orientation {exif_orientation}, will be applied after crop") # Smart cropping (on raw image - crop_bounds come from Gemini analyzing same raw image) cropped_image = image crop_size = None text_angle = None crop_method = 'none' if crop_bounds: # Manual crop bounds provided (from AI, based on raw image) crop_rect = ( crop_bounds['x'], crop_bounds['y'], crop_bounds['x'] + crop_bounds['width'], crop_bounds['y'] + crop_bounds['height'], ) msg1 = f"[CROP] Manual bounds: {crop_rect}" self.logger.info(msg1) print(f">>> {msg1}") # Explicit print cropped_image = image.crop(crop_rect) crop_size = cropped_image.size crop_method = 'manual' msg2 = f"[CROP] Result size: {crop_size}, pixels={crop_size[0]*crop_size[1]}" self.logger.info(msg2) print(f">>> {msg2}") # Explicit print else: # Try OpenCV smart crop on raw image self.logger.info("[CROP] Attempting OpenCV smart crop...") try: crop_result = self._smart_crop_opencv(image) if crop_result is not None: cropped_image, crop_size = crop_result crop_method = 'opencv' self.logger.info(f"[CROP] OpenCV success: {crop_size}, pixels={crop_size[0]*crop_size[1]}") # Detect text orientation within the cropped region text_angle, angle_status = self._detect_text_orientation( cropped_image ) if text_angle is not None: self.logger.info( f"[CROP] Text angle: {text_angle}° ({angle_status})" ) if angle_status in ['upside_down', 'sideways']: cropped_image = self._rotate_image( cropped_image, text_angle ) else: self.logger.warning("[CROP] OpenCV returned None, using full image") crop_method = 'pillow' except (IOError, ValueError, cv2.error) as e: # Fallback to Pillow if OpenCV fails self.logger.warning( f"[CROP] OpenCV failed: {e}, using full image" ) crop_method = 'pillow' # Now apply EXIF orientation to the cropped image if exif_orientation and exif_orientation > 1: cropped_image = self._rotate_by_orientation(cropped_image, exif_orientation) self.logger.info(f"Applied EXIF rotation: {exif_orientation}") # Apply manual rotation if provided (rotation_degrees already signed: positive=CCW, negative=CW) if abs(rotation_degrees) > 0.5: cropped_image = cropped_image.rotate(rotation_degrees, expand=True) msg = f"Applied manual rotation: {rotation_degrees}°" self.logger.info(msg) print(f">>> {msg}") # Explicit print # Resize and compress compressed_bytes = self._resize_and_compress(cropped_image) # Generate thumbnail thumbnail_bytes = self._generate_thumbnail(image) return { 'status': 'success', 'cropped_image_bytes': compressed_bytes, 'thumbnail_bytes': thumbnail_bytes, 'original_size': original_size, 'crop_size': crop_size, 'text_angle': text_angle, 'metadata': { 'exif_orientation': exif_orientation or 1, 'crop_method': crop_method, 'file_size_bytes': len(file_bytes), }, } except (IOError, ValueError, cv2.error) as e: self.logger.error(f"Image processing failed: {e}") return { 'status': 'error', 'error': str(e), 'cropped_image_bytes': None, 'thumbnail_bytes': None, } def _extract_exif_orientation(self, image: Image.Image) -> Optional[int]: """ Extract EXIF orientation tag from image. Returns: Orientation value (1-8) or None if not present """ try: # Use piexif for EXIF extraction (avoiding private PIL API) if hasattr(image, 'info') and 'exif' in image.info: exif_dict = piexif.load(image.info['exif']) orientation = exif_dict['0th'].get(piexif.ImageIFD.Orientation) if orientation: return orientation return None except (piexif.InvalidImageData, ValueError, IOError) as e: self.logger.debug(f"Could not extract EXIF orientation: {e}") return None def _rotate_by_orientation( self, image: Image.Image, orientation: int ) -> Image.Image: """ Rotate image based on EXIF orientation tag. Args: image: PIL Image orientation: EXIF orientation value (1-8) Returns: Rotated PIL Image """ if orientation == 1: return image elif orientation == 2: return image.transpose(Image.Transpose.FLIP_LEFT_RIGHT) elif orientation == 3: return image.transpose(Image.Transpose.ROTATE_180) elif orientation == 4: return image.transpose(Image.Transpose.FLIP_TOP_BOTTOM) elif orientation == 5: return image.transpose(Image.Transpose.TRANSPOSE) elif orientation == 6: return image.transpose(Image.Transpose.ROTATE_270) elif orientation == 7: return image.transpose(Image.Transpose.TRANSVERSE) elif orientation == 8: return image.transpose(Image.Transpose.ROTATE_90) return image def _smart_crop_opencv( self, image: Image.Image ) -> Optional[Tuple[Image.Image, Tuple[int, int]]]: """ Use OpenCV to detect and crop the main object in the image. Args: image: PIL Image Returns: Tuple of (cropped PIL Image, crop size) or None if no contours found """ try: # Convert PIL image to OpenCV format cv_image = cv2.cvtColor(np.array(image), cv2.COLOR_RGB2BGR) gray = cv2.cvtColor(cv_image, cv2.COLOR_BGR2GRAY) # Edge detection edges = cv2.Canny(gray, *self.CANNY_CROP_THRESHOLDS) # Find contours contours, _ = cv2.findContours( edges, cv2.RETR_TREE, cv2.CHAIN_APPROX_SIMPLE ) if not contours: self.logger.debug("No contours found in image") return None # Get bounding box of largest contour largest_contour = max(contours, key=cv2.contourArea) x, y, w, h = cv2.boundingRect(largest_contour) # Apply padding around bounds pad_x = int(w * self.CROP_PADDING_FACTOR) pad_y = int(h * self.CROP_PADDING_FACTOR) x1 = max(0, x - pad_x) y1 = max(0, y - pad_y) x2 = min(cv_image.shape[1], x + w + pad_x) y2 = min(cv_image.shape[0], y + h + pad_y) # Crop image cropped = image.crop((x1, y1, x2, y2)) crop_size = cropped.size self.logger.debug( f"OpenCV crop bounds: ({x1}, {y1}, {x2}, {y2}), size: {crop_size}" ) return cropped, crop_size except (IOError, ValueError, cv2.error) as e: self.logger.warning(f"OpenCV smart crop failed: {e}") return None def _detect_text_orientation( self, image: Image.Image ) -> Tuple[Optional[float], str]: """ Detect text orientation using Hough line transform. Args: image: PIL Image Returns: Tuple of (angle in degrees, status string) status: 'normal', 'upside_down', 'sideways', 'not_detected' """ try: # Convert to OpenCV format cv_image = cv2.cvtColor(np.array(image), cv2.COLOR_RGB2BGR) # DoS prevention: check resolution if cv_image.shape[0] * cv_image.shape[1] > 2000 * 2000: # >4MP self.logger.warning("ROI too large for text detection, skipping") return None, 'not_detected' gray = cv2.cvtColor(cv_image, cv2.COLOR_BGR2GRAY) # Edge detection edges = cv2.Canny(gray, *self.CANNY_TEXT_THRESHOLDS) # Hough line detection lines = cv2.HoughLines(edges, 1, np.pi / 180, self.HOUGH_THRESHOLD) if lines is None or len(lines) == 0: self.logger.debug("No lines detected for text orientation") return None, 'not_detected' # Extract angles from lines angles = [] for line in lines: rho, theta = line[0] angle = np.degrees(theta) angles.append(angle) # Normalize angles to 0-180 range angles = np.array(angles) angles = np.where(angles > 90, angles - 180, angles) # Find dominant angle mean_angle = np.mean(angles) # Determine orientation status status = 'normal' corrected_angle = mean_angle # Check for upside-down text (~180°) if abs(mean_angle) > self.ANGLE_UPSIDE_DOWN_THRESHOLD: status = 'upside_down' corrected_angle = mean_angle + 180 if mean_angle > 0 else mean_angle - 180 # Check for sideways text (~90°) elif abs(mean_angle) > self.ANGLE_SIDEWAYS_THRESHOLD: status = 'sideways' self.logger.debug( f"Text orientation: angle={mean_angle:.1f}°, status={status}" ) return corrected_angle, status except (IOError, ValueError, cv2.error) as e: self.logger.warning(f"Text orientation detection failed: {e}") return None, 'not_detected' def _rotate_image(self, image: Image.Image, angle: float) -> Image.Image: """ Rotate image by specified angle. Args: image: PIL Image angle: Rotation angle in degrees Returns: Rotated PIL Image """ return image.rotate(angle, expand=False, fillcolor='white') def _resize_and_compress(self, image: Image.Image) -> bytes: """ Resize image to 1200px on long side and compress to JPEG. Args: image: PIL Image Returns: Compressed JPEG bytes """ # Get current size width, height = image.size max_dim = max(width, height) # Only resize if necessary if max_dim > self.LONG_SIDE: scale = self.LONG_SIDE / max_dim new_width = int(width * scale) new_height = int(height * scale) image = image.resize((new_width, new_height), Image.Resampling.LANCZOS) self.logger.debug( f"Resized from {(width, height)} to {(new_width, new_height)}" ) # Convert to RGB if necessary (for JPEG) if image.mode in ('RGBA', 'LA', 'P'): # Extract alpha channel if present mask = None if image.mode == 'RGBA': alpha = image.split()[3] mask = alpha elif image.mode == 'LA': alpha = image.split()[1] mask = alpha rgb_image = Image.new('RGB', image.size, (255, 255, 255)) rgb_image.paste(image, mask=mask) image = rgb_image # Compress to JPEG output = io.BytesIO() image.save(output, format='JPEG', quality=self.JPEG_QUALITY, optimize=True) compressed_bytes = output.getvalue() self.logger.debug( f"Compressed to JPEG: {len(compressed_bytes)} bytes, " f"quality={self.JPEG_QUALITY}" ) return compressed_bytes def _generate_thumbnail(self, image: Image.Image) -> bytes: """ Generate 200px square thumbnail with center crop. Args: image: PIL Image Returns: Thumbnail JPEG bytes """ try: # Get current size width, height = image.size # Center crop to square min_dim = min(width, height) left = (width - min_dim) // 2 top = (height - min_dim) // 2 right = left + min_dim bottom = top + min_dim square = image.crop((left, top, right, bottom)) # Resize to thumbnail size thumbnail = square.resize( (self.THUMBNAIL_SIZE, self.THUMBNAIL_SIZE), Image.Resampling.LANCZOS, ) # Convert to RGB if necessary if thumbnail.mode in ('RGBA', 'LA', 'P'): # Extract alpha channel if present mask = None if thumbnail.mode == 'RGBA': alpha = thumbnail.split()[3] mask = alpha elif thumbnail.mode == 'LA': alpha = thumbnail.split()[1] mask = alpha rgb_thumbnail = Image.new('RGB', thumbnail.size, (255, 255, 255)) rgb_thumbnail.paste(thumbnail, mask=mask) thumbnail = rgb_thumbnail # Compress output = io.BytesIO() thumbnail.save(output, format='JPEG', quality=self.JPEG_QUALITY, optimize=True) thumbnail_bytes = output.getvalue() self.logger.debug( f"Generated thumbnail: {self.THUMBNAIL_SIZE}x{self.THUMBNAIL_SIZE}, " f"{len(thumbnail_bytes)} bytes" ) return thumbnail_bytes except (IOError, ValueError, cv2.error) as e: self.logger.error(f"Thumbnail generation failed: {e}") return b''