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

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

@@ -0,0 +1,606 @@
+#!/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())