""" Thumbnail generation using Fal.ai FLUX 2 Turbo with Gemini 3 fallback. Generates cluster-specific imagery for ANPR parking content. """ import asyncio import io import logging import os from dataclasses import dataclass from pathlib import Path from typing import Optional import fal_client import httpx from google import genai from google.genai import types from PIL import Image from config.settings import get_settings from database.models import Topic logger = logging.getLogger(__name__) # Thumbnail output settings OUTPUT_WIDTH = 1920 OUTPUT_HEIGHT = 1080 OUTPUT_FORMAT = "webp" OUTPUT_QUALITY = 85 # Cluster-specific style contexts for prompt generation CLUSTER_CONTEXTS = { "branchen": { "einzelhandel": "Modern supermarket parking lot with shopping carts, German retail environment, busy but organized", "fitness": "Contemporary gym facility parking area, health-focused atmosphere, modern architecture", "arzt": "Medical clinic parking, clean clinical environment, accessible parking spaces, German healthcare setting", "weg": "Residential apartment complex underground garage, modern German housing, organized parking spaces", "firma": "Corporate office parking structure, business park setting, professional environment", "hotel": "Hotel guest parking, hospitality environment, welcoming entrance", "default": "German commercial parking facility, modern urban environment, organized parking spaces", }, "kosten": "Financial visualization, ROI graphics, cost comparison charts, Euro currency, business analytics aesthetic", "recht": "Legal documentation, German law books, compliance certificates, official stamps, professional legal setting", "technik": "ANPR camera close-up, license plate recognition technology, infrared sensors, technical equipment, sleek modern design", "vergleich": "Split-screen comparison, before/after visualization, side-by-side technology, decision graphics", "howto": "Step-by-step process visualization, instructional graphics, numbered steps, clear guidance imagery", "trends": "Futuristic parking technology, smart city visualization, connected cars, digital innovation", "probleme": "Problem-solution visualization, parking challenges, unauthorized vehicles, resolution imagery", } @dataclass class ThumbnailResult: success: bool file_path: Optional[str] = None provider: Optional[str] = None error: Optional[str] = None cost_usd: float = 0.0 def get_cluster_context(topic: Topic) -> str: """Get the appropriate style context for a topic's cluster.""" cluster = topic.content_cluster.lower() if cluster == "branchen": # Extract industry from title/keywords title_lower = topic.title.lower() keywords_lower = topic.target_keywords.lower() for industry, context in CLUSTER_CONTEXTS["branchen"].items(): if industry in title_lower or industry in keywords_lower: return context return CLUSTER_CONTEXTS["branchen"]["default"] return CLUSTER_CONTEXTS.get(cluster, CLUSTER_CONTEXTS["branchen"]["default"]) async def generate_image_prompt(topic: Topic) -> str: """Generate an optimized image prompt using Gemini Flash.""" settings = get_settings() cluster_context = get_cluster_context(topic) prompt = f"""Generate a detailed image prompt for a blog thumbnail about ANPR parking management. ARTICLE TITLE: {topic.title} CLUSTER: {topic.content_cluster} STYLE CONTEXT: {cluster_context} REQUIREMENTS: - Photorealistic, professional photography style - 16:9 aspect ratio composition - German/European setting (no US-style parking lots) - Modern, clean aesthetic - No text or logos in the image - Appropriate for a professional B2B website - Focus on parking technology or parking environments OUTPUT: A single, detailed image generation prompt (max 200 words) describing the scene, lighting, composition, and style. Do not include any preamble or explanation.""" # Try with primary key, then fallback key on 503 api_keys = [settings.gemini_api_key] if settings.gemini_fallback_api_key: api_keys.append(settings.gemini_fallback_api_key) for i, api_key in enumerate(api_keys): try: client = genai.Client(api_key=api_key) response = client.models.generate_content( model="gemini-3-flash-preview", contents=prompt, ) image_prompt = response.text.strip() logger.debug(f"Generated image prompt: {image_prompt[:100]}...") return image_prompt except Exception as e: is_overloaded = "503" in str(e) or "overloaded" in str(e).lower() if is_overloaded and i < len(api_keys) - 1: logger.warning(f"Gemini 503 with key {i+1}, trying fallback key...") continue logger.error(f"Prompt generation failed: {e}") break # Fallback to basic prompt return f"Professional photograph of a modern German parking facility with ANPR technology, {cluster_context}, photorealistic, 16:9 aspect ratio, clean modern aesthetic" async def generate_with_fal(prompt: str, output_path: Path) -> ThumbnailResult: """Generate image using Fal.ai FLUX 2 Turbo.""" settings = get_settings() os.environ["FAL_KEY"] = settings.fal_api_key try: result = await asyncio.to_thread( fal_client.subscribe, "fal-ai/flux-2/turbo", arguments={ "prompt": prompt, "image_size": {"width": OUTPUT_WIDTH, "height": OUTPUT_HEIGHT}, "num_images": 1, "enable_safety_checker": True, }, ) if not result: return ThumbnailResult(success=False, error="Fal.ai returned empty result") if "images" not in result or not result["images"]: # Check if content was blocked by safety checker nsfw_flags = result.get("has_nsfw_concepts", []) if any(nsfw_flags): return ThumbnailResult(success=False, error="Image blocked by safety checker (NSFW)") return ThumbnailResult(success=False, error=f"No images in result: {list(result.keys())}") image_url = result["images"][0]["url"] # Download image async with httpx.AsyncClient() as client: response = await client.get(image_url) response.raise_for_status() # Save as WebP temp_path = output_path.with_suffix(".tmp") temp_path.write_bytes(response.content) # Convert to WebP with compression with Image.open(temp_path) as img: img.save(output_path, OUTPUT_FORMAT, quality=OUTPUT_QUALITY) temp_path.unlink() return ThumbnailResult( success=True, file_path=str(output_path), provider="fal-ai/flux-2-turbo", cost_usd=0.016, # FLUX 2 Turbo pricing ) except Exception as e: error_msg = str(e) or f"{type(e).__name__}: (no message)" logger.error(f"Fal.ai generation failed: {error_msg}") return ThumbnailResult(success=False, error=error_msg) async def generate_with_gemini(prompt: str, output_path: Path) -> ThumbnailResult: """Generate image using Gemini 3 Pro Image (fallback).""" settings = get_settings() # Try with primary key, then fallback key on 503 api_keys = [settings.gemini_api_key] if settings.gemini_fallback_api_key: api_keys.append(settings.gemini_fallback_api_key) last_error = None for i, api_key in enumerate(api_keys): try: client = genai.Client(api_key=api_key) config = types.GenerateContentConfig( response_modalities=["Image"], temperature=1.0, # Required for Gemini 3 image generation ) response = client.models.generate_content( model="gemini-3-pro-image-preview", contents=f"{prompt}\n\n[Image specifications: aspect ratio 16:9, high quality]", config=config, ) if not response.candidates or not response.candidates[0].content: return ThumbnailResult(success=False, error="No content in response") for part in response.candidates[0].content.parts: if hasattr(part, "inline_data") and part.inline_data: image_data = part.inline_data.data with Image.open(io.BytesIO(image_data)) as img: if img.size != (OUTPUT_WIDTH, OUTPUT_HEIGHT): img = img.resize((OUTPUT_WIDTH, OUTPUT_HEIGHT), Image.Resampling.LANCZOS) img.save(output_path, OUTPUT_FORMAT, quality=OUTPUT_QUALITY) return ThumbnailResult( success=True, file_path=str(output_path), provider="gemini-3-pro-image", cost_usd=0.134, ) return ThumbnailResult(success=False, error="No image in response parts") except Exception as e: last_error = e is_overloaded = "503" in str(e) or "overloaded" in str(e).lower() if is_overloaded and i < len(api_keys) - 1: logger.warning(f"Gemini image 503 with key {i+1}, trying fallback key...") continue logger.error(f"Gemini 3 image generation failed: {e}") break return ThumbnailResult(success=False, error=str(last_error)) async def generate_thumbnail_async(topic: Topic) -> ThumbnailResult: """ Generate thumbnail for article topic. Uses Fal.ai FLUX.2 as primary, Gemini as fallback. """ # Generate optimized prompt prompt = await generate_image_prompt(topic) # Prepare output path output_dir = Path(__file__).parent.parent / "tmp" / "thumbnails" output_dir.mkdir(parents=True, exist_ok=True) output_path = output_dir / f"{topic.slug}.{OUTPUT_FORMAT}" # Try Fal.ai first result = await generate_with_fal(prompt, output_path) if result.success: logger.info(f"Thumbnail generated with Fal.ai: {result.file_path}") return result logger.warning(f"Fal.ai failed, trying Gemini fallback: {result.error}") # Fallback to Gemini result = await generate_with_gemini(prompt, output_path) if result.success: logger.info(f"Thumbnail generated with Gemini: {result.file_path}") return result logger.error(f"All thumbnail generation failed: {result.error}") return result def generate_thumbnail(topic: Topic) -> ThumbnailResult: """Synchronous wrapper for thumbnail generation.""" return asyncio.run(generate_thumbnail_async(topic))