Create workflows/image_generator.py

workflows/image_generator.py

@@ -0,0 +1,97 @@
+from prefect import task
+import requests
+from typing import Optional
+from config.settings import settings
+from PIL import Image
+from io import BytesIO
+import base64
+
+@task(retries=2)
+def generate_image_with_dalle(prompt: str) -> Optional[str]:
+    """Generate image using DALL-E via OpenRouter (if available)"""
+    # Note: OpenRouter may not support image generation
+    # This is a placeholder - you might need alternative services
+    pass
+
+@task(retries=2)
+def generate_image_with_canva(prompt: str, template_id: str = None) -> Optional[str]:
+    """Generate image using Canva API"""
+    if not settings.canva_api_key:
+        return None
+    
+    # Canva API integration
+    # This is a simplified example - actual Canva API may differ
+    headers = {
+        "Authorization": f"Bearer {settings.canva_api_key}",
+        "Content-Type": "application/json"
+    }
+    
+    payload = {
+        "design": {
+            "template_id": template_id,
+            "elements": [
+                {
+                    "type": "text",
+                    "content": prompt[:500]  # Truncate if needed
+                }
+            ]
+        }
+    }
+    
+    try:
+        response = requests.post(
+            "https://api.canva.com/v1/designs",
+            headers=headers,
+            json=payload
+        )
+        response.raise_for_status()
+        
+        design_data = response.json()
+        return design_data.get("url")
+    except Exception as e:
+        print(f"Canva API error: {e}")
+        return None
+
+@task(retries=2)
+def generate_placeholder_image(text: str, size: tuple = (1080, 1080)) -> str:
+    """Generate a simple placeholder image"""
+    from PIL import Image, ImageDraw, ImageFont
+    
+    # Create image
+    img = Image.new('RGB', size, color='#4A90E2')
+    draw = ImageDraw.Draw(img)
+    
+    # Add text
+    text_wrapped = text[:100]  # Limit text
+    
+    try:
+        font = ImageFont.truetype("arial.ttf", 40)
+    except:
+        font = ImageFont.load_default()
+    
+    # Center text
+    bbox = draw.textbbox((0, 0), text_wrapped, font=font)
+    text_width = bbox[2] - bbox[0]
+    text_height = bbox[3] - bbox[1]
+    
+    position = ((size[0] - text_width) // 2, (size[1] - text_height) // 2)
+    draw.text(position, text_wrapped, fill='white', font=font)
+    
+    # Save to bytes
+    buffered = BytesIO()
+    img.save(buffered, format="PNG")
+    img_str = base64.b64encode(buffered.getvalue()).decode()
+    
+    return f"data:image/png;base64,{img_str}"
+
+@task
+def select_image_generator(prompt: str) -> str:
+    """Select and use available image generation service"""
+    # Try Canva first
+    image_url = generate_image_with_canva(prompt)
+    
+    if not image_url:
+        # Fallback to placeholder
+        image_url = generate_placeholder_image(prompt)
+    
+    return image_url