py_backend/generate_production_thumbnails.py

#!/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)