marketing_image_generator.py

```python
#!/usr/bin/env python3
"""
AI Forge Generative Marketing Image Generator
PyTorch-based Stable Diffusion fine-tuned for ad banners with Prophet forecasting
Optimized for 220% YoY demand growth in personalized content
"""

import os
import torch
import pandas as pd
import numpy as np
from diffusers import StableDiffusionPipeline, DPMSolverMultistepScheduler
from transformers import AutoTokenizer, AutoModelForCausalLM
from prophet import Prophet
import matplotlib.pyplot as plt
from PIL import Image, ImageDraw, ImageFont
import io
import base64
import warnings
from datetime import datetime, timedelta
import json
warnings.filterwarnings('ignore')

class CampaignDataProcessor:
    """Process historical campaign data for performance prediction"""
    
    def __init__(self):
        self.feature_columns = []
        
    def load_campaign_data(self, file_path):
        """Load historical campaign performance data"""
        try:
            df = pd.read_csv(file_path)
            print(f"Loaded campaign data with {len(df)} records")
            return df
        except Exception as e:
            print(f"Error loading campaign data: {e}")
            return None
    
    def prepare_time_series_data(self, df):
        """Prepare time series data for Prophet forecasting"""
        # Aggregate daily performance
        df['date'] = pd.to_datetime(df['date'])
        daily_performance = df.groupby('date').agg({
            'clicks': 'sum',
            'impressions': 'sum',
            'conversions': 'sum',
            'spend': 'sum'
        ).reset_index()
        
        # Calculate engagement metrics
        daily_performance['ctr'] = daily_performance['clicks'] / daily_performance['impressions']
        daily_performance['conversion_rate'] = daily_performance['conversions'] / daily_performance['clicks']
        daily_performance['cpc'] = daily_performance['spend'] / daily_performance['clicks']
        daily_performance['engagement_score'] = (
            daily_performance['ctr'] * 0.4 + 
            daily_performance['conversion_rate'] * 0.6
        )
        
        return daily_performance
    
    def extract_campaign_features(self, df):
        """Extract features for campaign performance prediction"""
        features = []
        
        # Campaign metadata
        campaign_features = ['campaign_type', 'target_audience', 'creative_format']
        
        # Performance features
        performance_features = ['ctr', 'conversion_rate', 'cpc']
        
        # Combine features
        for feature in campaign_features + performance_features:
            if feature in df.columns:
                if df[feature].dtype == 'object':
                    dummies = pd.get_dummies(df[feature], prefix=feature)
                    features.append(dummies)
        
        # Date-based features
        df['day_of_week'] = df['date'].dt.dayofweek
        df['month'] = df['date'].dt.month
        df['quarter'] = df['date'].dt.quarter
        
        # Create feature matrix
        X = pd.concat(features, axis=1)
        self.feature_columns = X.columns.tolist()
        
        return X

class CampaignPerformancePredictor:
    """Prophet-based time series forecasting for campaign performance"""
    
    def __init__(self):
        self.model = None
        
    def train_prophet_model(self, df):
        """Train Prophet model on historical campaign data"""
        # Prepare data for Prophet
        prophet_df = df[['date', 'engagement_score']].rename(
            columns={'date': 'ds', 'engagement_score': 'y'
        )
        
        # Initialize and train Prophet model
        self.model = Prophet(
            yearly_seasonality=True,
            weekly_seasonality=True,
            daily_seasonality=False,
            changepoint_prior_scale=0.05
        )
        
        # Add custom seasonality for marketing cycles
        self.model.add_seasonality(name='monthly', period=30.5, fourier_order=5)
        
        # Fit model
        self.model.fit(prophet_df)
        print("Prophet model trained successfully")
        
        return self.model
    
    def forecast_performance(self, periods=30):
        """Forecast campaign performance for future periods"""
        if self.model is None:
            print("Model not trained yet")
            return None
        
        # Create future dataframe
        future = self.model.make_future_dataframe(periods=periods)
        
        # Make predictions
        forecast = self.model.predict(future)
        
        return forecast
    
    def get_performance_insights(self, forecast):
        """Extract insights from forecast for creative conditioning"""
        # Calculate forecast statistics
        latest_prediction = forecast.iloc[-1]
        avg_engagement = forecast['yhat'].mean()
        trend_direction = 'increasing' if forecast['trend'].iloc[-1] > forecast['trend'].iloc[0] else 'decreasing'
        
        # Performance categories based on engagement
        if avg_engagement > 0.7:
            performance_level = 'high'
            creative_brief = "Create vibrant, attention-grabbing visuals with bold colors and dynamic compositions"
        elif avg_engagement > 0.5:
            performance_level = 'medium'
            creative_brief = "Use balanced, professional designs with moderate color saturation"
        else:
            performance_level = 'low'
            creative_brief = "Focus on clear messaging and simple, clean layouts"
        else:
            performance_level = 'baseline'
            creative_brief = "Standard clean designs with clear calls-to-action"
        
        insights = {
            'performance_level': performance_level,
            'predicted_engagement': avg_engagement,
            'trend_direction': trend_direction,
            'creative_brief': creative_brief
        )
        
        return insights

class MarketingImageGenerator:
    """Fine-tuned Stable Diffusion for marketing ad banners"""
    
    def __init__(self, model_name="runwayml/stable-diffusion-v1-5"):
        self.model_name = model_name
        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 and configure Stable Diffusion pipeline"""
        try:
            self.pipeline = StableDiffusionPipeline.from_pretrained(
                self.model_name,
                torch_dtype=torch.float16 if torch.cuda.is_available() else torch.float32
            )
            
            # Optimize scheduler for faster inference
            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_marketing_banner(self, base_prompt, performance_insights, dimensions=(1200, 400)):
        """Generate marketing banner conditioned on performance predictions"""
        
        # Enhance prompt based on performance insights
        enhanced_prompt = self._condition_prompt_on_performance(base_prompt, performance_insights)
        
        # Generate image with conditioning
        with torch.autocast(self.device.type):
            image = self.pipeline(
                enhanced_prompt,
                height=dimensions[1],
                width=dimensions[0],
                num_inference_steps=20,
                guidance_scale=7.5
        ).images[0]
        
        return image
    
    def _condition_prompt_on_performance(self, base_prompt, insights):
        """Condition the generation prompt on performance predictions"""
        
        performance_level = insights['performance_level']
        creative_brief = insights['creative_brief']
        
        # Style conditioning based on performance level
        style_mappings = {
            'high': {
                'style_descriptors': ['vibrant', 'dynamic', 'eye-catching', 'bold'],
                'color_palette': ['saturated colors', 'high contrast', 'vivid tones'],
                'composition': ['professional layout', 'balanced design', 'clear visual hierarchy']
            },
            'medium': {
                'style_descriptors': ['balanced', 'professional', 'moderate saturation'],
                'color_palette': ['moderate contrast', 'professional colors']
            },
            'low': {
                'style_descriptors': ['clean', 'simple', 'minimalist'],
                'composition': ['clear messaging', 'simple layout', 'readable typography']
            },
            'baseline': {
                'style_descriptors': ['standard', 'clear', 'professional'],
                'color_palette': ['neutral colors', 'soft contrast']
            }
        }
        
        style_config = style_mappings.get(performance_level, style_mappings['baseline'])
        
        # Build enhanced prompt
        enhanced_prompt = f"""
        Marketing ad banner for {base_prompt}. 
        Style: {', '.join(style_config['style_descriptors'])}.
        Colors: {', '.join(style_config['color_palette'])}.
        Composition: {', '.join(style_config['composition'])}.
        Creative brief: {creative_brief}.
        Professional quality, high resolution, marketing ready.
        """
        
        return enhanced_prompt.strip()
    
    def generate_campaign_batch(self, campaign_prompts, performance_insights_list, dimensions=(1200, 400)):
        """Generate multiple banners for a campaign"""
        images = []
        
        for i, prompt in enumerate(campaign_prompts):
            insights = performance_insights_list[i] if i < len(performance_insights_list) else insights
            image = self.generate_marketing_banner(prompt, insights, dimensions)
            images.append(image)
        
        return images

class MarketingCreativeConditioner:
    """Condition creative generation on performance predictions and audience data"""
    
    def __init__(self):
        self.performance_categories = {
            'high': {'engagement_threshold': 0.7, 'color_intensity': 'high', 'visual_complexity': 'dynamic']}
    
    def create_creative_brief(self, performance_insights, audience_data=None):
        """Create detailed creative brief for image generation"""
        
        brief = {
            'target_performance': performance_insights['predicted_engagement'],
            'creative_direction': performance_insights['creative_brief'],
            'performance_level': performance_insights['performance_level']
        }
        
        # Add audience-specific conditioning
        if audience_data:
            audience_type = audience_data.get('audience_type', 'general')
            audience_conditioning = {
                'millenials': ['modern', 'trendy', 'social_media_friendly'],
                'professionals': ['sophisticated', 'clean', 'corporate'],
                'families': ['warm', 'friendly', 'approachable']
            }
            
            audience_styles = audience_conditioning.get(audience_type, ['professional'])
            brief['audience_styles'] = audience_styles
        
        return brief

class RealTimeMarketingAPI:
    """FastAPI integration for real-time marketing image generation"""
    
    def __init__(self):
        self.data_processor = CampaignDataProcessor()
        self.performance_predictor = CampaignPerformancePredictor()
        self.image_generator = MarketingImageGenerator()
        self.creative_conditioner = MarketingCreativeConditioner()
        
    def initialize_system(self):
        """Initialize the complete marketing AI system"""
        print("Initializing Marketing AI System...")
        
        # Load and prepare data
        df = self.data_processor.load_campaign_data("./data/marketing_campaigns.csv")
        
        if df is not None:
            # Prepare time series data
            ts_data = self.data_processor.prepare_time_series_data(df)
            
            # Train performance predictor
            self.performance_predictor.train_prophet_model(ts_data)
            print("Performance predictor trained successfully")
        
        # Load image generator
        self.image_generator.load_model()
        
        print("Marketing AI System initialized and ready for real-time use")
    
    def process_campaign_request(self, campaign_data, base_prompts, forecast_periods=30):
        """Complete workflow: predict performance and generate conditioned images"""
        
        # Forecast campaign performance
        forecast = self.performance_predictor.forecast_performance(forecast_periods)
        
        # Get performance insights
        performance_insights = self.performance_predictor.get_performance_insights(forecast)
        
        # Generate images conditioned on predictions
        images = self.image_generator.generate_campaign_batch(
            base_prompts,
            [performance_insights] * len(base_prompts)
        )
        
        return images, performance_insights

# FastAPI Integration
from fastapi import FastAPI, HTTPException, UploadFile, File
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
import uvicorn

app = FastAPI(
    title="AI Forge Marketing Image Generator",
    description="Real-time generative AI for marketing ad banners with Prophet forecasting",
    version="1.0.0"
)

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

# Initialize system
marketing_system = RealTimeMarketingAPI()

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

class CampaignRequest(BaseModel):
    campaign_name: str
    target_audience: str
    campaign_type: str
    base_prompts: List[str]
    forecast_days: int = 30

class CampaignResponse(BaseModel):
    success: bool
    message: str
    generated_images: List[str] = []
    performance_insights: dict = {}
    creative_brief: dict = {}

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

@app.post("/api/generate-campaign-banners", response_model=CampaignResponse)
async def generate_campaign_banners(request: CampaignRequest):
    """Generate marketing banners conditioned on performance forecasts"""
    try:
        # Generate performance forecast
        forecast = marketing_system.performance_predictor.forecast_performance(request.forecast_days)
        
        # Get performance insights
        performance_insights = marketing_system.performance_predictor.get_performance_insights(forecast)
        
        # Create creative brief
        creative_brief = marketing_system.creative_conditioner.create_creative_brief(
            performance_insights,
            {'audience_type': request.target_audience}
        )
        
        # Generate images
        images = marketing_system.image_generator.generate_campaign_batch(
            request.base_prompts,
            [performance_insights] * len(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)
        
        return CampaignResponse(
            success=True,
            message=f"Successfully generated {len(images)} campaign banners")
            generated_images=base64_images,
            performance_insights=performance_insights,
            creative_brief=creative_brief
        )
        
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Generation error: {str(e)}")

@app.post("/api/upload-campaign-data")
async def upload_campaign_data(file: UploadFile = File(...)):
    """Upload historical campaign data for model retraining"""
    try:
        # Save uploaded file
        file_location = f"./data/uploaded_{file.filename}"
        with open(file_location, "wb+") as file_object:
            file_object.write(file.file.read())
        
        # Retrain model with new data
        marketing_system.initialize_system()
        
        return JSONResponse(
            content={"success": True, "message": "Campaign data uploaded and model retrained")
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Upload error: {str(e)}")

@app.get("/api/performance-forecast")
async def get_performance_forecast(days: int = 30):
    """Get performance forecast for the next N days"""
    try:
        forecast = marketing_system.performance_predictor.forecast_performance(days)
        
        # Convert forecast to JSON-serializable format
        forecast_data = forecast[['ds', 'yhat', 'yhat_lower', 'yhat_upper']].to_dict('records')
        
        return JSONResponse(
            content={
                "success": True,
                "forecast_periods': days,
                'predictions': forecast_data
            }
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Forecast error: {str(e)}")

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

def generate_sample_data():
    """Generate sample campaign data for testing"""
    dates = pd.date_range(start='2023-01-01', end='2024-01-01', freq='D')
    
    sample_data = []
    for date in dates:
        sample_data.append({
            'date': date.strftime('%Y-%m-%d'),
            'campaign_type': np.random.choice(['social_media', 'search_engine', 'email', 'display']),
            'clicks': np.random.randint(100, 5000),
            'impressions': np.random.randint(10000, 500000),
            'conversions': np.random.randint(10, 500),
            'spend': np.random.uniform(100, 5000),
            'target_audience': np.random.choice(['millenials', 'professionals', 'families', 'students']),
            'creative_format': np.random.choice(['banner', 'video', 'carousel', 'story']),
            'ctr': np.random.uniform(0.01, 0.05),
            'conversion_rate': np.random.uniform(0.02, 0.15),
            'cpc': np.random.uniform(0.5, 5.0)
        })
    
    df = pd.DataFrame(sample_data)
    df.to_csv("./data/marketing_campaigns.csv", index=False)
    print("Sample campaign data generated")

def main():
    """Main execution function"""
    print("="*70)
    print("AI FORGE MARKETING IMAGE GENERATOR")
    print("Optimized for 220% YoY demand growth in personalized content")
    print("="*70)
    
    # Generate sample data
    generate_sample_data()
    
    # Initialize and test the system
    marketing_system.initialize_system()
    
    # Sample campaign generation
    print("\nGenerating sample campaign banners...")
    
    sample_request = CampaignRequest(
        campaign_name="Summer Sale 2024",
        target_audience="millenials",
        campaign_type="social_media",
        base_prompts=[
            "Promotional banner for summer clothing collection",
            "Marketing visual for seasonal discount campaign",
            "Advertisement for limited-time offer"
    )
    
    # Generate sample images
    images, insights = marketing_system.process_campaign_request(
        {},
        sample_request.base_prompts,
        30
    )
    
    print(f"Generated {len(images)} campaign banners")
    print(f"Performance Insights: {insights}")
    
    # Save sample images
    os.makedirs("./generated_campaigns", exist_ok=True)
    
    for i, image in enumerate(images):
        image_path = f"./generated_campaigns/sample_banner_{i+1}.png")
        image.save(image_path)
        print(f"Saved sample banner: {image_path}")
    
    print("\nSystem ready for real-time marketing campaigns!")
    print("API endpoints available at http://localhost:8000")
    
    # Start the FastAPI server
    uvicorn.run(
        "marketing_image_generator:app",
        host="0.0.0.0",
        port=8000,
        reload=True
    )

if __name__ == "__main__":
    main()
```