ecommerce-image-generator.py

```python
#!/usr/bin/env python3
"""
AI Forge E-commerce Image Generator
Stable Diffusion + Predictive ML for personalized product mockups
Optimized for 220% YoY demand growth in visual content creation
"""

import os
import torch
import pandas as pd
import numpy as np
from diffusers import StableDiffusionPipeline, DPMSolverMultistepScheduler
import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers
import matplotlib.pyplot as plt
from PIL import Image
import io
import base64
import warnings
warnings.filterwarnings('ignore')

# Configuration
class Config:
    """Configuration parameters for the image generation system"""
    STABLE_DIFFUSION_MODEL = "runwayml/stable-diffusion-v1-5"
    IMAGE_SIZE = (512, 512)
    MAX_LENGTH = 77
    BATCH_SIZE = 4
    EPOCHS = 50
    LEARNING_RATE = 1e-4
    PREDICTION_MODEL_PATH = "./models/style_predictor.h5"
    TRAINING_DATA_PATH = "./data/ecommerce_sales.csv"
    OUTPUT_DIR = "./generated_images"
    
    # Style categories for prediction
    STYLE_CATEGORIES = ['minimalist', 'vintage', 'modern', 'luxury', 'tech', 'sporty']
    COLOR_CATEGORIES = ['blue', 'red', 'green', 'black', 'white', 'pastel', 'neon']
    PRODUCT_TYPES = ['clothing', 'electronics', 'home_decor', 'beauty', 'accessories']

class StylePredictor:
    """TensorFlow model for predicting trending styles and colors"""
    
    def __init__(self, input_dim):
        self.input_dim = input_dim
        self.model = None
        
    def build_model(self):
        """Build the style prediction model"""
        model = keras.Sequential([
            layers.Dense(256, activation='relu', input_shape=(input_dim,)),
            layers.Dropout(0.3),
            layers.Dense(128, activation='relu'),
            layers.Dropout(0.2),
            layers.Dense(64, activation='relu'),
            layers.Dense(len(Config.STYLE_CATEGORIES) + len(Config.COLOR_CATEGORIES)),
            layers.Activation('sigmoid')
        ])
        
        model.compile(
            optimizer=keras.optimizers.Adam(learning_rate=Config.LEARNING_RATE),
            loss='binary_crossentropy',
            metrics=['accuracy']
        )
        
        self.model = model
        return model
    
    def train(self, X_train, y_train, X_val=None, y_val=None):
        """Train the style prediction model"""
        callbacks = [
            keras.callbacks.EarlyStopping(patience=10, restore_best_weights=True),
            keras.callbacks.ReduceLROnPlateau(factor=0.5, patience=5),
            keras.callbacks.ModelCheckpoint(
                Config.PREDICTION_MODEL_PATH,
                save_best_only=True,
                monitor='val_loss' if X_val is not None else 'loss'
            )
        ]
        
        history = self.model.fit(
            X_train, y_train,
            batch_size=Config.BATCH_SIZE,
            epochs=Config.EPOCHS,
            validation_data=(X_val, y_val) if X_val is not None else None,
            callbacks=callbacks,
            verbose=1
        )
        
        return history
    
    def predict_trends(self, customer_data):
        """Predict trending styles and colors for customer segment"""
        predictions = self.model.predict(customer_data)
        
        # Split predictions into styles and colors
        style_predictions = predictions[:, :len(Config.STYLE_CATEGORIES)]
        color_predictions = predictions[:, len(Config.STYLE_CATEGORIES):]
        
        return style_predictions, color_predictions

class EcommerceDataProcessor:
    """Process e-commerce sales data for trend prediction"""
    
    def __init__(self):
        self.feature_columns = []
        
    def load_and_preprocess_data(self, file_path):
        """Load and preprocess e-commerce sales data"""
        try:
            df = pd.read_csv(file_path)
            print(f"Loaded dataset with {len(df)} rows")
            return df
        except Exception as e:
            print(f"Error loading data: {e}")
            return None
    
    def extract_features(self, df):
        """Extract features from e-commerce data"""
        features = []
        
        # Demographic features
        demographic_features = ['age', 'income_level', 'location_urban', 'gender_encoded']
        
        # Time-based features
        df['purchase_month'] = pd.to_datetime(df['purchase_date']).dt.month
        features.append(pd.get_dummies(df['purchase_month'], prefix='month'))
        
        # Product features
        product_features = ['price', 'category_encoded', 'brand_popularity']
        
        # Combine all features
        for feature in demographic_features + ['purchase_month', 'price', 'category_encoded', 'brand_popularity']:
            if feature in df.columns:
                features.append(df[[feature]]))
        
        # One-hot encode categorical variables
        categorical_cols = ['region', 'device_type', 'marketing_channel']
        for col in categorical_cols:
            if col in df.columns:
                dummies = pd.get_dummies(df[col], prefix=col)
                features.append(dummies)
        
        X = pd.concat(features, axis=1)
        self.feature_columns = X.columns.tolist()
        
        return X
    
    def prepare_training_labels(self, df):
        """Prepare training labels for style and color trends"""
        # Create binary labels for styles and colors based on sales performance
        labels = []
        
        for _, row in df.iterrows():
            # Style preferences (based on product attributes)
            style_vector = [0] * len(Config.STYLE_CATEGORIES)
            color_vector = [0] * len(Config.COLOR_CATEGORIES)
        
        # For each product, determine dominant style and color
            if row['sales_rank'] <= 100:  # Top selling products
                # Analyze product description for style keywords
                description = str(row.get('product_description', '')).lower()
            
            for i, style in enumerate(Config.STYLE_CATEGORIES):
                if style in description:
                    style_vector[i] = 1
            
            # Color analysis from product data
            color_data = str(row.get('color_data', '')).lower()
                for j, color in enumerate(Config.COLOR_CATEGORIES):
                    if color in description or color in str(row.get('primary_color', '')).lower():
                    color_vector[j] = 1
            
            labels.append(style_vector + color_vector)
        
        return np.array(labels)

class StableDiffusionGenerator:
    """Stable Diffusion image generator for e-commerce mockups"""
    
    def __init__(self):
        self.pipeline = None
        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
        print(f"Using device: {self.device}")
        
    def load_model(self):
        """Load Stable Diffusion model"""
        try:
            self.pipeline = StableDiffusionPipeline.from_pretrained(
                Config.STABLE_DIFFUSION_MODEL,
                torch_dtype=torch.float16 if torch.cuda.is_available() else torch.float32
            )
            self.pipeline.scheduler = DPMSolverMultistepScheduler.from_config(
                self.pipeline.scheduler.config
            )
            self.pipeline = self.pipeline.to(self.device)
            print("Stable Diffusion model loaded successfully")
        except Exception as e:
            print(f"Error loading model: {e}")
    
    def generate_product_mockup(self, prompt, style_weights=None, color_weights=None):
        """Generate product mockup with style and color conditioning"""
        
        # Enhance prompt based on predicted trends
        enhanced_prompt = self._enhance_prompt(prompt, style_weights, color_weights)
        
        # Generate image
        with torch.autocast(self.device.type):
            image = self.pipeline(
                prompt,
                height=Config.IMAGE_SIZE[0],
                width=Config.IMAGE_SIZE[1],
                num_inference_steps=25,
                guidance_scale=7.5
            ).images[0]
        
        return image
    
    def _enhance_prompt(self, base_prompt, style_weights, color_weights):
        """Enhance prompt with style and color conditioning"""
        
        if style_weights is not None:
            # Get top predicted styles
            top_style_indices = np.argsort(style_weights)[-2:]  # Top 2 styles
            top_colors_indices = np.argsort(color_weights)[-2:]
        
        # Add style descriptors
        style_descriptors = []
        for idx in top_style_indices:
            style_descriptors.append(Config.STYLE_CATEGORIES[idx])
        
        # Add color descriptors
        color_descriptors = []
        for idx in top_colors_indices:
            color_descriptors.append(Config.COLOR_CATEGORIES[idx])
        
        enhanced_prompt = f"{base_prompt}, {', '.join(style_descriptors)} style, colors: {', '.join(color_descriptors)}"
        
        return enhanced_prompt
    
    def generate_batch_mockups(self, prompts, style_predictions, color_predictions):
        """Generate multiple product mockups in batch"""
        images = []
        
        for i, prompt in enumerate(prompts):
            style_weights = style_predictions[i] if i < len(style_predictions) else None
            color_weights = color_predictions[i] if i < len(color_predictions) else None
            
            image = self.generate_product_mockup(prompt, style_weights, color_weights)
            images.append(image)
        
        return images

class EcommerceImageAPI:
    """FastAPI integration for the e-commerce image generation system"""
    
    def __init__(self):
        self.data_processor = EcommerceDataProcessor()
        self.style_predictor = None
        self.image_generator = StableDiffusionGenerator()
        
    def initialize_system(self):
        """Initialize the complete system"""
        print("Initializing E-commerce Image Generation System...")
        
        # Load data
        df = self.data_processor.load_and_preprocess_data(Config.TRAINING_DATA_PATH)
        
        if df is not None:
            # Prepare features and labels
            X = self.data_processor.extract_features(df)
        y = self.data_processor.prepare_training_labels(df)
        
            # Initialize and train style predictor
            self.style_predictor = StylePredictor(X.shape[1])
            self.style_predictor.build_model()
            
            # Split data
            from sklearn.model_selection import train_test_split
            X_train, X_test, y_train, y_test = train_test_split(
                X, y, test_size=0.2, random_state=42
            )
            
            # Train model
            print("Training style prediction model...")
            history = self.style_predictor.train(X_train, y_train, X_test, y_test)
            
            # Evaluate model
            test_loss, test_accuracy = self.style_predictor.model.evaluate(X_test, y_test)
            print(f"Model trained - Test Accuracy: {test_accuracy:.4f}")
        
        # Load image generator
        self.image_generator.load_model()
        
        print("System initialized successfully")
    
    def predict_and_generate(self, customer_segment_data, base_prompts):
        """Complete workflow: predict trends and generate images"""
        
        # Predict styles and colors
        style_predictions, color_predictions = self.style_predictor.predict_trends(customer_segment_data)
        
        # Generate images
        images = self.image_generator.generate_batch_mockups(
            base_prompts, style_predictions, color_predictions
        )
        
        return images, style_predictions, color_predictions

# FastAPI Integration
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from typing import List, Optional
import uvicorn

app = FastAPI(title="AI Forge E-commerce Image Generator")

# CORS middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"]
)

class GenerationRequest(BaseModel):
    customer_data: List[dict]
    base_prompts: List[str]
    num_images: int = 1

class GenerationResponse(BaseModel):
    success: bool
    message: str
    generated_images: Optional[List[str]] = None
    predicted_styles: Optional[List[str]] = None
    predicted_colors: Optional[List[str]] = None

# Initialize system
ecommerce_system = EcommerceImageAPI()

@app.on_event("startup")
async def startup_event():
    """Initialize system on startup"""
    ecommerce_system.initialize_system()

@app.get("/")
async def root():
    return {"message": "AI Forge E-commerce Image Generator API"}

@app.post("/api/generate-mockups", response_model=GenerationResponse)
async def generate_mockups(request: GenerationRequest):
    """Generate product mockups based on predicted trends"""
    try:
        # Convert customer data to DataFrame
        customer_df = pd.DataFrame(request.customer_data)
    
    # Process customer data
    X_customer = ecommerce_system.data_processor.extract_features(customer_df)
    
    # Generate images
    images, style_preds, color_preds = ecommerce_system.predict_and_generate(
        X_customer,
        request.base_prompts
    )
    
    # Convert images to base64 for API response
    base64_images = []
        for image in images:
            buffered = io.BytesIO()
            image.save(buffered, format="PNG")
        img_str = base64.b64encode(buffered.getvalue()).decode()
        base64_images.append(img_str)
    
    # Get top predicted styles and colors
    top_styles = []
        top_colors = []
        
        for style_pred in style_preds:
            top_indices = np.argsort(style_pred)[-2:]
        top_styles.append([Config.STYLE_CATEGORIES[i] for i in top_indices])
    top_colors = [[Config.COLOR_CATEGORIES[i] for i in np.argsort(color_pred)[-2:]] for color_pred in color_preds]
    
    return GenerationResponse(
        success=True,
        message=f"Successfully generated {len(images)} mockups")
        generated_images=base64_images,
        predicted_styles=top_styles,
        predicted_colors=top_colors
    )
    
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Generation error: {str(e)}")

@app.get("/api/health")
async def health_check():
    return {"status": "healthy", "service": "ecommerce_image_generator"}

def main():
    """Main execution function"""
    print("="*60)
    print("AI FORGE E-COMMERCE IMAGE GENERATOR")
    print("Optimized for 220% YoY demand growth")
    print("="*60)
    
    # Initialize and test the system
    system = EcommerceImageAPI()
    system.initialize_system()
    
    # Sample generation
    print("\nGenerating sample mockups...")
    
    sample_customer_data = [
        {
            'age': 35,
            'income_level': 4,
            'location_urban': 1,
            'gender_encoded': 0,
            'region': 'north_america',
            'device_type': 'mobile',
            'marketing_channel': 'social_media',
            'price': 89.99,
            'category_encoded': 2,
            'brand_popularity': 8
        }
    ]
    
    sample_prompts = [
        "professional product mockup for modern e-commerce website"
    ]
    
    images, styles, colors = system.predict_and_generate(
        pd.DataFrame(sample_customer_data),
        sample_prompts
    )
    
    print(f"Generated {len(images)} images successfully")
    print(f"Predicted top styles: {styles}")
    print(f"Predicted top colors: {colors}")
    
    # Save sample images
    os.makedirs(Config.OUTPUT_DIR, exist_ok=True)
    
    for i, image in enumerate(images):
        image_path = os.path.join(Config.OUTPUT_DIR, f"sample_mockup_{i+1}.png")
        image.save(image_path)
        print(f"Saved sample image: {image_path}")
    
    print("\nSystem ready for production deployment!")
    print("API endpoints available at http://localhost:8000")

if __name__ == "__main__":
    # Run the main function for testing
    main()
    
    # Start the FastAPI server
    uvicorn.run(
        "ecommerce_image_generator:app",
        host="0.0.0.0",
        port=8000,
        reload=True
    )
```