236 lines
8.6 KiB
Python
236 lines
8.6 KiB
Python
"""
|
|
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}") |