"""Image processing utilities - thumbnail generation.""" import contextlib import io from uuid import UUID from PIL import Image as PILImage from app.core.storage import get_storage_client # Thumbnail sizes (width in pixels, height proportional) THUMBNAIL_SIZES = { "low": 800, # For slow connections "medium": 1600, # For medium connections "high": 3200, # For fast connections } def generate_thumbnails(image_id: UUID, original_path: str, contents: bytes) -> dict[str, str]: """ Generate thumbnails at different resolutions. Args: image_id: Image ID for naming thumbnails original_path: Path to original image contents: Original image contents Returns: Dictionary mapping quality level to thumbnail storage path """ storage = get_storage_client() thumbnail_paths = {} # Load original image image = PILImage.open(io.BytesIO(contents)) # Convert to RGB if necessary (for JPEG compatibility) if image.mode in ("RGBA", "LA", "P"): # Create white background for transparent images background = PILImage.new("RGB", image.size, (255, 255, 255)) if image.mode == "P": image = image.convert("RGBA") background.paste(image, mask=image.split()[-1] if image.mode in ("RGBA", "LA") else None) image = background elif image.mode != "RGB": image = image.convert("RGB") # Get original dimensions orig_width, orig_height = image.size # Generate thumbnails for each size for quality, max_width in THUMBNAIL_SIZES.items(): # Skip if original is smaller than thumbnail size if orig_width <= max_width: thumbnail_paths[quality] = original_path continue # Calculate proportional height ratio = max_width / orig_width new_height = int(orig_height * ratio) # Resize image thumbnail = image.resize((max_width, new_height), PILImage.Resampling.LANCZOS) # Convert to WebP for better compression output = io.BytesIO() thumbnail.save(output, format="WEBP", quality=85, method=6) output.seek(0) # Generate storage path thumbnail_path = f"thumbnails/{quality}/{image_id}.webp" # Upload to MinIO storage.put_object( bucket_name="webref", object_name=thumbnail_path, data=output, length=len(output.getvalue()), content_type="image/webp", ) thumbnail_paths[quality] = thumbnail_path return thumbnail_paths async def delete_thumbnails(thumbnail_paths: dict[str, str]) -> None: """ Delete thumbnails from storage. Args: thumbnail_paths: Dictionary of quality -> path """ storage = get_storage_client() for path in thumbnail_paths.values(): with contextlib.suppress(Exception): # Log error but continue storage.remove_object(bucket_name="webref", object_name=path)