Add Fact-Image module: generate videos with fact text overlay on AI images

modules/fact_image/__init__.py

@@ -0,0 +1,79 @@
+"""
+Fact Image Module
+Generate short videos with fact text overlay on AI-generated images
+"""
+import logging
+from fastapi import FastAPI
+
+logger = logging.getLogger(__name__)
+
+# Module metadata (required by registry)
+MODULE_NAME = "fact_image"
+MODULE_PREFIX = "/api/fact-image"
+MODULE_DESCRIPTION = "Generate fact videos with text overlay on AI-generated images"
+
+
+def register(app: FastAPI, config):
+    """
+    Register the Fact Image module.
+    Called by ModuleRegistry.register_all()
+    """
+    from .router import router, fact_creator as router_fact_creator
+    from .services.fact_creator import FactCreator
+    
+    # Initialize FactCreator with available clients
+    nvidia_client = None
+    cloudflare_client = None
+    pexels_client = None
+    
+    # Reuse NVIDIA client from story_reels if available
+    try:
+        from modules.story_reels.services.nvidia_client import NvidiaClient
+        import os
+        nvidia_key = os.getenv("NVIDIA_API_KEY")
+        if nvidia_key:
+            nvidia_client = NvidiaClient(nvidia_key)
+            logger.info("Fact Image: NVIDIA client initialized")
+    except Exception as e:
+        logger.debug(f"NVIDIA client not available: {e}")
+    
+    # Reuse Cloudflare client from story_reels if available
+    try:
+        from modules.story_reels.services.cloudflare_client import CloudflareClient
+        import os
+        cf_account = os.getenv("CLOUDFLARE_ACCOUNT_ID")
+        cf_token = os.getenv("CLOUDFLARE_API_TOKEN")
+        if cf_account and cf_token:
+            cloudflare_client = CloudflareClient(cf_account, cf_token)
+            logger.info("Fact Image: Cloudflare client initialized")
+    except Exception as e:
+        logger.debug(f"Cloudflare client not available: {e}")
+    
+    # Initialize Pexels client
+    try:
+        from modules.video_creator.services.libraries.pexels import PexelsClient
+        import os
+        pexels_key = os.getenv("PEXELS_API_KEY")
+        if pexels_key:
+            pexels_client = PexelsClient(pexels_key)
+            logger.info("Fact Image: Pexels client initialized")
+    except Exception as e:
+        logger.debug(f"Pexels client not available: {e}")
+    
+    # Create FactCreator instance
+    creator = FactCreator(
+        config=config,
+        nvidia_client=nvidia_client,
+        cloudflare_client=cloudflare_client,
+        pexels_client=pexels_client
+    )
+    
+    # Set the global fact_creator in router
+    import modules.fact_image.router as router_module
+    router_module.fact_creator = creator
+    
+    # Register router
+    app.include_router(router)
+    
+    logger.info(f"Fact Image module registered at {MODULE_PREFIX}")
+    return True

modules/fact_image/router.py

@@ -0,0 +1,127 @@
+"""
+Fact Image Router
+FastAPI endpoints for fact-image video generation
+"""
+import logging
+from pathlib import Path
+from typing import Optional
+
+from fastapi import APIRouter, HTTPException, Depends
+from fastapi.responses import FileResponse, RedirectResponse
+
+from .schemas import (
+    FactImageRequest,
+    FactImageResponse,
+    FactImageStatus,
+    JobStatus
+)
+from .services.fact_creator import FactCreator
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/api/fact-image", tags=["Fact Image"])
+
+# Will be set during app startup
+fact_creator: Optional[FactCreator] = None
+
+
+def get_fact_creator() -> FactCreator:
+    """Dependency to get FactCreator instance"""
+    if fact_creator is None:
+        raise HTTPException(status_code=503, detail="Service not initialized")
+    return fact_creator
+
+
+@router.post("/", response_model=FactImageResponse)
+async def create_fact_image(
+    request: FactImageRequest,
+    creator: FactCreator = Depends(get_fact_creator)
+):
+    """
+    Create a new fact-image video.
+    
+    - **model**: Image generation model (nvidia, cloudflare, pexels)
+    - **image_prompt**: Prompt for background image
+    - **fact_text**: The fact/quote to overlay on the image
+    - **duration**: Video duration in seconds (4-7)
+    """
+    logger.info(f"New fact-image request: model={request.model}, duration={request.duration}s")
+    
+    job_id = creator.add_to_queue(
+        model=request.model,
+        image_prompt=request.image_prompt,
+        fact_text=request.fact_text,
+        duration=request.duration
+    )
+    
+    return FactImageResponse(
+        job_id=job_id,
+        status="processing",
+        status_url=f"/api/fact-image/{job_id}/status",
+        download_url=f"/api/fact-image/{job_id}"
+    )
+
+
+@router.get("/{job_id}/status", response_model=FactImageStatus)
+async def get_status(
+    job_id: str,
+    creator: FactCreator = Depends(get_fact_creator)
+):
+    """Get the status of a fact-image job"""
+    status = creator.get_status(job_id)
+    return FactImageStatus(**status)
+
+
+@router.get("/{job_id}")
+async def download_video(
+    job_id: str,
+    creator: FactCreator = Depends(get_fact_creator)
+):
+    """
+    Download the generated fact-image video.
+    
+    - If cloud-stored: redirects to HF Hub URL
+    - If local: returns the MP4 file
+    """
+    # Check for cloud storage first
+    cloud_file = creator.config.videos_dir_path / f"{job_id}.cloud"
+    if cloud_file.exists():
+        cloud_url = cloud_file.read_text().strip()
+        # Ensure download parameter
+        if "?download=true" not in cloud_url:
+            cloud_url = f"{cloud_url}?download=true"
+        return RedirectResponse(url=cloud_url)
+    
+    # Check for local file
+    video_path = creator.get_video_path(job_id)
+    if video_path and video_path.exists():
+        return FileResponse(
+            path=str(video_path),
+            media_type="video/mp4",
+            filename=f"{job_id}.mp4"
+        )
+    
+    raise HTTPException(status_code=404, detail="Video not found")
+
+
+@router.delete("/{job_id}")
+async def delete_video(
+    job_id: str,
+    creator: FactCreator = Depends(get_fact_creator)
+):
+    """Delete a fact-image video"""
+    # Delete from jobs dict
+    if job_id in creator.jobs:
+        del creator.jobs[job_id]
+    
+    # Delete video file
+    video_path = creator.get_video_path(job_id)
+    if video_path and video_path.exists():
+        video_path.unlink()
+    
+    # Delete cloud metadata
+    cloud_file = creator.config.videos_dir_path / f"{job_id}.cloud"
+    if cloud_file.exists():
+        cloud_file.unlink()
+    
+    return {"message": "Deleted", "job_id": job_id}

modules/fact_image/schemas.py

@@ -0,0 +1,67 @@
+"""
+Fact Image Schemas
+Pydantic models for request/response validation
+"""
+from pydantic import BaseModel, Field
+from typing import Literal, Optional
+from enum import Enum
+
+
+class ImageModel(str, Enum):
+    """Supported image generation models"""
+    nvidia = "nvidia"
+    cloudflare = "cloudflare"
+    pexels = "pexels"
+
+
+class JobStatus(str, Enum):
+    """Job status enum"""
+    queued = "queued"
+    generating_image = "generating_image"
+    adding_text = "adding_text"
+    creating_video = "creating_video"
+    ready = "ready"
+    failed = "failed"
+
+
+class FactImageRequest(BaseModel):
+    """Request schema for creating a fact image video"""
+    model: ImageModel = Field(
+        default=ImageModel.nvidia,
+        description="Image generation model: nvidia, cloudflare, or pexels"
+    )
+    image_prompt: str = Field(
+        ...,
+        min_length=10,
+        max_length=500,
+        description="Prompt for generating the background image"
+    )
+    fact_text: str = Field(
+        ...,
+        min_length=5,
+        max_length=200,
+        description="The fact text to overlay on the image"
+    )
+    duration: int = Field(
+        default=5,
+        ge=4,
+        le=7,
+        description="Video duration in seconds (4-7)"
+    )
+
+
+class FactImageResponse(BaseModel):
+    """Response schema for job creation"""
+    job_id: str
+    status: str
+    status_url: str
+    download_url: str
+
+
+class FactImageStatus(BaseModel):
+    """Response schema for job status"""
+    job_id: str
+    status: JobStatus
+    progress: int = 0
+    video_url: Optional[str] = None
+    error: Optional[str] = None

modules/fact_image/services/__init__.py

@@ -0,0 +1 @@
+# Fact Image Services

modules/fact_image/services/fact_creator.py

@@ -0,0 +1,328 @@
+"""
+Fact Creator Service
+Main orchestrator for generating fact-image videos
+"""
+import asyncio
+import logging
+import uuid
+import time
+from pathlib import Path
+from typing import Dict, List, Optional
+from datetime import datetime
+
+from ..schemas import JobStatus, ImageModel
+from .text_overlay import TextOverlay
+
+logger = logging.getLogger(__name__)
+
+
+# Master prompt for image generation
+IMAGE_SYSTEM_PROMPT = """You are an AI image generator for short-form vertical videos (Reels, TikTok, Shorts).
+
+Your task:
+Generate ONE high-quality image based on the user-provided `image_prompt`. The image must be ideal for placing text on top. 
+
+OUTPUT REQUIREMENTS:
+- Resolution: 1080x1920 (9:16 vertical)
+- Clean background, balanced lighting, no clutter
+- Ensure 40–60% empty or soft area for text overlay
+- Avoid too many distracting details
+- Aesthetic cinematic style
+- Professional color grading
+- Sharp subject, soft background depth
+- Keep main visual elements aligned in center or rule-of-thirds
+
+CHARACTERISTICS:
+- High-contrast but not noisy
+- Good separation of foreground and background
+- Smooth gradients, soft shadows
+- Aesthetic and modern visual style
+- Works well with motivational or psychology fact text overlay
+
+DO NOT:
+- Add any text in the image
+- Add watermarks
+- Add captions
+- Add UI elements
+- Crop faces weirdly
+- Use extreme zoom
+
+STYLE GUIDELINES:
+- Vibrant, cinematic, trending Instagram/TikTok look
+- Strong composition
+- Beautiful lighting (soft, natural, or dramatic depending on prompt)
+- Enhances emotional feel of the fact message
+
+[User Prompt]
+{image_prompt}"""
+
+
+class FactCreator:
+    """
+    Main orchestrator for fact-image video generation.
+    
+    Pipeline:
+    1. Generate clean image (NVIDIA/Cloudflare/Pexels)
+    2. Add text overlay (PIL)
+    3. Create short video (MoviePy)
+    4. Upload to cloud (optional)
+    """
+    
+    # Video settings
+    TARGET_WIDTH = 1080
+    TARGET_HEIGHT = 1920
+    FPS = 24
+    FADE_DURATION = 0.3
+    
+    def __init__(
+        self,
+        config,
+        nvidia_client=None,
+        cloudflare_client=None,
+        pexels_client=None
+    ):
+        self.config = config
+        self.nvidia = nvidia_client
+        self.cloudflare = cloudflare_client
+        self.pexels = pexels_client
+        self.text_overlay = TextOverlay()
+        
+        # Job tracking
+        self.jobs: Dict[str, Dict] = {}
+        self.queue: List[Dict] = []
+        self.processing = False
+    
+    def add_to_queue(
+        self,
+        model: ImageModel,
+        image_prompt: str,
+        fact_text: str,
+        duration: int = 5
+    ) -> str:
+        """
+        Add fact-image job to queue.
+        
+        Returns:
+            job_id for tracking
+        """
+        job_id = str(uuid.uuid4()).replace('-', '')[:16]
+        
+        job = {
+            "id": job_id,
+            "model": model,
+            "image_prompt": image_prompt,
+            "fact_text": fact_text,
+            "duration": duration,
+            "status": JobStatus.queued,
+            "progress": 0,
+            "created_at": datetime.now().isoformat(),
+            "video_url": None,
+            "error": None
+        }
+        
+        self.jobs[job_id] = job
+        self.queue.append(job)
+        
+        logger.info(f"Added job {job_id} to queue. Queue length: {len(self.queue)}")
+        
+        # Start processing if not already running
+        if not self.processing:
+            asyncio.create_task(self.process_queue())
+        
+        return job_id
+    
+    async def process_queue(self):
+        """Process jobs in queue"""
+        if self.processing:
+            return
+        
+        self.processing = True
+        
+        try:
+            while self.queue:
+                job = self.queue[0]
+                job_id = job["id"]
+                
+                logger.info(f"Processing job {job_id}")
+                
+                try:
+                    await self._process_job(job)
+                    job["status"] = JobStatus.ready
+                    job["progress"] = 100
+                    logger.info(f"Job {job_id} completed successfully")
+                except Exception as e:
+                    logger.error(f"Job {job_id} failed: {e}", exc_info=True)
+                    job["status"] = JobStatus.failed
+                    job["error"] = str(e)
+                finally:
+                    self.queue.pop(0)
+        finally:
+            self.processing = False
+    
+    async def _process_job(self, job: Dict):
+        """Process a single fact-image job"""
+        job_id = job["id"]
+        temp_dir = self.config.temp_dir_path / job_id
+        temp_dir.mkdir(parents=True, exist_ok=True)
+        
+        try:
+            # ====================
+            # Step 1: Generate Image
+            # ====================
+            job["status"] = JobStatus.generating_image
+            job["progress"] = 10
+            logger.info(f"[{job_id}] Generating image with {job['model']}...")
+            
+            image_path = temp_dir / "base_image.png"
+            
+            # Build full prompt
+            full_prompt = IMAGE_SYSTEM_PROMPT.format(image_prompt=job["image_prompt"])
+            
+            if job["model"] == ImageModel.nvidia and self.nvidia:
+                self.nvidia.generate_and_save(
+                    prompt=full_prompt,
+                    output_path=image_path,
+                    width=self.TARGET_WIDTH,
+                    height=self.TARGET_HEIGHT
+                )
+            elif job["model"] == ImageModel.cloudflare and self.cloudflare:
+                self.cloudflare.generate_and_save(
+                    prompt=full_prompt,
+                    output_path=image_path,
+                    width=self.TARGET_WIDTH,
+                    height=self.TARGET_HEIGHT
+                )
+            elif job["model"] == ImageModel.pexels and self.pexels:
+                # Pexels uses search, not generation
+                self.pexels.search_and_download(
+                    query=job["image_prompt"],
+                    output_path=image_path,
+                    orientation="portrait"
+                )
+            else:
+                # Fallback to any available client
+                if self.nvidia:
+                    self.nvidia.generate_and_save(full_prompt, image_path, self.TARGET_WIDTH, self.TARGET_HEIGHT)
+                elif self.cloudflare:
+                    self.cloudflare.generate_and_save(full_prompt, image_path, self.TARGET_WIDTH, self.TARGET_HEIGHT)
+                else:
+                    raise Exception("No image generation client available!")
+            
+            job["progress"] = 40
+            
+            # ====================
+            # Step 2: Add Text Overlay
+            # ====================
+            job["status"] = JobStatus.adding_text
+            logger.info(f"[{job_id}] Adding text overlay...")
+            
+            overlay_path = temp_dir / "overlay_image.png"
+            self.text_overlay.add_text(
+                image_path=image_path,
+                text=job["fact_text"],
+                output_path=overlay_path
+            )
+            
+            job["progress"] = 60
+            
+            # ====================
+            # Step 3: Create Video
+            # ====================
+            job["status"] = JobStatus.creating_video
+            logger.info(f"[{job_id}] Creating {job['duration']}s video...")
+            
+            output_path = self.config.videos_dir_path / f"{job_id}.mp4"
+            await self._create_video(overlay_path, output_path, job["duration"])
+            
+            job["video_url"] = str(output_path)
+            job["progress"] = 90
+            
+            # ====================
+            # Step 4: Upload to Cloud (Optional)
+            # ====================
+            from modules.shared.services.hf_storage import get_hf_client
+            hf_client = get_hf_client()
+            
+            if hf_client:
+                logger.info(f"[{job_id}] Uploading to HF Hub...")
+                cloud_url = hf_client.upload_video(output_path, "fact_image")
+                if cloud_url:
+                    job["video_url"] = cloud_url
+                    job["storage"] = "cloud"
+                    # Save cloud URL to metadata file
+                    cloud_file = output_path.with_suffix('.cloud')
+                    cloud_file.write_text(cloud_url)
+                    # Delete local file
+                    output_path.unlink()
+                    logger.info(f"[{job_id}] Uploaded to cloud, local file deleted")
+            
+            logger.info(f"[{job_id}] Video ready: {job['video_url']}")
+            
+        finally:
+            # Cleanup temp files
+            import shutil
+            if temp_dir.exists():
+                shutil.rmtree(temp_dir, ignore_errors=True)
+    
+    async def _create_video(self, image_path: Path, output_path: Path, duration: int):
+        """Create video from single image with fade effects"""
+        from moviepy.editor import ImageClip
+        
+        # Create image clip
+        clip = ImageClip(str(image_path)).set_duration(duration)
+        
+        # Add fade in/out
+        clip = clip.fadein(self.FADE_DURATION)
+        clip = clip.fadeout(self.FADE_DURATION)
+        
+        # Write video
+        logger.info(f"Writing video: {duration}s, fade in/out {self.FADE_DURATION}s")
+        clip.write_videofile(
+            str(output_path),
+            fps=self.FPS,
+            codec='libx264',
+            audio=False,  # No audio for fact videos
+            preset='medium',
+            ffmpeg_params=[
+                '-pix_fmt', 'yuv420p',
+                '-movflags', '+faststart',
+                '-profile:v', 'baseline',
+                '-level', '3.0'
+            ]
+        )
+        
+        clip.close()
+    
+    def get_status(self, job_id: str) -> Dict:
+        """Get job status"""
+        job = self.jobs.get(job_id)
+        
+        if not job:
+            # Check for .cloud file (cloud-stored video)
+            cloud_file = self.config.videos_dir_path / f"{job_id}.cloud"
+            if cloud_file.exists():
+                return {
+                    "job_id": job_id,
+                    "status": JobStatus.ready,
+                    "progress": 100,
+                    "video_url": cloud_file.read_text().strip()
+                }
+            
+            return {
+                "job_id": job_id,
+                "status": JobStatus.failed,
+                "progress": 0,
+                "error": "Job not found"
+            }
+        
+        return {
+            "job_id": job["id"],
+            "status": job["status"],
+            "progress": job.get("progress", 0),
+            "video_url": job.get("video_url"),
+            "error": job.get("error")
+        }
+    
+    def get_video_path(self, job_id: str) -> Optional[Path]:
+        """Get video file path"""
+        return self.config.videos_dir_path / f"{job_id}.mp4"

modules/fact_image/services/text_overlay.py

@@ -0,0 +1,186 @@
+"""
+Text Overlay Service
+Renders text on images using PIL/Pillow
+"""
+import logging
+import textwrap
+from pathlib import Path
+from typing import Tuple, Optional
+from PIL import Image, ImageDraw, ImageFont
+
+logger = logging.getLogger(__name__)
+
+
+class TextOverlay:
+    """
+    Service for adding text overlay to images.
+    Optimized for fact/motivational content on vertical videos.
+    """
+    
+    # Default settings
+    TARGET_WIDTH = 1080
+    TARGET_HEIGHT = 1920
+    TEXT_AREA_TOP = 0.55  # Text starts at 55% from top
+    TEXT_AREA_BOTTOM = 0.90  # Text ends at 90% from top
+    PADDING_X = 60  # Horizontal padding
+    
+    def __init__(self, font_path: Optional[str] = None):
+        """
+        Initialize text overlay service.
+        
+        Args:
+            font_path: Path to custom font file (optional)
+        """
+        self.font_path = font_path
+        self._font_cache = {}
+    
+    def _get_font(self, size: int) -> ImageFont.FreeTypeFont:
+        """Get font at specified size (cached)"""
+        if size not in self._font_cache:
+            try:
+                if self.font_path and Path(self.font_path).exists():
+                    self._font_cache[size] = ImageFont.truetype(self.font_path, size)
+                else:
+                    # Try common system fonts
+                    for font_name in ['DejaVuSans-Bold.ttf', 'Arial.ttf', 'Roboto-Bold.ttf']:
+                        try:
+                            self._font_cache[size] = ImageFont.truetype(font_name, size)
+                            break
+                        except:
+                            continue
+                    else:
+                        # Fallback to default
+                        self._font_cache[size] = ImageFont.load_default()
+            except Exception as e:
+                logger.warning(f"Font loading failed: {e}, using default")
+                self._font_cache[size] = ImageFont.load_default()
+        
+        return self._font_cache[size]
+    
+    def _wrap_text(self, text: str, max_words_per_line: int = 5) -> str:
+        """
+        Wrap text for optimal display.
+        
+        Rules:
+        - Max 5-6 words per line
+        - Natural line breaks
+        """
+        words = text.split()
+        lines = []
+        current_line = []
+        
+        for word in words:
+            current_line.append(word)
+            if len(current_line) >= max_words_per_line:
+                lines.append(' '.join(current_line))
+                current_line = []
+        
+        if current_line:
+            lines.append(' '.join(current_line))
+        
+        return '\n'.join(lines)
+    
+    def _calculate_font_size(
+        self,
+        text: str,
+        draw: ImageDraw.ImageDraw,
+        max_width: int,
+        max_height: int,
+        min_size: int = 40,
+        max_size: int = 80
+    ) -> int:
+        """Calculate optimal font size to fit text in given area"""
+        for size in range(max_size, min_size - 1, -2):
+            font = self._get_font(size)
+            wrapped = self._wrap_text(text)
+            
+            # Get text bounding box
+            bbox = draw.multiline_textbbox((0, 0), wrapped, font=font)
+            text_width = bbox[2] - bbox[0]
+            text_height = bbox[3] - bbox[1]
+            
+            if text_width <= max_width and text_height <= max_height:
+                return size
+        
+        return min_size
+    
+    def add_text(
+        self,
+        image_path: Path,
+        text: str,
+        output_path: Path,
+        text_color: Tuple[int, int, int] = (255, 255, 255),
+        shadow_color: Tuple[int, int, int] = (0, 0, 0),
+        shadow_offset: int = 3
+    ) -> Path:
+        """
+        Add text overlay to image.
+        
+        Args:
+            image_path: Path to input image
+            text: Text to overlay
+            output_path: Path for output image
+            text_color: RGB color for text (default: white)
+            shadow_color: RGB color for shadow (default: black)
+            shadow_offset: Shadow offset in pixels
+            
+        Returns:
+            Path to output image
+        """
+        logger.info(f"Adding text overlay: {text[:50]}...")
+        
+        # Load image
+        img = Image.open(image_path).convert('RGBA')
+        
+        # Resize if needed
+        if img.size != (self.TARGET_WIDTH, self.TARGET_HEIGHT):
+            img = img.resize((self.TARGET_WIDTH, self.TARGET_HEIGHT), Image.LANCZOS)
+        
+        # Create drawing context
+        draw = ImageDraw.Draw(img)
+        
+        # Calculate text area
+        text_area_y_start = int(self.TARGET_HEIGHT * self.TEXT_AREA_TOP)
+        text_area_y_end = int(self.TARGET_HEIGHT * self.TEXT_AREA_BOTTOM)
+        max_width = self.TARGET_WIDTH - (2 * self.PADDING_X)
+        max_height = text_area_y_end - text_area_y_start
+        
+        # Calculate optimal font size
+        font_size = self._calculate_font_size(text, draw, max_width, max_height)
+        font = self._get_font(font_size)
+        
+        # Wrap text
+        wrapped_text = self._wrap_text(text)
+        
+        # Calculate text position (centered)
+        bbox = draw.multiline_textbbox((0, 0), wrapped_text, font=font)
+        text_width = bbox[2] - bbox[0]
+        text_height = bbox[3] - bbox[1]
+        
+        x = (self.TARGET_WIDTH - text_width) // 2
+        y = text_area_y_start + (max_height - text_height) // 2
+        
+        # Draw shadow
+        draw.multiline_text(
+            (x + shadow_offset, y + shadow_offset),
+            wrapped_text,
+            font=font,
+            fill=shadow_color,
+            align='center'
+        )
+        
+        # Draw main text
+        draw.multiline_text(
+            (x, y),
+            wrapped_text,
+            font=font,
+            fill=text_color,
+            align='center'
+        )
+        
+        # Save output
+        img = img.convert('RGB')
+        img.save(output_path, 'PNG', quality=95)
+        
+        logger.info(f"Text overlay saved: {output_path}")
+        return output_path