Daniel Bedeleanu
1425856af5
Gemini analyzes images with EXIF orientation applied (e.g., 90° CW rotation), returning crop_bounds in that coordinate space. We need to transform them back to raw image coordinates before cropping. For EXIF orientation 6 (Rotate 90 CW): - Raw: 4032×3024 (landscape) - Displayed (after EXIF): 3024×4032 (portrait) - Transform portrait coords back to landscape before cropping This fixes the issue where crop was applied to wrong image region.
534 lines
19 KiB
Python
534 lines
19 KiB
Python
"""
|
||
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__)
|
||
|
||
|
||
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 yet)
|
||
exif_orientation = self._extract_exif_orientation(image)
|
||
|
||
# Transform crop_bounds from EXIF-applied space to raw space
|
||
if crop_bounds and exif_orientation and exif_orientation > 1:
|
||
crop_bounds = self._transform_crop_bounds_by_exif(
|
||
crop_bounds, exif_orientation, original_size[0], original_size[1]
|
||
)
|
||
msg = f"[CROP] EXIF {exif_orientation} transform applied: {crop_bounds}"
|
||
self.logger.info(msg)
|
||
print(f">>> {msg}")
|
||
|
||
# Smart cropping (on raw, un-rotated image so crop_bounds match AI analysis)
|
||
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 _transform_crop_bounds_by_exif(
|
||
self, crop_bounds: Dict, orientation: int, raw_width: int, raw_height: int
|
||
) -> Dict:
|
||
"""
|
||
Transform crop_bounds from EXIF-applied space to raw image space.
|
||
|
||
Gemini analyzes images with EXIF applied, returning crop_bounds in that space.
|
||
We need to transform them back to raw image coordinates before cropping.
|
||
|
||
Args:
|
||
crop_bounds: {x, y, width, height} in EXIF-applied space
|
||
orientation: EXIF orientation value (1-8)
|
||
raw_width: raw image width (before EXIF)
|
||
raw_height: raw image height (before EXIF)
|
||
|
||
Returns:
|
||
Transformed crop_bounds in raw image space
|
||
"""
|
||
if orientation == 1:
|
||
# No rotation, use as-is
|
||
return crop_bounds
|
||
elif orientation == 6:
|
||
# Rotate 90 CW: raw (w×h) appears as displayed (h×w)
|
||
# Transform from displayed (h×w) back to raw (w×h)
|
||
x, y, w, h = crop_bounds['x'], crop_bounds['y'], crop_bounds['width'], crop_bounds['height']
|
||
# After 90° CW: (x,y) in raw → (raw_height - y, x) in displayed
|
||
# Reverse: (x,y) in displayed → (y, raw_width - x) in raw
|
||
return {
|
||
'x': y,
|
||
'y': raw_width - x - w,
|
||
'width': h,
|
||
'height': w
|
||
}
|
||
else:
|
||
# Other orientations: return as-is (TODO: implement if needed)
|
||
self.logger.warning(f"EXIF orientation {orientation} not supported for crop_bounds transform, using as-is")
|
||
return crop_bounds
|
||
|
||
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''
|