Spaces:
Running
Running
File size: 12,316 Bytes
278a64b c79c7ae 278a64b |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 |
#!/usr/bin/env python3
"""
Script to convert existing thumbnails and detail images to WebP format using the new thumbnail generation logic.
This script should be run after deploying the new thumbnail service to update existing images.
"""
import os
import sys
import asyncio
import logging
from typing import List, Optional, Tuple
from sqlalchemy.orm import Session
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
# Add the current directory to Python path
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
from app.database import SessionLocal, engine
from app.models import Images
from app.services.thumbnail_service import ImageProcessingService
from app import storage
from app.config import settings
# Configure logging
try:
# Try to write to /tmp which should be writable in Docker
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.StreamHandler(sys.stdout),
logging.FileHandler('/tmp/thumbnail_conversion.log')
]
)
except PermissionError:
# Fallback to console-only logging if file writing fails
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.StreamHandler(sys.stdout)
]
)
logger = logging.getLogger(__name__)
class ThumbnailConverter:
def __init__(self):
self.db = SessionLocal()
self.converted_thumbnails = 0
self.converted_details = 0
self.skipped_count = 0
self.error_count = 0
def __del__(self):
if self.db:
self.db.close()
def get_all_images(self) -> List[Images]:
"""Fetch all images from the database"""
try:
images = self.db.query(Images).all()
logger.info(f"Found {len(images)} images in database")
return images
except Exception as e:
logger.error(f"Error fetching images from database: {e}")
return []
def needs_thumbnail_conversion(self, image: Images) -> bool:
"""Check if an image needs thumbnail conversion to WebP"""
# Skip if no thumbnail exists
if not image.thumbnail_key:
return False
# Skip if thumbnail is already WebP
if image.thumbnail_key.endswith('.webp'):
return False
# Convert if thumbnail is JPEG
if image.thumbnail_key.endswith('.jpg') or image.thumbnail_key.endswith('.jpeg'):
return True
# Skip other formats for now
return False
def needs_detail_conversion(self, image: Images) -> bool:
"""Check if an image needs detail image conversion to WebP"""
# Skip if no detail image exists
if not image.detail_key:
return False
# Skip if detail image is already WebP
if image.detail_key.endswith('.webp'):
return False
# Convert if detail image is JPEG
if image.detail_key.endswith('.jpg') or image.detail_key.endswith('.jpeg'):
return True
# Skip other formats for now
return False
def fetch_original_image(self, image: Images) -> Optional[bytes]:
"""Fetch the original image content from storage"""
try:
if hasattr(storage, 's3') and settings.STORAGE_PROVIDER != "local":
# S3 storage
response = storage.s3.get_object(
Bucket=settings.S3_BUCKET,
Key=image.file_key,
)
return response["Body"].read()
else:
# Local storage
file_path = os.path.join(settings.STORAGE_DIR, image.file_key)
if os.path.exists(file_path):
with open(file_path, 'rb') as f:
return f.read()
else:
logger.warning(f"Original image file not found: {file_path}")
return None
except Exception as e:
logger.error(f"Error fetching original image {image.image_id}: {e}")
return None
def delete_old_file(self, file_key: str, file_type: str) -> bool:
"""Delete the old file from storage"""
try:
if not file_key:
return True
if hasattr(storage, 's3') and settings.STORAGE_PROVIDER != "local":
# S3 storage
storage.s3.delete_object(
Bucket=settings.S3_BUCKET,
Key=file_key,
)
else:
# Local storage
file_path = os.path.join(settings.STORAGE_DIR, file_key)
if os.path.exists(file_path):
os.remove(file_path)
logger.info(f"Deleted old {file_type}: {file_key}")
return True
except Exception as e:
logger.error(f"Error deleting old {file_type} {file_key}: {e}")
return False
def convert_thumbnail(self, image: Images) -> bool:
"""Convert thumbnail to WebP format"""
try:
logger.info(f"Converting thumbnail for image {image.image_id}")
# Fetch original image
original_content = self.fetch_original_image(image)
if not original_content:
logger.error(f"Could not fetch original image for {image.image_id}")
return False
# Generate new WebP thumbnail
thumbnail_bytes, thumbnail_filename = ImageProcessingService.create_thumbnail(
original_content,
image.file_key
)
if not thumbnail_bytes or not thumbnail_filename:
logger.error(f"Failed to generate WebP thumbnail for {image.image_id}")
return False
# Upload new thumbnail
thumbnail_result = ImageProcessingService.upload_image_bytes(
thumbnail_bytes,
thumbnail_filename,
"WEBP"
)
if not thumbnail_result:
logger.error(f"Failed to upload WebP thumbnail for {image.image_id}")
return False
new_thumbnail_key, new_thumbnail_sha256 = thumbnail_result
# Delete old thumbnail
if not self.delete_old_file(image.thumbnail_key, "thumbnail"):
logger.warning(f"Could not delete old thumbnail for {image.image_id}")
# Update database record
image.thumbnail_key = new_thumbnail_key
image.thumbnail_sha256 = new_thumbnail_sha256
logger.info(f"Successfully converted thumbnail for {image.image_id}: {new_thumbnail_key}")
return True
except Exception as e:
logger.error(f"Error converting thumbnail for {image.image_id}: {e}")
return False
def convert_detail_image(self, image: Images) -> bool:
"""Convert detail image to WebP format"""
try:
logger.info(f"Converting detail image for image {image.image_id}")
# Fetch original image
original_content = self.fetch_original_image(image)
if not original_content:
logger.error(f"Could not fetch original image for {image.image_id}")
return False
# Generate new WebP detail image
detail_bytes, detail_filename = ImageProcessingService.create_detail_image(
original_content,
image.file_key
)
if not detail_bytes or not detail_filename:
logger.error(f"Failed to generate WebP detail image for {image.image_id}")
return False
# Upload new detail image
detail_result = ImageProcessingService.upload_image_bytes(
detail_bytes,
detail_filename,
"WEBP"
)
if not detail_result:
logger.error(f"Failed to upload WebP detail image for {image.image_id}")
return False
new_detail_key, new_detail_sha256 = detail_result
# Delete old detail image
if not self.delete_old_file(image.detail_key, "detail image"):
logger.warning(f"Could not delete old detail image for {image.image_id}")
# Update database record
image.detail_key = new_detail_key
image.detail_sha256 = new_detail_sha256
logger.info(f"Successfully converted detail image for {image.image_id}: {new_detail_key}")
return True
except Exception as e:
logger.error(f"Error converting detail image for {image.image_id}: {e}")
return False
def process_images(self) -> Tuple[int, int, int, int]:
"""Process all images and convert thumbnails and detail images to WebP"""
images = self.get_all_images()
if not images:
logger.warning("No images found in database")
return 0, 0, 0, 0
logger.info(f"Starting image conversion for {len(images)} images...")
for i, image in enumerate(images, 1):
try:
if i % 10 == 0:
logger.info(f"Progress: {i}/{len(images)} images processed")
needs_conversion = False
# Check and convert thumbnail
if self.needs_thumbnail_conversion(image):
if self.convert_thumbnail(image):
self.converted_thumbnails += 1
needs_conversion = True
else:
self.error_count += 1
# Check and convert detail image
if self.needs_detail_conversion(image):
if self.convert_detail_image(image):
self.converted_details += 1
needs_conversion = True
else:
self.error_count += 1
# Commit changes if any conversions were made
if needs_conversion:
self.db.commit()
else:
self.skipped_count += 1
except Exception as e:
logger.error(f"Error processing image {image.image_id}: {e}")
self.db.rollback()
self.error_count += 1
return self.converted_thumbnails, self.converted_details, self.skipped_count, self.error_count
def main():
"""Main function to run the thumbnail conversion"""
logger.info("Starting image conversion script...")
try:
converter = ThumbnailConverter()
converted_thumbnails, converted_details, skipped, errors = converter.process_images()
logger.info("=" * 50)
logger.info("IMAGE CONVERSION SUMMARY")
logger.info("=" * 50)
logger.info(f"Thumbnails converted to WebP: {converted_thumbnails}")
logger.info(f"Detail images converted to WebP: {converted_details}")
logger.info(f"Images skipped (already WebP or no images): {skipped}")
logger.info(f"Images with errors: {errors}")
logger.info(f"Total conversions: {converted_thumbnails + converted_details}")
logger.info("=" * 50)
if errors > 0:
logger.warning(f"Some images had errors during conversion. Check the log file for details.")
return 1
else:
logger.info("All image conversions completed successfully!")
return 0
except Exception as e:
logger.error(f"Fatal error during image conversion: {e}")
return 1
if __name__ == "__main__":
exit_code = main()
sys.exit(exit_code)
|