examples/use-cases/e-commerce/product_autmomation.py

#!/usr/bin/env python3
"""
BackgroundFX Pro - E-commerce Product Image Automation

Automates product photography workflow for e-commerce platforms:
- Batch process product images
- Apply consistent backgrounds
- Generate multiple sizes for different platforms
- Create transparent PNGs for overlays
- Generate marketing variations
"""

import os
import json
import csv
from pathlib import Path
from typing import List, Dict, Any, Optional, Tuple
from datetime import datetime
import asyncio
import aiohttp
from PIL import Image
import pandas as pd

# BackgroundFX API Configuration
API_KEY = os.getenv("BACKGROUNDFX_API_KEY")
API_URL = "https://api.backgroundfx.pro/v1"


class EcommerceProcessor:
    """
    Automated product image processor for e-commerce platforms.
    """
    
    # Platform-specific image requirements
    PLATFORM_SPECS = {
        'shopify': {
            'main': (2048, 2048),
            'thumbnail': (100, 100),
            'collection': (600, 600),
            'cart': (100, 100),
            'formats': ['jpg', 'webp']
        },
        'amazon': {
            'main': (2000, 2000),
            'zoom': (1600, 1600),
            'swatch': (30, 30),
            'formats': ['jpg']
        },
        'ebay': {
            'main': (1600, 1600),
            'gallery': (800, 800),
            'thumbnail': (140, 140),
            'formats': ['jpg']
        },
        'woocommerce': {
            'catalog': (300, 300),
            'single': (600, 600),
            'thumbnail': (150, 150),
            'formats': ['jpg', 'png']
        }
    }
    
    # Background presets for different product categories
    BACKGROUND_PRESETS = {
        'electronics': '#FFFFFF',
        'fashion': 'linear-gradient(180deg, #F5F5F5, #FFFFFF)',
        'jewelry': '#000000',
        'furniture': '#F8F8F8',
        'cosmetics': 'linear-gradient(45deg, #FFE5E5, #FFF0F5)',
        'sports': '#E8F4FD',
        'toys': 'linear-gradient(135deg, #667eea, #764ba2)',
        'food': '#FFF8DC'
    }
    
    def __init__(self, api_key: str):
        """Initialize the processor with API credentials."""
        self.api_key = api_key
        self.session = None
        self.stats = {
            'processed': 0,
            'failed': 0,
            'total_time': 0
        }
    
    async def __aenter__(self):
        """Async context manager entry."""
        self.session = aiohttp.ClientSession(
            headers={'Authorization': f'Bearer {self.api_key}'}
        )
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        """Async context manager exit."""
        if self.session:
            await self.session.close()
    
    async def process_product_catalog(
        self,
        csv_path: str,
        output_dir: str,
        platform: str = 'shopify'
    ) -> Dict[str, Any]:
        """
        Process entire product catalog from CSV.
        
        CSV Format:
        sku,image_path,category,title,background_override
        """
        print(f"📦 Processing product catalog for {platform}")
        
        # Read product catalog
        df = pd.read_csv(csv_path)
        total_products = len(df)
        
        results = []
        
        for idx, row in df.iterrows():
            print(f"\n[{idx+1}/{total_products}] Processing {row['sku']}...")
            
            try:
                result = await self.process_product(
                    image_path=row['image_path'],
                    sku=row['sku'],
                    category=row.get('category', 'general'),
                    output_dir=output_dir,
                    platform=platform,
                    custom_background=row.get('background_override')
                )
                results.append(result)
                self.stats['processed'] += 1
                
            except Exception as e:
                print(f"  ❌ Failed: {e}")
                self.stats['failed'] += 1
                results.append({
                    'sku': row['sku'],
                    'status': 'failed',
                    'error': str(e)
                })
        
        # Generate report
        report = self.generate_report(results, output_dir)
        
        return {
            'total': total_products,
            'processed': self.stats['processed'],
            'failed': self.stats['failed'],
            'report': report
        }
    
    async def process_product(
        self,
        image_path: str,
        sku: str,
        category: str,
        output_dir: str,
        platform: str,
        custom_background: Optional[str] = None
    ) -> Dict[str, Any]:
        """Process single product image."""
        
        start_time = datetime.now()
        
        # Step 1: Remove background
        processed_image = await self.remove_background(image_path)
        
        # Step 2: Apply appropriate background
        background = custom_background or self.BACKGROUND_PRESETS.get(
            category, '#FFFFFF'
        )
        final_image = await self.apply_background(
            processed_image['id'],
            background
        )
        
        # Step 3: Generate platform-specific sizes
        platform_images = await self.generate_platform_images(
            final_image['url'],
            sku,
            platform,
            output_dir
        )
        
        # Step 4: Create marketing variations
        marketing_images = await self.create_marketing_variations(
            processed_image['id'],
            sku,
            output_dir
        )
        
        elapsed = (datetime.now() - start_time).total_seconds()
        
        return {
            'sku': sku,
            'status': 'success',
            'processing_time': elapsed,
            'platform_images': platform_images,
            'marketing_images': marketing_images,
            'original': image_path,
            'processed': final_image['url']
        }
    
    async def remove_background(self, image_path: str) -> Dict[str, Any]:
        """Remove background from product image."""
        
        with open(image_path, 'rb') as f:
            data = aiohttp.FormData()
            data.add_field('file', f, filename=Path(image_path).name)
            data.add_field('quality', 'ultra')
            data.add_field('model', 'u2net')  # Best for products
            data.add_field('return_mask', 'true')
            
            async with self.session.post(
                f"{API_URL}/process/remove-background",
                data=data
            ) as response:
                response.raise_for_status()
                return await response.json()
    
    async def apply_background(
        self,
        image_id: str,
        background: str
    ) -> Dict[str, Any]:
        """Apply background to processed image."""
        
        payload = {
            'image_id': image_id,
            'background': background,
            'blend_mode': 'normal'
        }
        
        async with self.session.post(
            f"{API_URL}/process/replace-background",
            json=payload
        ) as response:
            response.raise_for_status()
            return await response.json()
    
    async def generate_platform_images(
        self,
        image_url: str,
        sku: str,
        platform: str,
        output_dir: str
    ) -> List[Dict[str, str]]:
        """Generate platform-specific image sizes."""
        
        specs = self.PLATFORM_SPECS.get(platform, self.PLATFORM_SPECS['shopify'])
        platform_dir = Path(output_dir) / platform / sku
        platform_dir.mkdir(parents=True, exist_ok=True)
        
        generated = []
        
        # Download processed image
        async with self.session.get(image_url) as response:
            image_data = await response.read()
        
        # Open with PIL
        from io import BytesIO
        img = Image.open(BytesIO(image_data))
        
        # Generate each required size
        for size_name, dimensions in specs.items():
            if size_name == 'formats':
                continue
                
            # Resize image
            resized = self.resize_image(img, dimensions)
            
            # Save in required formats
            for format_ext in specs['formats']:
                output_path = platform_dir / f"{sku}_{size_name}.{format_ext}"
                
                if format_ext == 'webp':
                    resized.save(output_path, 'WEBP', quality=90)
                else:
                    resized.save(output_path, 'JPEG', quality=95)
                
                generated.append({
                    'type': size_name,
                    'format': format_ext,
                    'path': str(output_path),
                    'dimensions': dimensions
                })
        
        print(f"  ✅ Generated {len(generated)} platform images")
        return generated
    
    async def create_marketing_variations(
        self,
        image_id: str,
        sku: str,
        output_dir: str
    ) -> List[Dict[str, str]]:
        """Create marketing variations with different backgrounds."""
        
        marketing_dir = Path(output_dir) / 'marketing' / sku
        marketing_dir.mkdir(parents=True, exist_ok=True)
        
        variations = [
            {'name': 'lifestyle', 'bg': 'https://example.com/lifestyle-bg.jpg'},
            {'name': 'seasonal_summer', 'bg': 'linear-gradient(to bottom, #87CEEB, #98FB98)'},
            {'name': 'seasonal_winter', 'bg': 'linear-gradient(to bottom, #B0E0E6, #FFFFFF)'},
            {'name': 'premium', 'bg': 'linear-gradient(45deg, #FFD700, #FFA500)'},
            {'name': 'minimal', 'bg': '#F5F5F5'},
            {'name': 'dark', 'bg': '#1a1a1a'},
            {'name': 'transparent', 'bg': 'transparent'}
        ]
        
        generated = []
        
        for variation in variations:
            try:
                result = await self.apply_background(image_id, variation['bg'])
                
                # Download and save
                async with self.session.get(result['url']) as response:
                    image_data = await response.read()
                
                output_path = marketing_dir / f"{sku}_{variation['name']}.png"
                with open(output_path, 'wb') as f:
                    f.write(image_data)
                
                generated.append({
                    'name': variation['name'],
                    'path': str(output_path),
                    'background': variation['bg']
                })
                
            except Exception as e:
                print(f"    ⚠️  Failed to create {variation['name']}: {e}")
        
        print(f"  ✅ Created {len(generated)} marketing variations")
        return generated
    
    def resize_image(
        self,
        img: Image.Image,
        target_size: Tuple[int, int]
    ) -> Image.Image:
        """
        Resize image maintaining aspect ratio and adding padding if needed.
        """
        # Calculate aspect ratios
        img_ratio = img.width / img.height
        target_ratio = target_size[0] / target_size[1]
        
        if img_ratio > target_ratio:
            # Image is wider - fit to width
            new_width = target_size[0]
            new_height = int(new_width / img_ratio)
        else:
            # Image is taller - fit to height
            new_height = target_size[1]
            new_width = int(new_height * img_ratio)
        
        # Resize image
        resized = img.resize((new_width, new_height), Image.Resampling.LANCZOS)
        
        # Create new image with padding
        final = Image.new('RGBA', target_size, (255, 255, 255, 0))
        
        # Calculate position to center the resized image
        x = (target_size[0] - new_width) // 2
        y = (target_size[1] - new_height) // 2
        
        # Paste resized image
        final.paste(resized, (x, y), resized if resized.mode == 'RGBA' else None)
        
        return final
    
    def generate_report(
        self,
        results: List[Dict[str, Any]],
        output_dir: str
    ) -> str:
        """Generate processing report."""
        
        report_path = Path(output_dir) / f"report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
        
        successful = [r for r in results if r.get('status') == 'success']
        failed = [r for r in results if r.get('status') == 'failed']
        
        report = {
            'timestamp': datetime.now().isoformat(),
            'summary': {
                'total': len(results),
                'successful': len(successful),
                'failed': len(failed),
                'success_rate': f"{(len(successful) / len(results) * 100):.1f}%",
                'total_processing_time': sum(r.get('processing_time', 0) for r in successful),
                'average_time_per_product': sum(r.get('processing_time', 0) for r in successful) / len(successful) if successful else 0
            },
            'successful_products': successful,
            'failed_products': failed
        }
        
        with open(report_path, 'w') as f:
            json.dump(report, f, indent=2)
        
        print(f"\n📊 Report saved to: {report_path}")
        return str(report_path)


class ShopifyIntegration:
    """
    Direct Shopify integration for automated product image updates.
    """
    
    def __init__(self, shop_url: str, access_token: str, backgroundfx_key: str):
        """Initialize Shopify integration."""
        self.shop_url = shop_url
        self.access_token = access_token
        self.processor = EcommerceProcessor(backgroundfx_key)
        self.shopify_api = f"https://{shop_url}/admin/api/2024-01"
    
    async def sync_product_images(self, product_id: str = None):
        """
        Sync product images with Shopify store.
        """
        # Get products from Shopify
        products = await self.get_shopify_products(product_id)
        
        for product in products:
            print(f"\n📦 Processing: {product['title']} (ID: {product['id']})")
            
            for image in product['images']:
                # Download original image
                original_path = await self.download_image(image['src'])
                
                # Process with BackgroundFX
                async with self.processor as proc:
                    result = await proc.process_product(
                        image_path=original_path,
                        sku=product['variants'][0]['sku'],
                        category=product['product_type'].lower(),
                        output_dir='shopify_processed',
                        platform='shopify'
                    )
                
                # Update Shopify with processed images
                if result['status'] == 'success':
                    await self.update_shopify_image(
                        product['id'],
                        image['id'],
                        result['processed']
                    )
    
    async def get_shopify_products(self, product_id: Optional[str] = None):
        """Fetch products from Shopify."""
        endpoint = f"{self.shopify_api}/products"
        if product_id:
            endpoint += f"/{product_id}"
        
        async with aiohttp.ClientSession() as session:
            async with session.get(
                endpoint + ".json",
                headers={'X-Shopify-Access-Token': self.access_token}
            ) as response:
                data = await response.json()
                return data['products'] if 'products' in data else [data['product']]
    
    async def download_image(self, url: str) -> str:
        """Download image from Shopify CDN."""
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as response:
                content = await response.read()
                
                # Save to temp file
                temp_path = Path('temp') / f"shopify_{datetime.now().timestamp()}.jpg"
                temp_path.parent.mkdir(exist_ok=True)
                
                with open(temp_path, 'wb') as f:
                    f.write(content)
                
                return str(temp_path)
    
    async def update_shopify_image(
        self,
        product_id: str,
        image_id: str,
        new_image_url: str
    ):
        """Update product image in Shopify."""
        endpoint = f"{self.shopify_api}/products/{product_id}/images/{image_id}.json"
        
        payload = {
            'image': {
                'id': image_id,
                'src': new_image_url
            }
        }
        
        async with aiohttp.ClientSession() as session:
            async with session.put(
                endpoint,
                json=payload,
                headers={'X-Shopify-Access-Token': self.access_token}
            ) as response:
                if response.status == 200:
                    print(f"  ✅ Updated image in Shopify")
                else:
                    print(f"  ❌ Failed to update Shopify: {response.status}")


async def main():
    """
    Main execution function with examples.
    """
    print("=" * 60)
    print("BackgroundFX Pro - E-commerce Automation")
    print("=" * 60)
    
    # Example 1: Process product catalog from CSV
    print("\n📋 Example 1: Batch Process Product Catalog")
    
    async with EcommerceProcessor(API_KEY) as processor:
        # Create sample CSV
        sample_csv = "products.csv"
        with open(sample_csv, 'w', newline='') as f:
            writer = csv.writer(f)
            writer.writerow(['sku', 'image_path', 'category', 'title', 'background_override'])
            writer.writerow(['PROD-001', 'images/shoe.jpg', 'fashion', 'Running Shoe', ''])
            writer.writerow(['PROD-002', 'images/watch.jpg', 'electronics', 'Smart Watch', '#000000'])
            writer.writerow(['PROD-003', 'images/bag.jpg', 'fashion', 'Leather Bag', ''])
        
        results = await processor.process_product_catalog(
            csv_path=sample_csv,
            output_dir='output/ecommerce',
            platform='shopify'
        )
        
        print(f"\n✅ Processed {results['processed']} products")
        print(f"❌ Failed: {results['failed']}")
        print(f"📊 Report: {results['report']}")
    
    # Example 2: Single product with multiple platforms
    print("\n📦 Example 2: Multi-Platform Product Processing")
    
    platforms = ['shopify', 'amazon', 'ebay']
    
    async with EcommerceProcessor(API_KEY) as processor:
        for platform in platforms:
            print(f"\n  Processing for {platform}...")
            
            result = await processor.process_product(
                image_path='images/product.jpg',
                sku='MULTI-001',
                category='electronics',
                output_dir='output/multiplatform',
                platform=platform
            )
            
            if result['status'] == 'success':
                print(f"    ✅ Generated {len(result['platform_images'])} images")
    
    # Example 3: Shopify Integration
    print("\n🛍️ Example 3: Direct Shopify Integration")
    
    if os.getenv('SHOPIFY_SHOP_URL') and os.getenv('SHOPIFY_ACCESS_TOKEN'):
        integration = ShopifyIntegration(
            shop_url=os.getenv('SHOPIFY_SHOP_URL'),
            access_token=os.getenv('SHOPIFY_ACCESS_TOKEN'),
            backgroundfx_key=API_KEY
        )
        
        await integration.sync_product_images()
    else:
        print("  ⚠️  Set SHOPIFY_SHOP_URL and SHOPIFY_ACCESS_TOKEN to test integration")
    
    # Example 4: A/B Testing Different Backgrounds
    print("\n🧪 Example 4: A/B Testing Backgrounds")
    
    test_backgrounds = [
        {'name': 'white', 'bg': '#FFFFFF'},
        {'name': 'gradient', 'bg': 'linear-gradient(180deg, #F5F5F5, #FFFFFF)'},
        {'name': 'colored', 'bg': '#E8F4FD'},
        {'name': 'lifestyle', 'bg': 'https://example.com/kitchen-bg.jpg'}
    ]
    
    async with EcommerceProcessor(API_KEY) as processor:
        # Remove background once
        processed = await processor.remove_background('images/product.jpg')
        
        # Test different backgrounds
        for test in test_backgrounds:
            result = await processor.apply_background(
                processed['id'],
                test['bg']
            )
            print(f"  ✅ Created variant: {test['name']}")
    
    print("\n" + "=" * 60)
    print("E-commerce automation complete!")
    print("=" * 60)


if __name__ == "__main__":
    # Check API key
    if not API_KEY:
        print("❌ Please set BACKGROUNDFX_API_KEY environment variable")
        exit(1)
    
    # Run async main
    asyncio.run(main())