""" Thumbnail Generator for image and video files """ from PIL import Image, ImageOps import io import os import hashlib import logging from typing import Optional, Tuple import asyncio from pathlib import Path logger = logging.getLogger(__name__) class ThumbnailGenerator: def __init__(self, minio_client, cache_dir: str = "/tmp/thumbnails"): self.minio_client = minio_client self.cache_dir = Path(cache_dir) self.cache_dir.mkdir(parents=True, exist_ok=True) # Supported image formats for thumbnail generation self.supported_formats = { 'image/jpeg', 'image/jpg', 'image/png', 'image/gif', 'image/webp', 'image/bmp', 'image/tiff' } def _get_cache_path(self, file_id: str, width: int, height: int) -> Path: """Generate cache file path for thumbnail""" cache_key = f"{file_id}_{width}x{height}" cache_hash = hashlib.md5(cache_key.encode()).hexdigest() return self.cache_dir / f"{cache_hash[:2]}" / f"{cache_hash}.jpg" async def generate_thumbnail(self, file_data: bytes, content_type: str, width: int = 200, height: int = 200) -> Optional[bytes]: """Generate a thumbnail from file data""" try: if content_type not in self.supported_formats: logger.warning(f"Unsupported format for thumbnail: {content_type}") return None loop = asyncio.get_event_loop() # Generate thumbnail in thread pool thumbnail_data = await loop.run_in_executor( None, self._create_thumbnail, file_data, width, height ) return thumbnail_data except Exception as e: logger.error(f"Failed to generate thumbnail: {e}") return None def _create_thumbnail(self, file_data: bytes, width: int, height: int) -> bytes: """Create thumbnail using PIL""" try: # Open image image = Image.open(io.BytesIO(file_data)) # Convert RGBA to RGB if necessary if image.mode in ('RGBA', 'LA', 'P'): # Create a white background background = Image.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 == 'RGBA' else None) image = background elif image.mode not in ('RGB', 'L'): image = image.convert('RGB') # Calculate thumbnail size maintaining aspect ratio image.thumbnail((width, height), Image.Resampling.LANCZOS) # Apply EXIF orientation if present image = ImageOps.exif_transpose(image) # Save thumbnail to bytes output = io.BytesIO() image.save(output, format='JPEG', quality=85, optimize=True) output.seek(0) return output.read() except Exception as e: logger.error(f"Thumbnail creation failed: {e}") raise async def get_thumbnail(self, file_id: str, bucket: str, object_name: str, width: int = 200, height: int = 200) -> Optional[bytes]: """Get or generate thumbnail for a file""" try: # Check cache first cache_path = self._get_cache_path(file_id, width, height) if cache_path.exists(): logger.info(f"Thumbnail found in cache: {cache_path}") with open(cache_path, 'rb') as f: return f.read() # Check if thumbnail exists in MinIO thumbnail_object = f"thumbnails/{file_id}_{width}x{height}.jpg" try: thumbnail_stream = await self.minio_client.get_file( bucket="thumbnails", object_name=thumbnail_object ) thumbnail_data = thumbnail_stream.read() # Save to cache await self._save_to_cache(cache_path, thumbnail_data) return thumbnail_data except: pass # Thumbnail doesn't exist, generate it # Get original file file_stream = await self.minio_client.get_file(bucket, object_name) file_data = file_stream.read() # Get file info for content type file_info = await self.minio_client.get_file_info(bucket, object_name) content_type = file_info.get("content_type", "") # Generate thumbnail thumbnail_data = await self.generate_thumbnail( file_data, content_type, width, height ) if thumbnail_data: # Save to MinIO await self.minio_client.upload_file( bucket="thumbnails", object_name=thumbnail_object, file_data=thumbnail_data, content_type="image/jpeg" ) # Save to cache await self._save_to_cache(cache_path, thumbnail_data) return thumbnail_data except Exception as e: logger.error(f"Failed to get thumbnail: {e}") return None async def _save_to_cache(self, cache_path: Path, data: bytes): """Save thumbnail to cache""" try: cache_path.parent.mkdir(parents=True, exist_ok=True) loop = asyncio.get_event_loop() await loop.run_in_executor( None, lambda: cache_path.write_bytes(data) ) logger.info(f"Thumbnail saved to cache: {cache_path}") except Exception as e: logger.error(f"Failed to save to cache: {e}") async def delete_thumbnail(self, file_id: str): """Delete all thumbnails for a file""" try: # Delete from cache for cache_file in self.cache_dir.rglob(f"*{file_id}*"): try: cache_file.unlink() logger.info(f"Deleted cache file: {cache_file}") except: pass # Delete from MinIO (list and delete all sizes) files = await self.minio_client.list_files( bucket="thumbnails", prefix=f"thumbnails/{file_id}_" ) for file in files: await self.minio_client.delete_file( bucket="thumbnails", object_name=file["name"] ) logger.info(f"Deleted thumbnail: {file['name']}") except Exception as e: logger.error(f"Failed to delete thumbnails: {e}") async def generate_multiple_sizes(self, file_data: bytes, content_type: str, file_id: str) -> dict: """Generate thumbnails in multiple sizes""" sizes = { "small": (150, 150), "medium": (300, 300), "large": (600, 600) } results = {} for size_name, (width, height) in sizes.items(): thumbnail = await self.generate_thumbnail( file_data, content_type, width, height ) if thumbnail: # Save to MinIO object_name = f"thumbnails/{file_id}_{size_name}.jpg" await self.minio_client.upload_file( bucket="thumbnails", object_name=object_name, file_data=thumbnail, content_type="image/jpeg" ) results[size_name] = { "size": len(thumbnail), "dimensions": f"{width}x{height}", "object_name": object_name } return results def clear_cache(self): """Clear thumbnail cache""" try: import shutil shutil.rmtree(self.cache_dir) self.cache_dir.mkdir(parents=True, exist_ok=True) logger.info("Thumbnail cache cleared") except Exception as e: logger.error(f"Failed to clear cache: {e}")