refactor: remove worker pool, use direct fal.ai API calls

- Remove services/worker/ and priority_worker_pool.py
- Remove services/gemini_service/ (replaced by fal_service)
- Update gemini.py to call FalService directly
- Use fal.ai submit() for non-blocking job start
- Use fal.ai status() for non-blocking status check
- Remove worker startup/shutdown from app.py
- All API responses now immediate, no background processing

.env.example

@@ -97,6 +97,21 @@ JOB_PER_API_KEY=2
 # Enable mock mode for testing without consuming API credits
 # GEMINI_MOCK_MODE=true
 
+# -----------------------------------------------------------------------------
+# AI Provider Configuration
+# -----------------------------------------------------------------------------
+# Which AI provider to use for video generation
+# Options: "google" (Gemini/Veo) or "fal" (fal.ai)
+AI_PROVIDER=google
+
+# Fal.ai API Key (required if AI_PROVIDER=fal)
+# Get from: https://fal.ai/dashboard/keys
+# Note: fal_client expects this env var to be named FAL_KEY
+# FAL_KEY=your-fal-api-key
+
+# Enable mock mode for fal.ai testing without consuming API credits
+# FAL_MOCK_MODE=true
+
 # -----------------------------------------------------------------------------
 # Email Configuration (Optional)
 # -----------------------------------------------------------------------------

app.py

@@ -159,24 +159,12 @@ async def lifespan(app: FastAPI):
     )
     logger.info("✅ Audit Service configured")
     
-    # Register API Key Service configuration
-    from services.gemini_service import APIKeyServiceConfig
-    APIKeyServiceConfig.register(
-        rotation_strategy="least_used",  # or "round_robin"
-        cooldown_seconds=60,  # Wait 1 min after quota error
-        max_requests_per_minute=60,
-        retry_on_quota_error=True  # Auto-retry with different key
-    )
-    logger.info("✅ API Key Service configured")
-    
-    # Worker Pool Section
+    # Job Processing Info
     logger.info("")
-    logger.info("👷 [WORKER POOL]")
+    logger.info("⚡ [JOB PROCESSING]")
+    logger.info("✅ Using inline processor (fire-and-forget async)")
     
-    # Start background job worker
-    from services.gemini_service import start_worker, stop_worker
-    await start_worker()
-    logger.info("✅ Worker pool started")
+
     
     # Log CORS configuration
     allowed_origins = os.getenv("CORS_ORIGINS").split(",")
@@ -189,17 +177,15 @@ async def lifespan(app: FastAPI):
     logger.info("═" * 60)
     logger.info("    🚀 API Gateway Ready")
     logger.info("    • Database: ✅ Ready")
-    logger.info("    • Services: 5 initialized (DB, Auth, Credit, Audit, API Key)")
-    logger.info("    • Workers: 15 active")
+    logger.info("    • Services: 4 initialized (DB, Auth, Credit, Audit)")
+    logger.info("    • Processing: Inline (no workers)")
     logger.info("    • Endpoint: http://0.0.0.0:8000")
     logger.info("═" * 60)
     logger.info("")
     
     yield
     
-    # Stop background job worker
-    await stop_worker()
-    logger.info("Background job worker stopped")
+    # No worker cleanup needed - inline processor uses fire-and-forget tasks
     
     # Shutdown: Upload DB to Drive
     logger.info("Shutdown: Uploading database to Google Drive...")
@@ -230,10 +216,6 @@ from services.audit_service import AuditMiddleware
 app.add_middleware(AuditMiddleware)
 
 
-from services.gemini_service import APIKeyMiddleware
-app.add_middleware(APIKeyMiddleware)
-
-
 from services.auth_service import AuthMiddleware
 app.add_middleware(AuthMiddleware)
 

requirements.txt

@@ -1,19 +1,18 @@
-# FastAPI URL Blink Application Dependencies
-fastapi>=0.104.0
-uvicorn[standard]>=0.24.0
-sqlalchemy>=2.0.0
-aiosqlite>=0.19.0
-cryptography>=41.0.0
-pydantic>=2.0.0
-httpx>=0.25.0
-
-passlib[bcrypt]>=1.7.4
-email-validator>=2.0.0
-python-dotenv>=1.0.0
-google-api-python-client>=2.0.0
-google-auth-oauthlib>=1.0.0
-google-auth-httplib2>=0.1.0
-google-genai>=1.0.0
-PyJWT>=2.8.0
-razorpay>=1.4.0
+fastapi==0.115.12
+uvicorn[standard]==0.34.3
+sqlalchemy==2.0.41
+aiosqlite==0.21.0
+cryptography==45.0.5
+pydantic==2.11.7
+httpx==0.28.1
 
+passlib[bcrypt]==1.7.4
+email-validator==2.2.0
+python-dotenv==1.1.1
+google-api-python-client==2.187.0
+google-auth-oauthlib==1.2.1
+google-auth-httplib2==0.2.0
+google-genai==1.57.0
+PyJWT==2.10.1
+razorpay==2.0.0
+fal-client==0.5.9

routers/gemini.py

@@ -1,5 +1,6 @@
 """
-Gemini Router - API endpoints for Gemini AI services.
+Video Router - API endpoints for AI video generation services.
+Uses fal.ai for video generation.
 """
 import os
 import uuid
@@ -12,10 +13,10 @@ from sqlalchemy import select, func
 
 from core.database import get_db
 from core.models import User, GeminiJob
-from services.gemini_service import MODELS, DOWNLOADS_DIR
+from services.fal_service import MODELS
 from datetime import datetime
 
-router = APIRouter(prefix="/gemini", tags=["gemini"])
+router = APIRouter(prefix="/gemini", tags=["video"])
 
 
 
@@ -71,18 +72,15 @@ async def create_job(
     input_data: dict,
     credits_reserved: int = 0
 ) -> GeminiJob:
-    """Create a new job in the queue."""
-    from services.gemini_service.job_processor import get_priority_for_job_type, get_pool
-    
+    """Create job and start processing on fal.ai (non-blocking)."""
     job_id = f"job_{uuid.uuid4().hex[:16]}"
-    priority = get_priority_for_job_type(job_type)
     
     job = GeminiJob(
         job_id=job_id,
         user_id=user.id,
         job_type=job_type,
         status="queued",
-        priority=priority,
+        priority="slow",
         input_data=input_data,
         credits_reserved=credits_reserved
     )
@@ -90,8 +88,22 @@ async def create_job(
     await db.commit()
     await db.refresh(job)
     
-
-    get_pool().notify_new_job(priority)
+    # Start fal.ai job immediately (non-blocking)
+    if job_type == "video":
+        from services.fal_service import FalService
+        fal = FalService()
+        result = await fal.start_video_generation(
+            base64_image=input_data.get("base64_image", ""),
+            mime_type=input_data.get("mime_type", "image/jpeg"),
+            prompt=input_data.get("prompt", "Animate this image with subtle movement."),
+            aspect_ratio=input_data.get("aspect_ratio", "16:9"),
+            resolution=input_data.get("resolution", "720p"),
+        )
+        job.status = "processing"
+        job.started_at = datetime.utcnow()
+        job.third_party_id = result.get("fal_request_id")
+        job.api_response = result
+        await db.commit()
     
     return job
 
@@ -322,7 +334,7 @@ async def get_job_status(
     req: Request,
     db: AsyncSession = Depends(get_db)
 ):
-    """Get job status and update if processing."""
+    """Get job status. Checks fal.ai if processing (non-blocking)."""
     user = req.state.user
     query = select(GeminiJob).where(
         GeminiJob.job_id == job_id,
@@ -337,13 +349,30 @@ async def get_job_status(
             detail="Job not found"
         )
     
-
+    # Check fal.ai status if processing (non-blocking)
     if job.status == "processing" and job.job_type == "video" and job.third_party_id:
-        from services.gemini_service.job_processor import GeminiJobProcessor
-        processor = GeminiJobProcessor()
-        job = await processor.check_status(job, db)
-        await db.commit()
-        await db.refresh(job)
+        from services.fal_service import FalService
+        fal = FalService()
+        result = await fal.check_video_status(job.third_party_id)
+        
+        if result.get("done"):
+            if result.get("status") == "completed":
+                job.status = "completed"
+                job.output_data = {"video_url": result.get("video_url")}
+                job.completed_at = datetime.utcnow()
+            else:
+                job.status = "failed"
+                job.error_message = result.get("error", "Unknown error")
+                job.completed_at = datetime.utcnow()
+            
+            # Handle credits on completion
+            if job.credits_reserved > 0:
+                try:
+                    from services.credit_service.credit_manager import handle_job_completion
+                    await handle_job_completion(db, job)
+                except Exception:
+                    pass
+            await db.commit()
     
     response = {
         "success": True,
@@ -353,23 +382,17 @@ async def get_job_status(
         "created_at": job.created_at.isoformat() if job.created_at else None,
         "credits_remaining": user.credits
     }
-    
 
     if job.job_type == "video" and job.input_data:
         response["prompt"] = job.input_data.get("prompt")
     
-    if job.status == "queued":
-        response["position"] = await get_queue_position(db, job.job_id)
-    
     if job.status == "processing":
         response["started_at"] = job.started_at.isoformat() if job.started_at else None
     
     if job.status == "completed":
         response["completed_at"] = job.completed_at.isoformat() if job.completed_at else None
-        
-
-        if job.output_data and "prompt" in job.output_data:
-            response["prompt"] = job.output_data["prompt"]
+        if job.output_data and job.output_data.get("video_url"):
+            response["video_url"] = job.output_data.get("video_url")
     
     if job.status == "failed":
         response["error"] = job.error_message

services/fal_service/__init__.py

@@ -0,0 +1,17 @@
+"""
+Fal.ai Service for video generation.
+Provides access to Veo 3.1 and other models through fal.ai's unified API.
+"""
+from services.fal_service.api_client import (
+    FalService,
+    MODELS,
+    MOCK_MODE,
+    get_fal_api_key,
+)
+
+__all__ = [
+    "FalService",
+    "MODELS",
+    "MOCK_MODE",
+    "get_fal_api_key",
+]

services/fal_service/api_client.py

@@ -0,0 +1,254 @@
+"""
+Fal.ai Service for video generation.
+Python implementation using fal-client SDK.
+Uses server-side API key from environment.
+"""
+import asyncio
+import logging
+import os
+from typing import Optional, Literal
+
+logger = logging.getLogger(__name__)
+
+# Model names - easily configurable
+MODELS = {
+    "video_generation": "fal-ai/veo3.1/fast/image-to-video"
+}
+
+# Type aliases
+AspectRatio = Literal["16:9", "9:16", "auto"]
+Resolution = Literal["720p", "1080p"]
+
+# Mock mode for local testing (set FAL_MOCK_MODE=true to skip real API calls)
+MOCK_MODE = os.getenv("FAL_MOCK_MODE", "false").lower() == "true"
+
+# Sample video URL for mock mode
+MOCK_VIDEO_URL = "https://v3b.fal.media/files/mock/mock-video.mp4"
+
+
+def get_fal_api_key() -> str:
+    """Get Fal.ai API key from environment."""
+    api_key = os.getenv("FAL_KEY")
+    if not api_key:
+        raise ValueError("Server Authentication Error: FAL_KEY not configured")
+    return api_key
+
+
+class FalService:
+    """
+    Fal.ai Service for video generation.
+    Uses server-side API key from environment (FAL_KEY).
+    """
+
+    def __init__(self, api_key: Optional[str] = None):
+        """Initialize the Fal.ai client with API key from env or provided."""
+        self.api_key = api_key or get_fal_api_key()
+        # fal_client reads from FAL_KEY env var by default
+        # Set it explicitly if a custom key is provided
+        if api_key:
+            os.environ["FAL_KEY"] = api_key
+
+    def _handle_api_error(self, error: Exception, context: str):
+        """Handle API errors with descriptive messages."""
+        msg = str(error)
+        if "401" in msg or "Unauthorized" in msg:
+            raise ValueError(
+                f"Authentication failed ({context}). Check your FAL_KEY is valid."
+            )
+        if "402" in msg or "Payment Required" in msg:
+            raise ValueError(
+                f"Insufficient credits ({context}). Add credits at fal.ai."
+            )
+        if "429" in msg or "Rate limit" in msg.lower():
+            raise ValueError(
+                f"Rate limit exceeded ({context}). Wait and retry."
+            )
+        raise error
+
+    async def start_video_generation(
+        self,
+        base64_image: str,
+        mime_type: str,
+        prompt: str,
+        aspect_ratio: AspectRatio = "16:9",
+        resolution: Resolution = "720p",
+        number_of_videos: int = 1
+    ) -> dict:
+        """
+        Start video generation using Fal.ai Veo 3.1 model.
+        Unlike Gemini, fal.ai subscribe() handles polling internally,
+        so this returns the completed video directly.
+        
+        Returns dict with:
+            - fal_request_id: Request ID for reference
+            - done: Always True (fal.ai waits for completion)
+            - status: "completed" or "failed"
+            - video_url: URL to the generated video
+        """
+        # Mock mode for testing without API credits
+        if MOCK_MODE:
+            import uuid
+            mock_request_id = f"mock_fal_{uuid.uuid4().hex[:16]}"
+            logger.info(f"[MOCK MODE] Video generation: {mock_request_id}")
+            await asyncio.sleep(2)  # Simulate API delay
+            return {
+                "fal_request_id": mock_request_id,
+                "done": True,
+                "status": "completed",
+                "video_url": MOCK_VIDEO_URL
+            }
+
+        try:
+            import fal_client
+
+            # Use submit() instead of subscribe() - returns immediately without waiting
+            # This starts the job and returns a request_id for status checking
+            handle = await asyncio.to_thread(
+                fal_client.submit,
+                MODELS["video_generation"],
+                arguments={
+                    "prompt": prompt,
+                    "image_url": f"data:{mime_type};base64,{base64_image}",
+                    "aspect_ratio": aspect_ratio,
+                    "resolution": resolution,
+                    "generate_audio": True,
+                },
+            )
+
+            # Get the request ID from the handle
+            request_id = handle.request_id if hasattr(handle, 'request_id') else str(handle)
+
+            return {
+                "fal_request_id": request_id,
+                "done": False,
+                "status": "processing",
+            }
+
+        except ImportError:
+            raise ValueError(
+                "fal-client package not installed. Run: pip install fal-client"
+            )
+        except Exception as error:
+            self._handle_api_error(error, MODELS["video_generation"])
+
+    async def check_video_status(self, fal_request_id: str) -> dict:
+        """
+        Check the status of a video generation request.
+        Returns immediately with current status (does not wait).
+        """
+        # Mock mode for testing
+        if MOCK_MODE:
+            import random
+            # Simulate completion after a few checks
+            if random.random() > 0.7:
+                return {
+                    "fal_request_id": fal_request_id,
+                    "done": True,
+                    "status": "completed",
+                    "video_url": MOCK_VIDEO_URL
+                }
+            return {
+                "fal_request_id": fal_request_id,
+                "done": False,
+                "status": "processing"
+            }
+
+        try:
+            import fal_client
+
+            # Get status without waiting
+            status = await asyncio.to_thread(
+                fal_client.status,
+                MODELS["video_generation"],
+                fal_request_id,
+                with_logs=False
+            )
+
+            # Check if completed
+            if hasattr(status, 'status'):
+                if status.status == "COMPLETED":
+                    # Get the result
+                    result = await asyncio.to_thread(
+                        fal_client.result,
+                        MODELS["video_generation"],
+                        fal_request_id
+                    )
+                    
+                    # Extract video URL
+                    video_url = None
+                    if isinstance(result, dict) and "video" in result:
+                        video_url = result["video"].get("url")
+                    elif hasattr(result, "video") and hasattr(result.video, "url"):
+                        video_url = result.video.url
+
+                    return {
+                        "fal_request_id": fal_request_id,
+                        "done": True,
+                        "status": "completed",
+                        "video_url": video_url
+                    }
+                elif status.status == "FAILED":
+                    return {
+                        "fal_request_id": fal_request_id,
+                        "done": True,
+                        "status": "failed",
+                        "error": getattr(status, 'error', 'Unknown error')
+                    }
+                else:
+                    # Still processing (IN_QUEUE, IN_PROGRESS)
+                    return {
+                        "fal_request_id": fal_request_id,
+                        "done": False,
+                        "status": "processing"
+                    }
+            
+            # Fallback - assume still processing
+            return {
+                "fal_request_id": fal_request_id,
+                "done": False,
+                "status": "processing"
+            }
+
+        except ImportError:
+            raise ValueError(
+                "fal-client package not installed. Run: pip install fal-client"
+            )
+        except Exception as error:
+            logger.error(f"Error checking status for {fal_request_id}: {error}")
+            return {
+                "fal_request_id": fal_request_id,
+                "done": False,
+                "status": "processing",
+                "error": str(error)
+            }
+
+    async def download_video(self, video_url: str, request_id: str) -> str:
+        """
+        Download video from fal.ai to local storage.
+        Returns the local filename.
+        """
+        import httpx
+        
+        # Use same downloads directory as Gemini service
+        downloads_dir = os.path.join(
+            os.path.dirname(os.path.dirname(__file__)), 
+            "downloads"
+        )
+        os.makedirs(downloads_dir, exist_ok=True)
+        
+        filename = f"{request_id}.mp4"
+        filepath = os.path.join(downloads_dir, filename)
+
+        try:
+            async with httpx.AsyncClient(timeout=120.0, follow_redirects=True) as client:
+                response = await client.get(video_url)
+                response.raise_for_status()
+
+                with open(filepath, 'wb') as f:
+                    f.write(response.content)
+
+                logger.info(f"Downloaded video to {filepath}")
+                return filename
+        except Exception as e:
+            logger.error(f"Failed to download video: {e}")
+            raise ValueError(f"Failed to download video: {e}")

services/gemini_service/__init__.py

@@ -1,55 +0,0 @@
-"""
-Gemini Service - AI-powered image and video generation
-
-Provides:
-- Text generation
-- Image editing  
-- Video generation
-- Job processing and background workers
-"""
-
-# API Client exports
-from services.gemini_service.api_client import (
-    GeminiService,
-    MODELS,
-    DOWNLOADS_DIR,
-    get_gemini_api_key,
-    MOCK_MODE,
-    MOCK_VIDEO_URL,
-)
-
-# Job Processor exports
-from services.gemini_service.job_processor import (
-    GeminiJobProcessor,
-    PriorityWorkerPool,
-    get_pool,
-    get_priority_for_job_type,
-    start_worker,
-    stop_worker,
-)
-
-# API Key Middleware exports # Added
-from services.gemini_service.api_key_config import APIKeyServiceConfig # Added
-from services.gemini_service.api_key_middleware import APIKeyMiddleware # Added
-
-__all__ = [
-    # API Client
-    'GeminiService',
-    'MODELS',
-    'DOWNLOADS_DIR',
-    'get_gemini_api_key',
-    'MOCK_MODE',
-    'MOCK_VIDEO_URL',
-    
-    # Job Processor
-    'GeminiJobProcessor',
-    'PriorityWorkerPool',
-    'get_pool',
-    'get_priority_for_job_type',
-    'start_worker',
-    'stop_worker',
-    
-    # API Key Middleware
-    'APIKeyServiceConfig',
-    'APIKeyMiddleware',
-]

services/gemini_service/api_client.py

@@ -1,401 +0,0 @@
-"""
-Gemini AI Service for image and video generation.
-Python port of the TypeScript geminiService.ts
-Uses server-side API key from environment.
-"""
-import asyncio
-import logging
-import os
-import uuid
-import httpx
-from typing import Optional, Literal
-from google import genai
-from google.genai import types
-
-logger = logging.getLogger(__name__)
-
-# Model names - easily configurable
-MODELS = {
-    "text_generation": "gemini-2.5-flash",
-    "image_edit": "gemini-2.5-flash-image",
-    "video_generation": "veo-3.1-generate-preview"
-}
-
-# Type aliases
-AspectRatio = Literal["16:9", "9:16"]
-Resolution = Literal["720p", "1080p"]
-
-# Video downloads directory
-DOWNLOADS_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "downloads")
-
-# Ensure downloads directory exists
-os.makedirs(DOWNLOADS_DIR, exist_ok=True)
-
-# Mock mode for local testing (set GEMINI_MOCK_MODE=true to skip real API calls)
-MOCK_MODE = os.getenv("GEMINI_MOCK_MODE", "false").lower() == "true"
-MOCK_MODE_SLEEP_TIME = os.getenv("GEMINI_MOCK_MODE_SLEEP_TIME", "0.5")
-
-# Sample video URL for mock mode (a public test video)
-MOCK_VIDEO_URL = "https://video.twimg.com/amplify_video/1994083297756848128/vid/avc1/576x576/ue31qU0xts8L9tXD.mp4?tag=21"
-
-# Concurrency limits from environment (defaults)
-MAX_CONCURRENT_VIDEOS = int(os.getenv("MAX_CONCURRENT_VIDEOS", "2"))
-MAX_CONCURRENT_IMAGES = int(os.getenv("MAX_CONCURRENT_IMAGES", "5"))
-MAX_CONCURRENT_TEXT = int(os.getenv("MAX_CONCURRENT_TEXT", "10"))
-
-# Semaphores for concurrency control
-_video_semaphore: Optional[asyncio.Semaphore] = None
-_image_semaphore: Optional[asyncio.Semaphore] = None
-_text_semaphore: Optional[asyncio.Semaphore] = None
-
-
-def get_video_semaphore() -> asyncio.Semaphore:
-    """Get or create video semaphore."""
-    global _video_semaphore
-    if _video_semaphore is None:
-        _video_semaphore = asyncio.Semaphore(MAX_CONCURRENT_VIDEOS)
-        logger.info(f"Video semaphore initialized with limit: {MAX_CONCURRENT_VIDEOS}")
-    return _video_semaphore
-
-
-def get_image_semaphore() -> asyncio.Semaphore:
-    """Get or create image semaphore."""
-    global _image_semaphore
-    if _image_semaphore is None:
-        _image_semaphore = asyncio.Semaphore(MAX_CONCURRENT_IMAGES)
-        logger.info(f"Image semaphore initialized with limit: {MAX_CONCURRENT_IMAGES}")
-    return _image_semaphore
-
-
-def get_text_semaphore() -> asyncio.Semaphore:
-    """Get or create text semaphore."""
-    global _text_semaphore
-    if _text_semaphore is None:
-        _text_semaphore = asyncio.Semaphore(MAX_CONCURRENT_TEXT)
-        logger.info(f"Text semaphore initialized with limit: {MAX_CONCURRENT_TEXT}")
-    return _text_semaphore
-
-
-def get_gemini_api_key() -> str:
-    """Get Gemini API key from environment."""
-    api_key = os.getenv("GEMINI_API_KEY")
-    if not api_key:
-        raise ValueError("Server Authentication Error with GEMINI")
-    return api_key
-
-
-class GeminiService:
-    """
-    Gemini AI Service for text, image, and video generation.
-    Uses server-side API key from environment.
-    """
-
-    def __init__(self, api_key: Optional[str] = None):
-        """Initialize the Gemini client with API key from env or provided."""
-        self.api_key = api_key or get_gemini_api_key()
-        self.client = genai.Client(api_key=self.api_key)
-
-    def _handle_api_error(self, error: Exception, context: str):
-        """Handle API errors with descriptive messages."""
-        msg = str(error)
-        if "404" in msg or "NOT_FOUND" in msg or "Requested entity was not found" in msg or "[5," in msg:
-            raise ValueError(
-                f"Model not found ({context}). Ensure your API key project has access to this model. "
-                "Veo requires a paid account."
-            )
-        raise error
-
-    async def generate_animation_prompt(
-        self,
-        base64_image: str,
-        mime_type: str,
-        custom_prompt: Optional[str] = None
-    ) -> str:
-        """
-        Analyzes the image to generate a suitable animation prompt.
-        """
-        # Mock mode for testing
-        if MOCK_MODE:
-            logger.info("[MOCK MODE] Generating animation prompt")
-            await asyncio.sleep(GEMINI_MOCK_MODE_SLEEP_TIME)  # Simulate API delay
-            return "A gentle breeze rustles through the scene as soft light dances across the surface. The camera slowly zooms in with a subtle parallax effect, creating depth and movement."
-        
-        default_prompt = custom_prompt or "Describe how this image could be subtly animated with cinematic movement."
-        async with get_text_semaphore():
-            try:
-                response = await asyncio.to_thread(
-                    self.client.models.generate_content,
-                    model=MODELS["text_generation"],
-                    contents=types.Content(
-                        parts=[
-                            types.Part.from_bytes(
-                                data=base64_image,
-                                mime_type=mime_type
-                            ),
-                            types.Part.from_text(text=default_prompt)
-                        ]
-                    )
-                )
-                return response.text or "Cinematic subtle movement"
-            except Exception as error:
-                self._handle_api_error(error, MODELS["text_generation"])
-
-    async def edit_image(
-        self,
-        base64_image: str,
-        mime_type: str,
-        prompt: str
-    ) -> str:
-        """
-        Edit an image using Gemini image model.
-        Returns base64 data URI of the edited image.
-        """
-        # Mock mode for testing - return a sample image
-        if MOCK_MODE:
-            logger.info(f"[MOCK MODE] Editing image with prompt: {prompt}")
-            await asyncio.sleep(1)  # Simulate API delay
-            # Return a small red placeholder image (1x1 pixel)
-            return "data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8z8DwHwAFBQIAX8jx0gAAAABJRU5ErkJggg=="
-        
-        async with get_image_semaphore():
-            try:
-                response = await asyncio.to_thread(
-                    self.client.models.generate_content,
-                    model=MODELS["image_edit"],
-                    contents=types.Content(
-                        parts=[
-                            types.Part.from_bytes(
-                                data=base64_image,
-                                mime_type=mime_type
-                            ),
-                            types.Part.from_text(text=prompt or "Enhance this image")
-                        ]
-                    )
-                )
-
-                candidates = response.candidates
-                if not candidates:
-                    raise ValueError("No candidates returned from Gemini.")
-
-                for part in candidates[0].content.parts:
-                    if hasattr(part, 'inline_data') and part.inline_data and part.inline_data.data:
-                        result_mime = part.inline_data.mime_type or 'image/png'
-                        return f"data:{result_mime};base64,{part.inline_data.data}"
-
-                raise ValueError("No image data found in the response.")
-            except Exception as error:
-                self._handle_api_error(error, MODELS["image_edit"])
-
-    async def start_video_generation(
-        self,
-        base64_image: str,
-        mime_type: str,
-        prompt: str,
-        aspect_ratio: AspectRatio = "16:9",
-        resolution: Resolution = "720p",
-        number_of_videos: int = 1
-    ) -> dict:
-        """
-        Start video generation using Veo model.
-        Returns operation details for polling.
-        """
-        # Mock mode for testing without API credits
-        if MOCK_MODE:
-            import uuid
-            mock_operation_name = f"mock_operation_{uuid.uuid4().hex[:16]}"
-            logger.info(f"[MOCK MODE] Starting video generation: {mock_operation_name}")
-            return {
-                "gemini_operation_name": mock_operation_name,
-                "done": False,
-                "status": "pending"
-            }
-        
-        async with get_video_semaphore():
-            try:
-                # Start video generation
-                operation = await asyncio.to_thread(
-                    self.client.models.generate_videos,
-                    model=MODELS["video_generation"],
-                    prompt=prompt,
-                    image=types.Image(
-                        image_bytes=base64_image,
-                        mime_type=mime_type
-                    ),
-                    config=types.GenerateVideosConfig(
-                        number_of_videos=number_of_videos,
-                        resolution=resolution,
-                        aspect_ratio=aspect_ratio
-                    )
-                )
-                
-                # Return operation details
-                return {
-                    "gemini_operation_name": operation.name,
-                    "done": operation.done,
-                    "status": "completed" if operation.done else "pending"
-                }
-            except Exception as error:
-                self._handle_api_error(error, MODELS["video_generation"])
-
-    async def check_video_status(self, gemini_operation_name: str) -> dict:
-        """
-        Check the status of a video generation operation.
-        Returns status and video URL if complete.
-        """
-        # Mock mode for testing without API credits
-        if MOCK_MODE:
-            # Simulate processing time: complete after 2 checks (track via a simple mechanism)
-            # For simplicity, always return completed with mock video URL
-            logger.info(f"[MOCK MODE] Checking video status: {gemini_operation_name}")
-            await asyncio.sleep(2)  # Simulate API delay
-            return {
-                "gemini_operation_name": gemini_operation_name,
-                "done": True,
-                "status": "completed",
-                "video_url": MOCK_VIDEO_URL
-            }
-        
-        try:
-            # Get operation status using the operation object
-            # First, we need to recreate the operation from the name
-            from google.genai.types import GenerateVideosOperation
-            
-            operation = await asyncio.to_thread(
-                self.client.operations.get,
-                GenerateVideosOperation(name=gemini_operation_name, done=False)
-            )
-            
-            if not operation.done:
-                return {
-                    "gemini_operation_name": gemini_operation_name,
-                    "done": False,
-                    "status": "pending"
-                }
-            
-            # Check for error - handle both string and object types
-            if operation.error:
-                error_msg = operation.error
-                if hasattr(operation.error, 'message'):
-                    error_msg = operation.error.message
-                return {
-                    "gemini_operation_name": gemini_operation_name,
-                    "done": True,
-                    "status": "failed",
-                    "error": str(error_msg) or "Unknown error"
-                }
-            
-            # Extract video URI from result
-            result = operation.result
-            if result and hasattr(result, 'generated_videos') and result.generated_videos:
-                video = result.generated_videos[0]
-                if hasattr(video, 'video') and video.video and hasattr(video.video, 'uri'):
-                    video_uri = video.video.uri
-                    return {
-                        "gemini_operation_name": gemini_operation_name,
-                        "done": True,
-                        "status": "completed",
-                        "video_url": f"{video_uri}&key={self.api_key}"
-                    }
-            
-            return {
-                "gemini_operation_name": gemini_operation_name,
-                "done": True,
-                "status": "failed",
-                "error": "No video URI returned. May be due to safety filters."
-            }
-            
-        except Exception as error:
-            msg = str(error)
-            if "404" in msg or "NOT_FOUND" in msg or "Requested entity was not found" in msg:
-                return {
-                    "gemini_operation_name": gemini_operation_name,
-                    "done": True,
-                    "status": "failed",
-                    "error": "Operation not found (404). It may have expired."
-                }
-            raise error
-
-    async def download_video(self, video_url: str, operation_id: str) -> str:
-        """
-        Download video from Gemini to local storage.
-        Returns the local filename.
-        """
-        filename = f"{operation_id}.mp4"
-        filepath = os.path.join(DOWNLOADS_DIR, filename)
-        
-        try:
-            # follow_redirects=True is required as Gemini returns 302 redirects
-            async with httpx.AsyncClient(timeout=120.0, follow_redirects=True) as client:
-                response = await client.get(video_url)
-                response.raise_for_status()
-                
-                with open(filepath, 'wb') as f:
-                    f.write(response.content)
-                
-                logger.info(f"Downloaded video to {filepath}")
-                return filename
-        except Exception as e:
-            logger.error(f"Failed to download video: {e}")
-            raise ValueError(f"Failed to download video: {e}")
-
-    async def generate_text(
-        self,
-        prompt: str,
-        model: Optional[str] = None
-    ) -> str:
-        """
-        Simple text generation with Gemini.
-        """
-        # Mock mode for testing
-        if MOCK_MODE:
-            logger.info(f"[MOCK MODE] Generating text for prompt: {prompt[:50]}...")
-            await asyncio.sleep(MOCK_MODE_SLEEP_TIME)  # Simulate API delay
-            return f"This is a mock response for your prompt: '{prompt[:100]}...'. In production, this would be generated by Gemini AI."
-        
-        model_name = model or MODELS["text_generation"]
-        async with get_text_semaphore():
-            try:
-                response = await asyncio.to_thread(
-                    self.client.models.generate_content,
-                    model=model_name,
-                    contents=types.Content(
-                        parts=[types.Part.from_text(text=prompt)]
-                    )
-                )
-                return response.text or ""
-            except Exception as error:
-                self._handle_api_error(error, model_name)
-
-    async def analyze_image(
-        self,
-        base64_image: str,
-        mime_type: str,
-        prompt: str
-    ) -> str:
-        """
-        Analyze image with custom prompt.
-        """
-        # Mock mode for testing
-        if MOCK_MODE:
-            logger.info(f"[MOCK MODE] Analyzing image with prompt: {prompt[:50]}...")
-            await asyncio.sleep(MOCK_MODE_SLEEP_TIME)  # Simulate API delay
-            return f"Mock analysis result: The image appears to show a scene that matches your query '{prompt[:50]}...'. This is placeholder content for testing."
-        
-        async with get_text_semaphore():
-            try:
-                response = await asyncio.to_thread(
-                    self.client.models.generate_content,
-                    model=MODELS["text_generation"],
-                    contents=types.Content(
-                        parts=[
-                            types.Part.from_bytes(
-                                data=base64_image,
-                                mime_type=mime_type
-                            ),
-                            types.Part.from_text(text=prompt)
-                        ]
-                    )
-                )
-                return response.text or ""
-            except Exception as error:
-                self._handle_api_error(error, MODELS["text_generation"])

services/gemini_service/api_key_config.py

@@ -1,100 +0,0 @@
-"""
-API Key Service Configuration
-
-Configures automatic API key selection and rotation via middleware.
-"""
-from typing import List, Optional
-import os
-import logging
-
-logger = logging.getLogger(__name__)
-
-
-class APIKeyServiceConfig:
-    """Configuration for API key middleware."""
-    
-    _rotation_strategy: str = "least_used"  # or "round_robin"
-    _cooldown_seconds: int = 60
-    _max_requests_per_minute: int = 60
-    _retry_on_quota_error: bool = True
-    _api_keys: Optional[List[str]] = None
-    
-    @classmethod
-    def register(
-        cls,
-        rotation_strategy: str = "least_used",
-        cooldown_seconds: int = 60,
-        max_requests_per_minute: int = 60,
-        retry_on_quota_error: bool = True
-    ) -> None:
-        """
-        Register API key service configuration.
-        
-        Args:
-            rotation_strategy: "least_used" or "round_robin"
-            cooldown_seconds: Time to wait before reusing a key after quota error
-            max_requests_per_minute: Rate limit per key
-            retry_on_quota_error: Auto-retry with different key on 429
-        
-        Example:
-            APIKeyServiceConfig.register(
-                rotation_strategy="least_used",
-                cooldown_seconds=60,
-                retry_on_quota_error=True
-            )
-        """
-        cls._rotation_strategy = rotation_strategy
-        cls._cooldown_seconds = cooldown_seconds
-        cls._max_requests_per_minute = max_requests_per_minute
-        cls._retry_on_quota_error = retry_on_quota_error
-        
-        # Load API keys from env
-        cls._load_api_keys()
-        
-        logger.info(
-            f"API Key Service configured: "
-            f"keys={len(cls._api_keys or [])}, "
-            f"strategy={rotation_strategy}, "
-            f"retry={retry_on_quota_error}"
-        )
-    
-    @classmethod
-    def _load_api_keys(cls):
-        """Load API keys from environment variables."""
-        keys_str = os.getenv("GEMINI_API_KEYS", "")
-        if not keys_str:
-            # Fallback to single key
-            single_key = os.getenv("GEMINI_API_KEY", "")
-            if single_key:
-                cls._api_keys = [single_key]
-            else:
-                cls._api_keys = []
-                logger.warning("No Gemini API keys configured!")
-        else:
-            cls._api_keys = [k.strip() for k in keys_str.split(",") if k.strip()]
-        
-        if cls._api_keys:
-            logger.info(f"Loaded {len(cls._api_keys)} Gemini API key(s)")
-    
-    @classmethod
-    def get_api_keys(cls) -> List[str]:
-        """Get loaded API keys."""
-        if cls._api_keys is None:
-            cls._load_api_keys()
-        return cls._api_keys or []
-    
-    @classmethod
-    def get_key_count(cls) -> int:
-        """Get number of available keys."""
-        return len(cls.get_api_keys())
-    
-    @classmethod
-    def get_config(cls) -> dict:
-        """Get current configuration."""
-        return {
-            "key_count": cls.get_key_count(),
-            "rotation_strategy": cls._rotation_strategy,
-            "cooldown_seconds": cls._cooldown_seconds,
-            "max_requests_per_minute": cls._max_requests_per_minute,
-            "retry_on_quota_error": cls._retry_on_quota_error
-        }

services/gemini_service/api_key_middleware.py

@@ -1,180 +0,0 @@
-"""
-API Key Middleware - Automatic key selection and rotation
-
-Automatically selects and injects Gemini API keys for requests.
-Handles quota errors with automatic key rotation and retry.
-"""
-import time
-import logging
-from datetime import datetime, timedelta
-from typing import Optional, Dict
-from fastapi import Request, Response
-from starlette.middleware.base import BaseHTTPMiddleware
-from starlette.types import ASGIApp
-
-from core.database import async_session_maker
-from services.gemini_service.api_key_config import APIKeyServiceConfig
-
-logger = logging.getLogger(__name__)
-
-
-# Track key cooldowns in memory
-_key_cooldowns: Dict[int, datetime] = {}
-
-
-class APIKeyMiddleware(BaseHTTPMiddleware):
-    """
-    Middleware for automatic API key management.
-    
-    Features:
-    - Automatic key selection based on strategy
-    - Quota error detection and recovery
-    - Key cooldown management
-    - Usage tracking
-    """
-    
-    def __init__(self, app: ASGIApp):
-        super().__init__(app)
-    
-    async def dispatch(self, request: Request, call_next):
-        """
-        Process request with automatic API key injection.
-        
-        Flow:
-        1. Check if Gemini request
-        2. Select best available key
-        3. Inject into request state
-        4. Handle response (quota errors)
-        """
-        # Only handle Gemini requests
-        if not self._is_gemini_request(request):
-            return await call_next(request)
-        
-        # Select API key
-        try:
-            key_index, api_key = await self._select_api_key()
-            request.state.gemini_api_key = api_key
-            request.state.gemini_key_index = key_index
-        except ValueError as e:
-            # No keys available
-            logger.error(f"No API keys available: {e}")
-            return Response(
-                content=f'{{"detail": "{str(e)}"}}',
-                status_code=503,
-                media_type="application/json"
-            )
-        
-        # Process request
-        response = await call_next(request)
-        
-        # Handle quota errors
-        if response.status_code == 429 and APIKeyServiceConfig._retry_on_quota_error:
-            logger.warning(f"Quota error on key {key_index}, attempting retry")
-            
-            # Mark key in cooldown
-            self._mark_cooldown(key_index)
-            
-            # Try to select different key
-            try:
-                key_index, api_key = await self._select_api_key(exclude_index=key_index)
-                request.state.gemini_api_key = api_key
-                request.state.gemini_key_index = key_index
-                
-                # Retry request
-                logger.info(f"Retrying with key {key_index}")
-                response = await call_next(request)
-            except ValueError:
-                # No other keys available
-                logger.error("All API keys in cooldown or exhausted")
-        
-        # Track usage
-        success = response.status_code < 400
-        await self._track_usage(key_index, success, response.status_code)
-        
-        return response
-    
-    def _is_gemini_request(self, request: Request) -> bool:
-        """Check if request is for Gemini service."""
-        path = request.url.path
-        gemini_paths = ["/gemini/", "/api/gemini"]
-        return any(path.startswith(p) for p in gemini_paths)
-    
-    async def _select_api_key(self, exclude_index: Optional[int] = None) -> tuple[int, str]:
-        """
-        Select best available API key.
-        
-        Args:
-            exclude_index: Key index to exclude (e.g., after quota error)
-        
-        Returns:
-            Tuple of (key_index, api_key)
-        
-        Raises:
-            ValueError: If no keys available
-        """
-        keys = APIKeyServiceConfig.get_api_keys()
-        if not keys:
-            raise ValueError("No API keys configured")
-        
-        # Filter out excluded and cooldown keys
-        available_indices = []
-        for i in range(len(keys)):
-            if i == exclude_index:
-                continue
-            if self._is_in_cooldown(i):
-                continue
-            available_indices.append(i)
-        
-        if not available_indices:
-            raise ValueError("All API keys in cooldown")
-        
-        # Select based on strategy
-        if APIKeyServiceConfig._rotation_strategy == "round_robin":
-            # Simple round-robin
-            selected_index = available_indices[0]
-        else:  # least_used
-            # Get usage stats from DB
-            async with async_session_maker() as db:
-                from services.api_key_manager import get_least_used_key
-                try:
-                    selected_index, _ = await get_least_used_key(db)
-                    if selected_index not in available_indices:
-                        # Fallback to first available
-                        selected_index = available_indices[0]
-                except Exception as e:
-                    logger.error(f"Error getting least used key: {e}")
-                    selected_index = available_indices[0]
-        
-        logger.debug(f"Selected API key index {selected_index}")
-        return selected_index, keys[selected_index]
-    
-    def _is_in_cooldown(self, key_index: int) -> bool:
-        """Check if key is in cooldown period."""
-        if key_index not in _key_cooldowns:
-            return False
-        
-        cooldown_until = _key_cooldowns[key_index]
-        if datetime.utcnow() > cooldown_until:
-            # Cooldown expired
-            del _key_cooldowns[key_index]
-            return False
-        
-        return True
-    
-    def _mark_cooldown(self, key_index: int):
-        """Mark key as in cooldown."""
-        cooldown_seconds = APIKeyServiceConfig._cooldown_seconds
-        cooldown_until = datetime.utcnow() + timedelta(seconds=cooldown_seconds)
-        _key_cooldowns[key_index] = cooldown_until
-        logger.info(f"Key {key_index} in cooldown until {cooldown_until}")
-    
-    async def _track_usage(self, key_index: int, success: bool, status_code: int):
-        """Track API key usage."""
-        try:
-            async with async_session_maker() as db:
-                from services.api_key_manager import record_usage
-                error_message = f"HTTP {status_code}" if not success else None
-                await record_usage(db, key_index, success, error_message)
-                await db.commit()
-        except Exception as e:
-            logger.error(f"Failed to track usage: {e}")

services/gemini_service/job_processor.py

@@ -1,378 +0,0 @@
-"""
-Gemini Job Worker - Specific implementation using the modular PriorityWorkerPool.
-
-This file shows how to use the modular PriorityWorkerPool with Gemini-specific
-job processing logic.
-"""
-import logging
-from datetime import datetime, timedelta
-from typing import Optional
-from sqlalchemy.ext.asyncio import AsyncSession
-
-from core.database import DATABASE_URL
-from core.models import GeminiJob
-from services.priority_worker_pool import (
-    PriorityWorkerPool,
-    JobProcessor,
-    WorkerConfig,
-    get_interval_for_priority
-)
-from services.gemini_service.api_client import GeminiService
-from services.drive_service import DriveService
-import asyncio
-
-logger = logging.getLogger(__name__)
-
-# Job type to priority mapping for Gemini jobs
-JOB_PRIORITY_MAP = {
-    "text": "fast",
-    "analyze": "fast",
-    "animation_prompt": "fast",
-    "image": "medium",
-    "edit_image": "medium",
-    "video": "slow"
-}
-
-
-def get_priority_for_job_type(job_type: str) -> str:
-    """Get the priority tier for a Gemini job type."""
-    return JOB_PRIORITY_MAP.get(job_type, "fast")
-
-
-class GeminiJobProcessor(JobProcessor[GeminiJob]):
-    """Processes Gemini AI jobs (text, image, video generation) with round-robin API keys."""
-    
-    def __init__(self):
-        self.drive_service = DriveService()
-        
-    async def _get_service_with_key(self, session: AsyncSession) -> tuple:
-        """Get a GeminiService with the least-used API key."""
-        from services.api_key_manager import get_least_used_key
-        key_index, api_key = await get_least_used_key(session)
-        return key_index, GeminiService(api_key=api_key)
-    
-    async def _record_usage(self, session: AsyncSession, key_index: int, success: bool, error_message: Optional[str] = None):
-        """Record API key usage after request."""
-        from services.api_key_manager import record_usage
-        await record_usage(session, key_index, success, error_message)
-    
-    def _handle_error(self, job: GeminiJob, error: Exception, reset_to_queued: bool = False) -> tuple[bool, str]:
-        """
-        Handle job errors with retry logic.
-        
-        Args:
-            job: The job object
-            error: The exception raised
-            reset_to_queued: Whether to reset status to 'queued' on retry (for process())
-            
-        Returns:
-            Tuple of (success, error_message)
-            success is False (since it's an error)
-            error_message is the formatted error string
-        """
-        error_str = str(error)
-        is_retryable = False
-        log_msg = ""
-        
-        # Check for Rate Limit (429)
-        if "429" in error_str or "ResourceExhausted" in error_str:
-            is_retryable = True
-            log_msg = f"Rate limit hit for job {job.job_id}"
-        
-        # Check for Auth/Billing errors (401, 403, API key not found, API key not valid, FAILED_PRECONDITION)
-        elif "401" in error_str or "403" in error_str or "Unauthenticated" in error_str or "PermissionDenied" in error_str or "API key not found" in error_str or "API key not valid" in error_str or "FAILED_PRECONDITION" in error_str:
-            is_retryable = True
-            log_msg = f"Auth/Billing error for job {job.job_id}: {error_str}. Rescheduling to try different key."
-        
-        # Check for Server errors (500, 503, 504)
-        elif "500" in error_str or "503" in error_str or "504" in error_str or "INTERNAL" in error_str or "UNAVAILABLE" in error_str or "DEADLINE_EXCEEDED" in error_str:
-            is_retryable = True
-            log_msg = f"Server error for job {job.job_id}: {error_str}"
-            
-        # Try to parse JSON error details if present
-        try:
-            import json
-            import re
-            # Look for JSON-like structure in error string
-            json_match = re.search(r"(\{.*\})", error_str)
-            if json_match:
-                job.api_response = json.loads(json_match.group(1))
-            else:
-                job.api_response = {"error": error_str}
-        except Exception:
-            job.api_response = {"error": error_str}
-
-        if is_retryable:
-            logger.warning(f"{log_msg}. Rescheduling.")
-            job.retry_count += 1
-            config = WorkerConfig.from_env()
-            # Use a longer delay for these errors (e.g., 30s)
-            interval = 30 
-            job.next_process_at = datetime.utcnow() + timedelta(seconds=interval)
-            
-            if reset_to_queued:
-                job.status = "queued"
-                
-            return False, f"Retryable error: {error_str}"
-        else:
-            logger.error(f"Error processing job {job.job_id}: {error}")
-            job.status = "failed"
-            job.error_message = str(error)
-            job.completed_at = datetime.utcnow()
-            return False, str(error)
-
-    async def process(self, job: GeminiJob, session: AsyncSession) -> GeminiJob:
-        """Start processing a new job with round-robin API key."""
-        key_index, service = await self._get_service_with_key(session)
-        input_data = job.input_data or {}
-        success = False
-        error_msg = None
-        
-        try:
-            if job.job_type == "video":
-                job = await self._start_video(job, session, service, input_data)
-                success = True
-            elif job.job_type == "image":
-                job = await self._process_image(job, service, input_data)
-                success = True
-            elif job.job_type == "text":
-                job = await self._process_text(job, service, input_data)
-                success = True
-            elif job.job_type == "analyze":
-                job = await self._process_analyze(job, service, input_data)
-                success = True
-            elif job.job_type == "animation_prompt":
-                job = await self._process_animation_prompt(job, service, input_data)
-                success = True
-            else:
-                job.status = "failed"
-                job.error_message = f"Unknown job type: {job.job_type}"
-                job.completed_at = datetime.utcnow()
-                error_msg = job.error_message
-        except Exception as e:
-            # Use helper for error handling
-            # reset_to_queued=True because if we fail to start, we want to try starting again from scratch
-            success, error_msg = self._handle_error(job, e, reset_to_queued=True)
-        
-        # Record usage
-        await self._record_usage(session, key_index, success, error_msg)
-        
-        return job
-    
-    async def check_status(self, job: GeminiJob, session: AsyncSession) -> GeminiJob:
-        """Check status of an in-progress job (video generation)."""
-        if job.job_type != "video" or not job.third_party_id:
-            job.status = "failed"
-            job.error_message = "Invalid job state for status check"
-            job.completed_at = datetime.utcnow()
-            return job
-        
-        # Use round-robin key for status check
-        key_index, service = await self._get_service_with_key(session)
-        success = False
-        error_msg = None
-        
-        try:
-            status_result = await service.check_video_status(job.third_party_id)
-            # Save raw response
-            job.api_response = status_result
-            
-            if status_result.get("done"):
-                if status_result.get("status") == "completed":
-                    video_url = status_result.get("video_url")
-                    if video_url:
-                        # Store video URL - download will happen on-demand when client requests
-                        job.status = "completed"
-                        job.output_data = {"video_url": video_url}
-                        job.error_message = None  # Clear any previous error
-                        job.completed_at = datetime.utcnow()
-                        success = True
-                        # Sync DB on success
-                        from services.backup_service import get_backup_service
-                        backup_service = get_backup_service()
-                        await backup_service.backup_async()
-                    else:
-                        job.status = "failed"
-                        job.error_message = "No video URL returned"
-                        job.completed_at = datetime.utcnow()
-                        error_msg = job.error_message
-                else:
-                    job.status = "failed"
-                    job.error_message = status_result.get("error", "Unknown error")
-                    job.completed_at = datetime.utcnow()
-                    error_msg = job.error_message
-            else:
-                # Not done - reschedule
-                job.retry_count += 1
-                config = WorkerConfig.from_env()
-                interval = get_interval_for_priority(job.priority, config)
-                job.next_process_at = datetime.utcnow() + timedelta(seconds=interval)
-                success = True  # Status check succeeded even if video not ready
-        
-        except Exception as e:
-            # Use helper for error handling
-            # reset_to_queued=False because we want to continue checking status, not restart
-            success, error_msg = self._handle_error(job, e, reset_to_queued=False)
-        
-        # Record usage
-        await self._record_usage(session, key_index, success, error_msg)
-        
-        return job
-        
-        # Record usage
-        await self._record_usage(session, key_index, success, error_msg)
-        
-        return job
-    
-    async def _start_video(self, job: GeminiJob, session: AsyncSession, service: GeminiService, input_data: dict) -> GeminiJob:
-        """Start async video generation."""
-        prompt = input_data.get("prompt", "")
-        
-        # If prompt is missing, generate one using the animation template
-        if not prompt:
-            try:
-                import os
-                template_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), "prompt", "animation.md")
-                if os.path.exists(template_path):
-                    with open(template_path, "r") as f:
-                        template_prompt = f.read().strip()
-                    
-                    logger.info(f"Generating auto-prompt for job {job.job_id} using template")
-                    prompt = await service.generate_animation_prompt(
-                        base64_image=input_data.get("base64_image", ""),
-                        mime_type=input_data.get("mime_type", "image/jpeg"),
-                        custom_prompt=template_prompt
-                    )
-                    logger.info(f"Generated prompt for job {job.job_id}: {prompt}")
-                    
-                    # Update input data with generated prompt for reference
-                    # Create a new dictionary to ensure SQLAlchemy detects the change
-                    new_input_data = dict(input_data)
-                    new_input_data["prompt"] = prompt
-                    job.input_data = new_input_data
-                    # We need to commit this change to DB so it persists
-                    # But session commit happens outside this method usually? 
-                    # Actually process() calls this, and process() returns job, 
-                    # but doesn't explicitly commit job changes until later?
-                    # The worker loop commits after process() returns.
-                else:
-                    logger.warning(f"Animation prompt template not found at {template_path}")
-            except Exception as e:
-                logger.error(f"Failed to generate auto-prompt: {e}")
-                # Fallback to empty prompt or error? 
-                # Let's let it proceed with empty prompt which might fail at API level or use API default
-        
-        result = await service.start_video_generation(
-            base64_image=input_data.get("base64_image", ""),
-            mime_type=input_data.get("mime_type", "image/jpeg"),
-            prompt=prompt,
-            aspect_ratio=input_data.get("aspect_ratio", "16:9"),
-            resolution=input_data.get("resolution", "720p"),
-            number_of_videos=input_data.get("number_of_videos", 1)
-        )
-        job.third_party_id = result.get("gemini_operation_name")
-        job.api_response = result
-        
-        # Schedule first status check
-        config = WorkerConfig.from_env()
-        interval = get_interval_for_priority(job.priority, config)
-        job.next_process_at = datetime.utcnow() + timedelta(seconds=interval)
-        
-        return job
-    
-    async def _process_image(self, job: GeminiJob, service: GeminiService, input_data: dict) -> GeminiJob:
-        """Process image edit (synchronous)."""
-        result = await service.edit_image(
-            base64_image=input_data.get("base64_image", ""),
-            mime_type=input_data.get("mime_type", "image/jpeg"),
-            prompt=input_data.get("prompt", "")
-        )
-        job.status = "completed"
-        job.output_data = {"image": result}
-        # Don't save full base64 image to api_response
-        job.api_response = {"status": "success", "type": "image_edit"}
-        job.completed_at = datetime.utcnow()
-        # Sync DB on success
-        from services.backup_service import get_backup_service
-        backup_service = get_backup_service()
-        await backup_service.backup_async()
-        return job
-    
-    async def _process_text(self, job: GeminiJob, service: GeminiService, input_data: dict) -> GeminiJob:
-        """Process text generation (synchronous)."""
-        result = await service.generate_text(
-            prompt=input_data.get("prompt", ""),
-            model=input_data.get("model")
-        )
-        job.status = "completed"
-        job.output_data = {"text": result}
-        job.api_response = {"result": result}
-        job.completed_at = datetime.utcnow()
-        # Sync DB on success
-        from services.backup_service import get_backup_service
-        backup_service = get_backup_service()
-        await backup_service.backup_async()
-        return job
-    
-    async def _process_analyze(self, job: GeminiJob, service: GeminiService, input_data: dict) -> GeminiJob:
-        """Process image analysis (synchronous)."""
-        result = await service.analyze_image(
-            base64_image=input_data.get("base64_image", ""),
-            mime_type=input_data.get("mime_type", "image/jpeg"),
-            prompt=input_data.get("prompt", "")
-        )
-        job.status = "completed"
-        job.output_data = {"analysis": result}
-        job.api_response = {"result": result}
-        job.completed_at = datetime.utcnow()
-        # Sync DB on success
-        from services.backup_service import get_backup_service
-        backup_service = get_backup_service()
-        await backup_service.backup_async()
-        return job
-    
-    async def _process_animation_prompt(self, job: GeminiJob, service: GeminiService, input_data: dict) -> GeminiJob:
-        """Process animation prompt generation (synchronous)."""
-        result = await service.generate_animation_prompt(
-            base64_image=input_data.get("base64_image", ""),
-            mime_type=input_data.get("mime_type", "image/jpeg"),
-            custom_prompt=input_data.get("custom_prompt")
-        )
-        job.status = "completed"
-        job.output_data = {"prompt": result}
-        job.api_response = {"result": result}
-        job.completed_at = datetime.utcnow()
-        # Sync DB on success
-        from services.backup_service import get_backup_service
-        backup_service = get_backup_service()
-        await backup_service.backup_async()
-        return job
-
-
-# Singleton pool instance
-_pool: Optional[PriorityWorkerPool] = None
-
-
-def get_pool() -> PriorityWorkerPool:
-    """Get the global Gemini worker pool instance."""
-    global _pool
-    if _pool is None:
-        _pool = PriorityWorkerPool(
-            database_url=DATABASE_URL,
-            job_model=GeminiJob,
-            job_processor=GeminiJobProcessor(),
-            config=WorkerConfig.from_env()
-        )
-    return _pool
-
-
-async def start_worker():
-    """Start the Gemini job worker pool."""
-    pool = get_pool()
-    await pool.start()
-
-
-async def stop_worker():
-    """Stop the Gemini job worker pool."""
-    pool = get_pool()
-    await pool.stop()

services/priority_worker_pool.py

@@ -1,547 +0,0 @@
-"""
-Modular Priority-Tier Worker Pool
-
-A self-contained, plug-and-play worker pool for processing async jobs
-with priority-tier scheduling. Can be used in any Python application.
-
-Usage:
-    from services.priority_worker_pool import PriorityWorkerPool, WorkerConfig
-    
-    # Define your job processor function
-    async def process_my_job(job, session):
-        # Process job and return updated job
-        job.status = "completed"
-        job.output_data = {"result": "done"}
-        return job
-    
-    # Configure and start pool
-    pool = PriorityWorkerPool(
-        database_url="sqlite+aiosqlite:///./my_db.db",
-        job_model=MyJobModel,
-        job_processor=process_my_job,
-        config=WorkerConfig(fast_workers=5, medium_workers=5, slow_workers=5)
-    )
-    await pool.start()
-
-Environment Variables (optional):
-    FAST_WORKERS: Number of fast workers (default: 5)
-    MEDIUM_WORKERS: Number of medium workers (default: 5)
-    SLOW_WORKERS: Number of slow workers (default: 5)
-    FAST_INTERVAL: Fast tier polling interval in seconds (default: 5)
-    MEDIUM_INTERVAL: Medium tier polling interval in seconds (default: 30)
-    SLOW_INTERVAL: Slow tier polling interval in seconds (default: 60)
-
-Dependencies:
-    sqlalchemy[asyncio]>=2.0.0
-    aiosqlite (for SQLite) or asyncpg (for PostgreSQL)
-
-Job Model Requirements:
-    Your job model must have these columns:
-    - job_id: str (unique identifier)
-    - status: str (queued, processing, completed, failed, cancelled)
-    - priority: str (fast, medium, slow)
-    - next_process_at: datetime (nullable, for rescheduling)
-    - retry_count: int (default 0)
-    - created_at: datetime
-    - started_at: datetime (nullable)
-    - completed_at: datetime (nullable)
-    - error_message: str (nullable)
-"""
-import asyncio
-import logging
-import os
-from abc import ABC, abstractmethod
-from dataclasses import dataclass, field
-from datetime import datetime, timedelta
-from typing import Optional, List, Callable, Any, TypeVar, Generic
-from sqlalchemy import select, or_, and_
-from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
-
-logger = logging.getLogger(__name__)
-
-# Generic type for job model
-JobType = TypeVar('JobType')
-
-
-@dataclass
-class WorkerConfig:
-    """Configuration for the worker pool."""
-    fast_workers: int = 5
-    medium_workers: int = 5
-    slow_workers: int = 5
-    fast_interval: int = 2      # seconds
-    medium_interval: int = 10   # seconds
-    slow_interval: int = 15     # seconds
-    max_retries: int = 60       # Max retry attempts before failing
-    job_per_api_key: int = 1    # Max concurrent jobs per API key
-    
-    @classmethod
-    def from_env(cls) -> 'WorkerConfig':
-        """Create config from environment variables."""
-        return cls(
-            fast_workers=int(os.getenv("FAST_WORKERS", "5")),
-            medium_workers=int(os.getenv("MEDIUM_WORKERS", "5")),
-            slow_workers=int(os.getenv("SLOW_WORKERS", "5")),
-            fast_interval=int(os.getenv("FAST_INTERVAL", "5")),
-            medium_interval=int(os.getenv("MEDIUM_INTERVAL", "30")),
-            slow_interval=int(os.getenv("SLOW_INTERVAL", "60")),
-            job_per_api_key=int(os.getenv("JOB_PER_API_KEY", "1")),
-        )
-
-
-@dataclass
-class PriorityMapping:
-    """Maps job types to priority tiers."""
-    mappings: dict = field(default_factory=dict)
-    
-    def get_priority(self, job_type: str, default: str = "fast") -> str:
-        """Get priority for a job type."""
-        return self.mappings.get(job_type, default)
-    
-    def get_interval(self, priority: str, config: WorkerConfig) -> int:
-        """Get polling interval for a priority tier."""
-        if priority == "fast":
-            return config.fast_interval
-        elif priority == "medium":
-            return config.medium_interval
-        else:
-            return config.slow_interval
-
-
-class JobProcessor(ABC, Generic[JobType]):
-    """Abstract base class for job processors."""
-    
-    @abstractmethod
-    async def process(self, job: JobType, session: AsyncSession) -> JobType:
-        """
-        Process a job and return the updated job.
-        
-        Args:
-            job: The job to process
-            session: Database session for updates
-            
-        Returns:
-            The updated job with new status/output
-        """
-        pass
-    
-    @abstractmethod
-    async def check_status(self, job: JobType, session: AsyncSession) -> JobType:
-        """
-        Check status of an in-progress job (for async third-party operations).
-        
-        Args:
-            job: The job to check
-            session: Database session for updates
-            
-        Returns:
-            The updated job. Set next_process_at to reschedule if not done.
-        """
-        pass
-
-
-class PriorityWorker(Generic[JobType]):
-    """Worker that processes jobs of a specific priority tier."""
-    
-    def __init__(
-        self,
-        worker_id: int,
-        priority: str,
-        poll_interval: int,
-        session_maker: async_sessionmaker,
-        job_model: type,
-        job_processor: JobProcessor[JobType],
-        max_retries: int = 60,
-        wake_event: Optional[asyncio.Event] = None,
-        config: Optional[WorkerConfig] = None
-    ):
-        self.worker_id = worker_id
-        self.priority = priority
-        self.poll_interval = poll_interval
-        self.session_maker = session_maker
-        self.job_model = job_model
-        self.job_processor = job_processor
-        self.max_retries = max_retries
-        self._running = False
-        self._current_job_id: Optional[str] = None
-        self._wake_event = wake_event  # Event to wake worker immediately when new jobs arrive
-        self._config = config or WorkerConfig.from_env()
-    
-    async def start(self):
-        """Start the worker polling loop."""
-        self._running = True
-        logger.debug(f"Worker {self.worker_id} ({self.priority}) started, polling every {self.poll_interval}s")
-        asyncio.create_task(self._poll_loop())
-    
-    async def stop(self):
-        """Stop the worker."""
-        self._running = False
-        logger.info(f"Worker {self.worker_id} ({self.priority}) stopped")
-    
-    async def _poll_loop(self):
-        """Main polling loop with optimized scheduling.
-        
-        Optimizations:
-        - When no jobs are found, sleep for poll_interval before checking again
-        - When a job is processed, immediately check for the next job (no waiting)
-        - This ensures first job starts immediately when queue was empty
-        - This ensures next job starts immediately after current job finishes
-        """
-        while self._running:
-            job_found = False
-            try:
-                job_found = await self._process_one_job()
-            except Exception as e:
-                logger.error(f"Worker {self.worker_id}: Error in poll loop: {e}")
-            
-            # Only sleep if no job was found - otherwise immediately look for next job
-            if not job_found:
-                # Wait on event with timeout - allows immediate wake-up when new job arrives
-                if self._wake_event:
-                    try:
-                        # Wait for event or timeout (whichever comes first)
-                        await asyncio.wait_for(
-                            self._wake_event.wait(),
-                            timeout=self.poll_interval
-                        )
-                        # Clear event after waking (we'll check for jobs)
-                        self._wake_event.clear()
-                    except asyncio.TimeoutError:
-                        pass  # Normal timeout, check for jobs
-                else:
-                    await asyncio.sleep(self.poll_interval)
-    
-    async def _process_one_job(self) -> bool:
-        """Find and process one job.
-        
-        Enforces constraints:
-        1. Only one job per user can be in processing state at a time
-        2. Total processing jobs limited to JOB_PER_API_KEY * number of API keys
-        
-        Returns:
-            True if a job was found and processed, False if no jobs available
-        """
-        async with self.session_maker() as session:
-            from sqlalchemy import func
-            
-            now = datetime.utcnow()
-            
-            # Get number of API keys for capacity calculation
-            try:
-                from services.api_key_manager import get_key_count
-                num_api_keys = get_key_count()
-                max_processing = self._config.job_per_api_key * num_api_keys
-            except ImportError:
-                max_processing = 10  # Default fallback
-            
-            # Check if we're at max processing capacity (only for new jobs being picked up)
-            count_query = select(func.count()).where(
-                self.job_model.status == "processing"
-            )
-            count_result = await session.execute(count_query)
-            current_processing = count_result.scalar() or 0
-            
-            # Query for jobs matching this priority tier
-            query = select(self.job_model).where(
-                and_(
-                    self.job_model.priority == self.priority,
-                    self.job_model.status.in_(["queued", "processing"]),
-                    or_(
-                        self.job_model.next_process_at.is_(None),
-                        self.job_model.next_process_at <= now
-                    )
-                )
-            ).order_by(self.job_model.created_at).limit(1)
-            
-            result = await session.execute(query)
-            job = result.scalar_one_or_none()
-            
-            if not job:
-                return False
-            
-            # For queued jobs, apply the constraints
-            if job.status == "queued":
-                # Constraint 1: Check if this user already has a job in processing
-                user_processing_query = select(func.count()).where(
-                    and_(
-                        self.job_model.user_id == job.user_id,
-                        self.job_model.status == "processing"
-                    )
-                )
-                user_result = await session.execute(user_processing_query)
-                user_processing_count = user_result.scalar() or 0
-                
-                if user_processing_count > 0:
-                    logger.debug(f"Worker {self.worker_id}: User {job.user_id} already has a job processing, skipping")
-                    return False
-                
-                # Constraint 2: Check if we're at max total processing capacity
-                if current_processing >= max_processing:
-                    logger.debug(f"Worker {self.worker_id}: At max capacity ({current_processing}/{max_processing}), skipping new job")
-                    return False
-            
-            self._current_job_id = job.job_id
-            
-            try:
-                await self._process_job(session, job)
-                return True
-            except Exception as e:
-                logger.error(f"Worker {self.worker_id}: Error processing job {job.job_id}: {e}")
-                job.status = "failed"
-                job.error_message = str(e)
-                job.completed_at = datetime.utcnow()
-                await session.commit()
-                return True  # Job was found, even though it failed
-            finally:
-                self._current_job_id = None
-    
-    async def _process_job(self, session: AsyncSession, job: JobType):
-        """Process a single job."""
-        logger.info(f"Worker {self.worker_id}: Processing job {job.job_id} (status: {job.status})")
-        
-        from sqlalchemy import update
-        
-        if job.status == "queued":
-            # New job - try to claim it atomically
-            # Set next_process_at to future to prevent others from picking it up while we process
-            next_check = datetime.utcnow() + timedelta(seconds=self.poll_interval * 2)
-            
-            stmt = (
-                update(self.job_model)
-                .where(
-                    self.job_model.job_id == job.job_id,
-                    self.job_model.status == "queued"
-                )
-                .values(
-                    status="processing",
-                    started_at=datetime.utcnow(),
-                    next_process_at=next_check
-                )
-            )
-            result = await session.execute(stmt)
-            await session.commit()
-            
-            if result.rowcount == 0:
-                logger.info(f"Worker {self.worker_id}: Failed to claim job {job.job_id} (already taken)")
-                return
-
-            # We claimed it. Refresh and process.
-            await session.refresh(job)
-            job = await self.job_processor.process(job, session)
-            
-        else:
-            # Already processing - try to claim for status check
-            # Ensure we only pick it up if next_process_at matches (or is null/past)
-            # But the SELECT already filtered for that.
-            # We just need to ensure no one else grabbed it between SELECT and UPDATE.
-            
-            # Update next_process_at to future to lock it for this check
-            next_check = datetime.utcnow() + timedelta(seconds=self.poll_interval * 2)
-            
-            stmt = (
-                update(self.job_model)
-                .where(
-                    self.job_model.job_id == job.job_id,
-                    or_(
-                        self.job_model.next_process_at.is_(None),
-                        self.job_model.next_process_at <= datetime.utcnow()
-                    )
-                )
-                .values(next_process_at=next_check)
-            )
-            result = await session.execute(stmt)
-            await session.commit()
-            
-            if result.rowcount == 0:
-                logger.info(f"Worker {self.worker_id}: Failed to claim job {job.job_id} for check (already taken)")
-                return
-                
-            await session.refresh(job)
-            job = await self.job_processor.check_status(job, session)
-        
-        # Handle retry limit
-        if job.status == "processing" and job.retry_count > self.max_retries:
-            job.status = "failed"
-            job.error_message = f"Max retries ({self.max_retries}) exceeded"
-            job.completed_at = datetime.utcnow()
-        
-        # Handle credit finalization for jobs with reserved credits
-        if job.status in ("completed", "failed", "cancelled"):
-            await self._handle_job_credits(session, job)
-        
-        await session.commit()
-    
-    async def _handle_job_credits(self, session: AsyncSession, job: JobType):
-        """Handle credit finalization when job reaches terminal state."""
-        # Check if job has credits_reserved attribute (credit-enabled jobs)
-        if not hasattr(job, 'credits_reserved') or job.credits_reserved <= 0:
-            return
-        
-        try:
-            from services.credit_service.credit_manager import handle_job_completion
-            await handle_job_completion(session, job)
-        except ImportError:
-            # Credit service not available - skip
-            logger.debug(f"Credit service not available for job {job.job_id}")
-        except Exception as e:
-            logger.error(f"Error handling credits for job {job.job_id}: {e}")
-
-
-class PriorityWorkerPool(Generic[JobType]):
-    """
-    Modular priority-tier worker pool.
-    
-    Can be used with any job model that follows the required schema.
-    """
-    
-    def __init__(
-        self,
-        database_url: str,
-        job_model: type,
-        job_processor: JobProcessor[JobType],
-        config: Optional[WorkerConfig] = None
-    ):
-        """
-        Initialize the worker pool.
-        
-        Args:
-            database_url: SQLAlchemy async database URL
-            job_model: Your ORM model class for jobs
-            job_processor: Instance of JobProcessor to handle jobs
-            config: Worker configuration (uses env vars if not provided)
-        """
-        self.database_url = database_url
-        self.job_model = job_model
-        self.job_processor = job_processor
-        self.config = config or WorkerConfig.from_env()
-        
-        self.engine = create_async_engine(database_url, echo=False)
-        self.session_maker = async_sessionmaker(
-            self.engine,
-            class_=AsyncSession,
-            expire_on_commit=False
-        )
-        self.workers: List[PriorityWorker] = []
-        self._running = False
-        
-        # Wake events for each priority tier - allows immediate job notification
-        self._wake_events: dict[str, asyncio.Event] = {
-            "fast": asyncio.Event(),
-            "medium": asyncio.Event(),
-            "slow": asyncio.Event()
-        }
-    
-    async def start(self):
-        """Start all workers."""
-        self._running = True
-        worker_id = 0
-        
-        # Create fast workers
-        for i in range(self.config.fast_workers):
-            worker = PriorityWorker(
-                worker_id=worker_id,
-                priority="fast",
-                poll_interval=self.config.fast_interval,
-                session_maker=self.session_maker,
-                job_model=self.job_model,
-                job_processor=self.job_processor,
-                max_retries=self.config.max_retries,
-                wake_event=self._wake_events["fast"],
-                config=self.config
-            )
-            self.workers.append(worker)
-            await worker.start()
-            worker_id += 1
-        
-        # Create medium workers
-        for i in range(self.config.medium_workers):
-            worker = PriorityWorker(
-                worker_id=worker_id,
-                priority="medium",
-                poll_interval=self.config.medium_interval,
-                session_maker=self.session_maker,
-                job_model=self.job_model,
-                job_processor=self.job_processor,
-                max_retries=self.config.max_retries,
-                wake_event=self._wake_events["medium"],
-                config=self.config
-            )
-            self.workers.append(worker)
-            await worker.start()
-            worker_id += 1
-        
-        # Create slow workers
-        for i in range(self.config.slow_workers):
-            worker = PriorityWorker(
-                worker_id=worker_id,
-                priority="slow",
-                poll_interval=self.config.slow_interval,
-                session_maker=self.session_maker,
-                job_model=self.job_model,
-                job_processor=self.job_processor,
-                max_retries=self.config.max_retries,
-                wake_event=self._wake_events["slow"],
-                config=self.config
-            )
-            self.workers.append(worker)
-            await worker.start()
-            worker_id += 1
-        
-        total = self.config.fast_workers + self.config.medium_workers + self.config.slow_workers
-        logger.info(
-            f"PriorityWorkerPool started with {total} workers: "
-            f"{self.config.fast_workers} fast, {self.config.medium_workers} medium, {self.config.slow_workers} slow"
-        )
-    
-    def notify_new_job(self, priority: str):
-        """
-        Wake sleeping workers of the specified priority tier.
-        Call this when a new job is created to start processing immediately.
-        
-        Args:
-            priority: Priority tier ("fast", "medium", or "slow")
-        """
-        if priority in self._wake_events:
-            self._wake_events[priority].set()
-            logger.debug(f"Notified {priority} workers of new job")
-    
-    async def stop(self):
-        """Stop all workers and refund orphaned jobs."""
-        self._running = False
-        
-        # Refund credits for any jobs that were processing when server stopped
-        await self._refund_orphaned_jobs()
-        
-        for worker in self.workers:
-            await worker.stop()
-        logger.info("PriorityWorkerPool stopped")
-    
-    async def _refund_orphaned_jobs(self):
-        """Refund credits for jobs abandoned during shutdown."""
-        try:
-            from services.credit_service.credit_manager import refund_orphaned_jobs
-            async with self.session_maker() as session:
-                refund_count = await refund_orphaned_jobs(session)
-                if refund_count > 0:
-                    logger.info(f"Shutdown: Refunded {refund_count} orphaned job(s)")
-        except ImportError:
-            logger.debug("Credit service not available for orphaned job refunds")
-        except Exception as e:
-            logger.error(f"Error refunding orphaned jobs during shutdown: {e}")
-
-
-# Convenience functions for priority mapping
-def get_priority_for_job_type(job_type: str, mappings: dict) -> str:
-    """Get priority tier for a job type using provided mappings."""
-    return mappings.get(job_type, "fast")
-
-
-def get_interval_for_priority(priority: str, config: Optional[WorkerConfig] = None) -> int:
-    """Get polling interval for a priority tier."""
-    cfg = config or WorkerConfig.from_env()
-    if priority == "fast":
-        return cfg.fast_interval
-    elif priority == "medium":
-        return cfg.medium_interval
-    else:
-        return cfg.slow_interval

tests/debug_gemini_service.py

@@ -1,165 +0,0 @@
-"""
-Debug script to test Gemini service with API keys from environment.
-Keys should be in GEMINI_KEYS environment variable, comma-separated.
-
-Usage:
-    GEMINI_KEYS="key1,key2,key3" python tests/debug_gemini_service.py
-"""
-import os
-import sys
-import asyncio
-import logging
-import base64
-from dotenv import load_dotenv
-
-# Load environment variables
-load_dotenv()
-
-# Add parent directory to path
-sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
-
-from services.gemini_service import GeminiService, MODELS
-
-# Configure logging
-logging.basicConfig(level=logging.INFO)
-logger = logging.getLogger(__name__)
-
-# Test image path
-TEST_IMAGE_PATH = os.path.join(os.path.dirname(os.path.dirname(__file__)), "test.jpg")
-
-
-def load_test_image():
-    """Load test image and return base64 + mime type."""
-    if not os.path.exists(TEST_IMAGE_PATH):
-        logger.error(f"Test image not found: {TEST_IMAGE_PATH}")
-        return None, None
-    
-    with open(TEST_IMAGE_PATH, "rb") as f:
-        image_data = f.read()
-    
-    base64_image = base64.b64encode(image_data).decode("utf-8")
-    mime_type = "image/jpeg"
-    
-    logger.info(f"Loaded test image: {TEST_IMAGE_PATH} ({len(image_data)} bytes)")
-    return base64_image, mime_type
-
-
-async def test_generate_text(service: GeminiService, key_index: int):
-    """Test simple text generation."""
-    logger.info(f"[Key {key_index}] Testing text generation...")
-    try:
-        result = await service.generate_text("Say hello in one word.")
-        logger.info(f"[Key {key_index}] Text generation result: {result[:100]}...")
-        return True
-    except Exception as e:
-        logger.error(f"[Key {key_index}] Text generation failed: {e}")
-        return False
-
-
-async def test_analyze_image(service: GeminiService, key_index: int, base64_image: str, mime_type: str):
-    """Test image analysis."""
-    logger.info(f"[Key {key_index}] Testing image analysis...")
-    try:
-        result = await service.analyze_image(
-            base64_image=base64_image,
-            mime_type=mime_type,
-            prompt="Describe this image in one sentence."
-        )
-        logger.info(f"[Key {key_index}] Image analysis result: {result[:100]}...")
-        return True
-    except Exception as e:
-        logger.error(f"[Key {key_index}] Image analysis failed: {e}")
-        return False
-
-
-async def test_generate_animation_prompt(service: GeminiService, key_index: int, base64_image: str, mime_type: str):
-    """Test animation prompt generation."""
-    logger.info(f"[Key {key_index}] Testing animation prompt generation...")
-    try:
-        result = await service.generate_animation_prompt(
-            base64_image=base64_image,
-            mime_type=mime_type
-        )
-        logger.info(f"[Key {key_index}] Animation prompt result: {result[:100]}...")
-        return True
-    except Exception as e:
-        logger.error(f"[Key {key_index}] Animation prompt generation failed: {e}")
-        return False
-
-
-async def test_key(api_key: str, key_index: int, base64_image: str, mime_type: str):
-    """Test all basic operations with a single API key."""
-    logger.info(f"\n{'='*50}")
-    logger.info(f"Testing Key {key_index}: {api_key[:10]}...{api_key[-4:]}")
-    logger.info(f"{'='*50}")
-    
-    try:
-        service = GeminiService(api_key)
-    except Exception as e:
-        logger.error(f"[Key {key_index}] Failed to initialize service: {e}")
-        return {"key_index": key_index, "valid": False, "error": str(e)}
-    
-    results = {
-        "key_index": key_index,
-        "key_preview": f"{api_key[:10]}...{api_key[-4:]}",
-        "text_generation": await test_generate_text(service, key_index),
-        "image_analysis": await test_analyze_image(service, key_index, base64_image, mime_type),
-        "animation_prompt": await test_generate_animation_prompt(service, key_index, base64_image, mime_type),
-    }
-    
-    results["valid"] = all([
-        results["text_generation"],
-        results["image_analysis"],
-        results["animation_prompt"]
-    ])
-    
-    return results
-
-
-async def main():
-    # Load test image
-    base64_image, mime_type = load_test_image()
-    if not base64_image:
-        logger.error("Cannot run tests without test image. Please add test.jpg to project root.")
-        return
-    
-    gemini_keys_str = os.getenv("GEMINI_KEYS", "")
-    
-    if not gemini_keys_str:
-        logger.error("GEMINI_KEYS environment variable not set.")
-        logger.info("Usage: GEMINI_KEYS='key1,key2,key3' python tests/debug_gemini_service.py")
-        return
-    
-    keys = [k.strip() for k in gemini_keys_str.split(",") if k.strip()]
-    
-    if not keys:
-        logger.error("No valid keys found in GEMINI_KEYS.")
-        return
-    
-    logger.info(f"Found {len(keys)} API key(s) to test.")
-    logger.info(f"Available models: {MODELS}")
-    
-    all_results = []
-    for i, key in enumerate(keys):
-        result = await test_key(key, i + 1, base64_image, mime_type)
-        all_results.append(result)
-    
-    # Summary
-    logger.info(f"\n{'='*50}")
-    logger.info("SUMMARY")
-    logger.info(f"{'='*50}")
-    
-    valid_count = sum(1 for r in all_results if r.get("valid", False))
-    logger.info(f"Valid keys: {valid_count}/{len(keys)}")
-    
-    for result in all_results:
-        status = "✓ VALID" if result.get("valid") else "✗ INVALID"
-        logger.info(f"  Key {result['key_index']}: {status}")
-        if not result.get("valid"):
-            for test_name in ["text_generation", "image_analysis", "animation_prompt"]:
-                if test_name in result and not result[test_name]:
-                    logger.info(f"    - {test_name}: FAILED")
-
-
-if __name__ == "__main__":
-    asyncio.run(main())

tests/test_fal_service.py

@@ -0,0 +1,290 @@
+"""
+Tests for Fal.ai Service.
+
+Tests cover:
+1. Initialization & API key handling
+2. Video generation
+3. Error handling
+4. Mock mode
+"""
+import pytest
+import asyncio
+import os
+from unittest.mock import patch, MagicMock, AsyncMock
+
+
+# =============================================================================
+# 1. Initialization & Configuration Tests
+# =============================================================================
+
+class TestFalServiceInit:
+    """Test FalService initialization and configuration."""
+
+    def test_init_with_explicit_api_key(self):
+        """Service initializes with explicit API key."""
+        with patch.dict(os.environ, {"FAL_KEY": "env-key"}):
+            from services.fal_service import FalService
+
+            service = FalService(api_key="test-key-123")
+
+            assert service.api_key == "test-key-123"
+
+    def test_init_with_env_fallback(self):
+        """Service falls back to environment variable for API key."""
+        with patch.dict(os.environ, {"FAL_KEY": "env-key-456"}):
+            from services.fal_service import FalService
+
+            service = FalService()
+
+            assert service.api_key == "env-key-456"
+
+    def test_init_fails_without_api_key(self):
+        """Service raises error when no API key available."""
+        with patch.dict(os.environ, {}, clear=True):
+            os.environ.pop("FAL_KEY", None)
+
+            from services.fal_service import get_fal_api_key
+
+            with pytest.raises(ValueError, match="FAL_KEY not configured"):
+                get_fal_api_key()
+
+    def test_models_dict_has_required_entries(self):
+        """MODELS dictionary has all required model names."""
+        from services.fal_service import MODELS
+
+        assert "video_generation" in MODELS
+        assert "veo3" in MODELS["video_generation"].lower() or "image-to-video" in MODELS["video_generation"]
+
+
+# =============================================================================
+# 2. Video Generation Tests
+# =============================================================================
+
+class TestFalVideoGeneration:
+    """Test video generation methods."""
+
+    @pytest.mark.asyncio
+    async def test_start_video_generation_mock_mode(self):
+        """Video generation works in mock mode."""
+        with patch.dict(os.environ, {"FAL_KEY": "test-key"}):
+            with patch('services.fal_service.api_client.MOCK_MODE', True):
+                from services.fal_service import FalService
+
+                service = FalService(api_key="test-key")
+                result = await service.start_video_generation(
+                    base64_image="base64data",
+                    mime_type="image/jpeg",
+                    prompt="Animate this"
+                )
+
+                assert result["done"] is True
+                assert result["status"] == "completed"
+                assert "video_url" in result
+                assert "fal_request_id" in result
+
+    @pytest.mark.asyncio
+    async def test_start_video_generation_success(self):
+        """Video generation returns video URL on success."""
+        with patch.dict(os.environ, {"FAL_KEY": "test-key"}):
+            with patch('services.fal_service.api_client.MOCK_MODE', False):
+                with patch('services.fal_service.api_client.asyncio.to_thread') as mock_to_thread:
+                    from services.fal_service import FalService
+
+                    # Mock fal_client response
+                    mock_result = {
+                        "video": {"url": "https://fal.ai/video.mp4"},
+                        "request_id": "req-123"
+                    }
+                    mock_to_thread.return_value = mock_result
+
+                    service = FalService(api_key="test-key")
+                    result = await service.start_video_generation(
+                        base64_image="base64data",
+                        mime_type="image/jpeg",
+                        prompt="Animate this"
+                    )
+
+                    assert result["done"] is True
+                    assert result["status"] == "completed"
+                    assert result["video_url"] == "https://fal.ai/video.mp4"
+
+    @pytest.mark.asyncio
+    async def test_start_video_generation_no_video_url(self):
+        """Video generation returns failed when no URL in response."""
+        with patch.dict(os.environ, {"FAL_KEY": "test-key"}):
+            with patch('services.fal_service.api_client.MOCK_MODE', False):
+                with patch('services.fal_service.api_client.asyncio.to_thread') as mock_to_thread:
+                    from services.fal_service import FalService
+
+                    # Mock response without video URL
+                    mock_result = {"status": "error"}
+                    mock_to_thread.return_value = mock_result
+
+                    service = FalService(api_key="test-key")
+                    result = await service.start_video_generation(
+                        base64_image="base64data",
+                        mime_type="image/jpeg",
+                        prompt="Animate this"
+                    )
+
+                    assert result["done"] is True
+                    assert result["status"] == "failed"
+                    assert "error" in result
+
+    @pytest.mark.asyncio
+    async def test_start_video_generation_with_params(self):
+        """Video generation passes aspect_ratio and resolution."""
+        with patch.dict(os.environ, {"FAL_KEY": "test-key"}):
+            with patch('services.fal_service.api_client.MOCK_MODE', False):
+                with patch('services.fal_service.api_client.asyncio.to_thread') as mock_to_thread:
+                    from services.fal_service import FalService
+
+                    mock_result = {"video": {"url": "https://fal.ai/video.mp4"}}
+                    mock_to_thread.return_value = mock_result
+
+                    service = FalService(api_key="test-key")
+                    await service.start_video_generation(
+                        base64_image="base64data",
+                        mime_type="image/jpeg",
+                        prompt="Animate",
+                        aspect_ratio="9:16",
+                        resolution="720p"
+                    )
+
+                    # Verify arguments were passed
+                    call_args = mock_to_thread.call_args
+                    arguments = call_args.kwargs.get("arguments") or call_args[1].get("arguments")
+                    assert arguments["aspect_ratio"] == "9:16"
+                    assert arguments["resolution"] == "720p"
+
+
+# =============================================================================
+# 3. Error Handling Tests
+# =============================================================================
+
+class TestFalErrorHandling:
+    """Test error handling methods."""
+
+    def test_handle_api_error_401(self):
+        """_handle_api_error raises ValueError for 401."""
+        with patch.dict(os.environ, {"FAL_KEY": "test-key"}):
+            from services.fal_service import FalService
+
+            service = FalService(api_key="test-key")
+
+            with pytest.raises(ValueError, match="Authentication failed"):
+                service._handle_api_error(Exception("401 Unauthorized"), "test")
+
+    def test_handle_api_error_402(self):
+        """_handle_api_error raises ValueError for 402."""
+        with patch.dict(os.environ, {"FAL_KEY": "test-key"}):
+            from services.fal_service import FalService
+
+            service = FalService(api_key="test-key")
+
+            with pytest.raises(ValueError, match="Insufficient credits"):
+                service._handle_api_error(Exception("402 Payment Required"), "test")
+
+    def test_handle_api_error_429(self):
+        """_handle_api_error raises ValueError for 429."""
+        with patch.dict(os.environ, {"FAL_KEY": "test-key"}):
+            from services.fal_service import FalService
+
+            service = FalService(api_key="test-key")
+
+            with pytest.raises(ValueError, match="Rate limit"):
+                service._handle_api_error(Exception("429 Rate limit exceeded"), "test")
+
+    def test_handle_api_error_reraises_other(self):
+        """_handle_api_error re-raises non-handled errors."""
+        with patch.dict(os.environ, {"FAL_KEY": "test-key"}):
+            from services.fal_service import FalService
+
+            service = FalService(api_key="test-key")
+
+            with pytest.raises(RuntimeError, match="Connection timeout"):
+                service._handle_api_error(RuntimeError("Connection timeout"), "test")
+
+
+# =============================================================================
+# 4. Video Download Tests
+# =============================================================================
+
+class TestFalVideoDownload:
+    """Test download_video method."""
+
+    @pytest.mark.asyncio
+    async def test_download_video_saves_file(self):
+        """download_video saves file and returns filename."""
+        with patch.dict(os.environ, {"FAL_KEY": "test-key"}):
+            from services.fal_service import FalService
+
+            # Mock httpx client at module level
+            with patch('httpx.AsyncClient') as mock_client:
+                mock_response = MagicMock()
+                mock_response.content = b"fake video data"
+                mock_response.raise_for_status = MagicMock()
+
+                mock_client_instance = AsyncMock()
+                mock_client_instance.get.return_value = mock_response
+                mock_client_instance.__aenter__.return_value = mock_client_instance
+                mock_client_instance.__aexit__.return_value = None
+                mock_client.return_value = mock_client_instance
+
+                # Mock file operations
+                with patch('services.fal_service.api_client.os.makedirs'):
+                    mock_file = MagicMock()
+                    with patch('builtins.open', MagicMock(return_value=mock_file)):
+                        mock_file.__enter__ = MagicMock(return_value=mock_file)
+                        mock_file.__exit__ = MagicMock(return_value=False)
+                        
+                        service = FalService(api_key="test-key")
+                        result = await service.download_video(
+                            "https://fal.ai/video.mp4",
+                            "test-req-123"
+                        )
+
+                        assert result == "test-req-123.mp4"
+                        mock_file.write.assert_called_once_with(b"fake video data")
+
+    @pytest.mark.asyncio
+    async def test_download_video_http_error(self):
+        """download_video raises error on HTTP failure."""
+        with patch.dict(os.environ, {"FAL_KEY": "test-key"}):
+            from services.fal_service import FalService
+
+            with patch('httpx.AsyncClient') as mock_client:
+                mock_client_instance = AsyncMock()
+                mock_client_instance.get.side_effect = Exception("Connection refused")
+                mock_client_instance.__aenter__.return_value = mock_client_instance
+                mock_client_instance.__aexit__.return_value = None
+                mock_client.return_value = mock_client_instance
+
+                service = FalService(api_key="test-key")
+
+                with pytest.raises(ValueError, match="Failed to download"):
+                    await service.download_video(
+                        "https://fal.ai/video.mp4",
+                        "test-req-123"
+                    )
+
+
+# =============================================================================
+# 5. Check Status Tests  
+# =============================================================================
+
+class TestFalCheckStatus:
+    """Test check_video_status method."""
+
+    @pytest.mark.asyncio
+    async def test_check_status_returns_completed(self):
+        """check_video_status returns completed (fal.ai is sync)."""
+        with patch.dict(os.environ, {"FAL_KEY": "test-key"}):
+            from services.fal_service import FalService
+
+            service = FalService(api_key="test-key")
+            result = await service.check_video_status("req-123")
+
+            assert result["done"] is True
+            assert result["status"] == "completed"
+            assert result["fal_request_id"] == "req-123"

tests/test_gemini_service.py

@@ -1,814 +0,0 @@
-"""
-Rigorous Tests for Gemini AI Service.
-
-Tests cover:
-1. Initialization & API key handling
-2. Concurrency semaphores
-3. Text generation
-4. Animation prompt generation
-5. Image analysis & editing
-6. Video generation, status checking, downloading
-7. Error handling
-"""
-import pytest
-import asyncio
-import os
-import tempfile
-from unittest.mock import patch, MagicMock, AsyncMock, PropertyMock
-from datetime import datetime
-
-
-# =============================================================================
-# 1. Initialization & Configuration Tests
-# =============================================================================
-
-class TestGeminiServiceInit:
-    """Test GeminiService initialization and configuration."""
-    
-    def test_init_with_explicit_api_key(self):
-        """Service initializes with explicit API key."""
-        with patch('services.gemini_service.genai') as mock_genai:
-            from services.gemini_service import GeminiService
-            
-            service = GeminiService(api_key="test-key-123")
-            
-            assert service.api_key == "test-key-123"
-            mock_genai.Client.assert_called_once_with(api_key="test-key-123")
-    
-    def test_init_with_env_fallback(self):
-        """Service falls back to environment variable for API key."""
-        with patch('services.gemini_service.genai') as mock_genai:
-            with patch.dict(os.environ, {"GEMINI_API_KEY": "env-key-456"}):
-                from services.gemini_service import GeminiService
-                
-                service = GeminiService()
-                
-                assert service.api_key == "env-key-456"
-    
-    def test_init_fails_without_api_key(self):
-        """Service raises error when no API key available."""
-        with patch.dict(os.environ, {}, clear=True):
-            # Remove GEMINI_API_KEY if present
-            os.environ.pop("GEMINI_API_KEY", None)
-            os.environ.pop("GEMINI_API_KEYS", None)
-            
-            from services.gemini_service import get_gemini_api_key
-            
-            with pytest.raises(ValueError, match="Server Authentication Error"):
-                get_gemini_api_key()
-    
-    def test_models_dict_has_required_entries(self):
-        """MODELS dictionary has all required model names."""
-        from services.gemini_service import MODELS
-        
-        assert "text_generation" in MODELS
-        assert "image_edit" in MODELS
-        assert "video_generation" in MODELS
-        assert all(isinstance(v, str) for v in MODELS.values())
-
-
-# =============================================================================
-# 2. Semaphore Concurrency Tests
-# =============================================================================
-
-class TestSemaphoreConcurrency:
-    """Test concurrency control via semaphores."""
-    
-    def test_video_semaphore_respects_limit(self):
-        """Video semaphore uses MAX_CONCURRENT_VIDEOS."""
-        # Reset global
-        import services.gemini_service as gs
-        gs._video_semaphore = None
-        
-        with patch.object(gs, 'MAX_CONCURRENT_VIDEOS', 3):
-            gs._video_semaphore = None  # Reset
-            sem = gs.get_video_semaphore()
-            # Semaphore internal value
-            assert sem._value == 3
-    
-    def test_image_semaphore_respects_limit(self):
-        """Image semaphore uses MAX_CONCURRENT_IMAGES."""
-        import services.gemini_service as gs
-        gs._image_semaphore = None
-        
-        with patch.object(gs, 'MAX_CONCURRENT_IMAGES', 5):
-            gs._image_semaphore = None
-            sem = gs.get_image_semaphore()
-            assert sem._value == 5
-    
-    def test_text_semaphore_respects_limit(self):
-        """Text semaphore uses MAX_CONCURRENT_TEXT."""
-        import services.gemini_service as gs
-        gs._text_semaphore = None
-        
-        with patch.object(gs, 'MAX_CONCURRENT_TEXT', 10):
-            gs._text_semaphore = None
-            sem = gs.get_text_semaphore()
-            assert sem._value == 10
-    
-    def test_semaphores_are_singletons(self):
-        """Calling get_*_semaphore multiple times returns same object."""
-        import services.gemini_service as gs
-        gs._video_semaphore = None
-        gs._image_semaphore = None
-        gs._text_semaphore = None
-        
-        video1 = gs.get_video_semaphore()
-        video2 = gs.get_video_semaphore()
-        assert video1 is video2
-        
-        image1 = gs.get_image_semaphore()
-        image2 = gs.get_image_semaphore()
-        assert image1 is image2
-        
-        text1 = gs.get_text_semaphore()
-        text2 = gs.get_text_semaphore()
-        assert text1 is text2
-
-
-# =============================================================================
-# 3. Text Generation Tests
-# =============================================================================
-
-class TestTextGeneration:
-    """Test generate_text method."""
-    
-    @pytest.mark.asyncio
-    async def test_generate_text_success(self):
-        """generate_text returns text on success."""
-        with patch('services.gemini_service.genai') as mock_genai:
-            from services.gemini_service import GeminiService
-            
-            # Mock response
-            mock_response = MagicMock()
-            mock_response.text = "Generated text response"
-            mock_genai.Client.return_value.models.generate_content.return_value = mock_response
-            
-            service = GeminiService(api_key="test-key")
-            result = await service.generate_text("Hello world")
-            
-            assert result == "Generated text response"
-    
-    @pytest.mark.asyncio
-    async def test_generate_text_with_custom_model(self):
-        """generate_text uses custom model when provided."""
-        with patch('services.gemini_service.genai') as mock_genai:
-            from services.gemini_service import GeminiService
-            
-            mock_response = MagicMock()
-            mock_response.text = "Custom model response"
-            mock_genai.Client.return_value.models.generate_content.return_value = mock_response
-            
-            service = GeminiService(api_key="test-key")
-            result = await service.generate_text("Hello", model="custom-model")
-            
-            # Verify custom model was used
-            call_args = mock_genai.Client.return_value.models.generate_content.call_args
-            assert call_args.kwargs.get('model') == "custom-model"
-    
-    @pytest.mark.asyncio
-    async def test_generate_text_empty_response(self):
-        """generate_text returns empty string for None response."""
-        with patch('services.gemini_service.genai') as mock_genai:
-            from services.gemini_service import GeminiService
-            
-            mock_response = MagicMock()
-            mock_response.text = None
-            mock_genai.Client.return_value.models.generate_content.return_value = mock_response
-            
-            service = GeminiService(api_key="test-key")
-            result = await service.generate_text("Hello")
-            
-            assert result == ""
-    
-    @pytest.mark.asyncio
-    async def test_generate_text_api_error_404(self):
-        """generate_text raises ValueError for 404 error."""
-        with patch('services.gemini_service.genai') as mock_genai:
-            from services.gemini_service import GeminiService
-            
-            mock_genai.Client.return_value.models.generate_content.side_effect = Exception("404 NOT_FOUND")
-            
-            service = GeminiService(api_key="test-key")
-            
-            with pytest.raises(ValueError, match="Model not found"):
-                await service.generate_text("Hello")
-
-
-# =============================================================================
-# 4. Animation Prompt Tests
-# =============================================================================
-
-class TestAnimationPrompt:
-    """Test generate_animation_prompt method."""
-    
-    @pytest.mark.asyncio
-    async def test_generate_animation_prompt_default(self):
-        """generate_animation_prompt uses default prompt."""
-        with patch('services.gemini_service.genai') as mock_genai:
-            with patch('services.gemini_service.types'):
-                from services.gemini_service import GeminiService
-                
-                mock_response = MagicMock()
-                mock_response.text = "Subtle zoom with camera pan"
-                mock_genai.Client.return_value.models.generate_content.return_value = mock_response
-                
-                service = GeminiService(api_key="test-key")
-                result = await service.generate_animation_prompt(
-                    base64_image="base64data",
-                    mime_type="image/jpeg"
-                )
-                
-                assert result == "Subtle zoom with camera pan"
-    
-    @pytest.mark.asyncio
-    async def test_generate_animation_prompt_custom(self):
-        """generate_animation_prompt uses custom prompt when provided."""
-        with patch('services.gemini_service.genai') as mock_genai:
-            with patch('services.gemini_service.types'):
-                from services.gemini_service import GeminiService
-                
-                mock_response = MagicMock()
-                mock_response.text = "Custom animation"
-                mock_genai.Client.return_value.models.generate_content.return_value = mock_response
-                
-                service = GeminiService(api_key="test-key")
-                result = await service.generate_animation_prompt(
-                    base64_image="base64data",
-                    mime_type="image/jpeg",
-                    custom_prompt="Make it dramatic"
-                )
-                
-                assert result == "Custom animation"
-    
-    @pytest.mark.asyncio
-    async def test_generate_animation_prompt_fallback(self):
-        """generate_animation_prompt returns fallback on empty response."""
-        with patch('services.gemini_service.genai') as mock_genai:
-            with patch('services.gemini_service.types'):
-                from services.gemini_service import GeminiService
-                
-                mock_response = MagicMock()
-                mock_response.text = None
-                mock_genai.Client.return_value.models.generate_content.return_value = mock_response
-                
-                service = GeminiService(api_key="test-key")
-                result = await service.generate_animation_prompt(
-                    base64_image="base64data",
-                    mime_type="image/jpeg"
-                )
-                
-                assert result == "Cinematic subtle movement"
-
-
-# =============================================================================
-# 5. Image Analysis Tests
-# =============================================================================
-
-class TestImageAnalysis:
-    """Test analyze_image method."""
-    
-    @pytest.mark.asyncio
-    async def test_analyze_image_success(self):
-        """analyze_image returns analysis text."""
-        with patch('services.gemini_service.genai') as mock_genai:
-            with patch('services.gemini_service.types'):
-                from services.gemini_service import GeminiService
-                
-                mock_response = MagicMock()
-                mock_response.text = "This image shows a sunset over mountains"
-                mock_genai.Client.return_value.models.generate_content.return_value = mock_response
-                
-                service = GeminiService(api_key="test-key")
-                result = await service.analyze_image(
-                    base64_image="base64data",
-                    mime_type="image/jpeg",
-                    prompt="Describe this image"
-                )
-                
-                assert result == "This image shows a sunset over mountains"
-    
-    @pytest.mark.asyncio
-    async def test_analyze_image_empty_response(self):
-        """analyze_image returns empty string for None response."""
-        with patch('services.gemini_service.genai') as mock_genai:
-            with patch('services.gemini_service.types'):
-                from services.gemini_service import GeminiService
-                
-                mock_response = MagicMock()
-                mock_response.text = None
-                mock_genai.Client.return_value.models.generate_content.return_value = mock_response
-                
-                service = GeminiService(api_key="test-key")
-                result = await service.analyze_image(
-                    base64_image="base64data",
-                    mime_type="image/jpeg",
-                    prompt="Describe"
-                )
-                
-                assert result == ""
-
-
-# =============================================================================
-# 6. Image Editing Tests
-# =============================================================================
-
-class TestImageEditing:
-    """Test edit_image method."""
-    
-    @pytest.mark.asyncio
-    async def test_edit_image_returns_data_uri(self):
-        """edit_image returns base64 data URI."""
-        with patch('services.gemini_service.genai') as mock_genai:
-            from services.gemini_service import GeminiService
-            
-            # Create mock response structure
-            mock_inline_data = MagicMock()
-            mock_inline_data.data = "base64imagedata"
-            mock_inline_data.mime_type = "image/png"
-            
-            mock_part = MagicMock()
-            mock_part.inline_data = mock_inline_data
-            
-            mock_content = MagicMock()
-            mock_content.parts = [mock_part]
-            
-            mock_candidate = MagicMock()
-            mock_candidate.content = mock_content
-            
-            mock_response = MagicMock()
-            mock_response.candidates = [mock_candidate]
-            
-            mock_genai.Client.return_value.models.generate_content.return_value = mock_response
-            
-            service = GeminiService(api_key="test-key")
-            result = await service.edit_image(
-                base64_image="input-base64",
-                mime_type="image/jpeg",
-                prompt="Make it colorful"
-            )
-            
-            assert result == "data:image/png;base64,base64imagedata"
-    
-    @pytest.mark.asyncio
-    async def test_edit_image_no_candidates(self):
-        """edit_image raises error when no candidates returned."""
-        with patch('services.gemini_service.genai') as mock_genai:
-            from services.gemini_service import GeminiService
-            
-            mock_response = MagicMock()
-            mock_response.candidates = []
-            mock_genai.Client.return_value.models.generate_content.return_value = mock_response
-            
-            service = GeminiService(api_key="test-key")
-            
-            with pytest.raises(ValueError, match="No candidates returned"):
-                await service.edit_image(
-                    base64_image="input-base64",
-                    mime_type="image/jpeg",
-                    prompt="Edit"
-                )
-    
-    @pytest.mark.asyncio
-    async def test_edit_image_no_image_data(self):
-        """edit_image raises error when no image data in parts."""
-        with patch('services.gemini_service.genai') as mock_genai:
-            from services.gemini_service import GeminiService
-            
-            # Part without inline_data
-            mock_part = MagicMock()
-            mock_part.inline_data = None
-            
-            mock_content = MagicMock()
-            mock_content.parts = [mock_part]
-            
-            mock_candidate = MagicMock()
-            mock_candidate.content = mock_content
-            
-            mock_response = MagicMock()
-            mock_response.candidates = [mock_candidate]
-            
-            mock_genai.Client.return_value.models.generate_content.return_value = mock_response
-            
-            service = GeminiService(api_key="test-key")
-            
-            with pytest.raises(ValueError, match="No image data found"):
-                await service.edit_image(
-                    base64_image="input-base64",
-                    mime_type="image/jpeg",
-                    prompt="Edit"
-                )
-    
-    @pytest.mark.asyncio
-    async def test_edit_image_default_prompt(self):
-        """edit_image uses default prompt when empty."""
-        with patch('services.gemini_service.genai') as mock_genai:
-            with patch('services.gemini_service.types'):
-                from services.gemini_service import GeminiService
-                
-                mock_inline_data = MagicMock()
-                mock_inline_data.data = "base64data"
-                mock_inline_data.mime_type = "image/png"
-                
-                mock_part = MagicMock()
-                mock_part.inline_data = mock_inline_data
-                
-                mock_content = MagicMock()
-                mock_content.parts = [mock_part]
-                
-                mock_candidate = MagicMock()
-                mock_candidate.content = mock_content
-                
-                mock_response = MagicMock()
-                mock_response.candidates = [mock_candidate]
-                
-                mock_genai.Client.return_value.models.generate_content.return_value = mock_response
-                
-                service = GeminiService(api_key="test-key")
-                result = await service.edit_image(
-                    base64_image="input",
-                    mime_type="image/jpeg",
-                    prompt=""  # Empty prompt
-                )
-                
-                assert "data:" in result
-
-
-# =============================================================================
-# 7. Video Generation Tests
-# =============================================================================
-
-class TestVideoGeneration:
-    """Test start_video_generation method."""
-    
-    @pytest.mark.asyncio
-    async def test_start_video_returns_operation_dict(self):
-        """start_video_generation returns operation dictionary."""
-        with patch('services.gemini_service.genai') as mock_genai:
-            with patch('services.gemini_service.types'):
-                from services.gemini_service import GeminiService
-                
-                mock_operation = MagicMock()
-                mock_operation.name = "operations/video-123"
-                mock_operation.done = False
-                
-                mock_genai.Client.return_value.models.generate_videos.return_value = mock_operation
-                
-                service = GeminiService(api_key="test-key")
-                result = await service.start_video_generation(
-                    base64_image="base64data",
-                    mime_type="image/jpeg",
-                    prompt="Animate this"
-                )
-                
-                assert result["gemini_operation_name"] == "operations/video-123"
-                assert result["done"] == False
-                assert result["status"] == "pending"
-    
-    @pytest.mark.asyncio
-    async def test_start_video_completed_immediately(self):
-        """start_video_generation returns completed when done=True."""
-        with patch('services.gemini_service.genai') as mock_genai:
-            with patch('services.gemini_service.types'):
-                from services.gemini_service import GeminiService
-                
-                mock_operation = MagicMock()
-                mock_operation.name = "operations/video-123"
-                mock_operation.done = True
-                
-                mock_genai.Client.return_value.models.generate_videos.return_value = mock_operation
-                
-                service = GeminiService(api_key="test-key")
-                result = await service.start_video_generation(
-                    base64_image="base64data",
-                    mime_type="image/jpeg",
-                    prompt="Animate this"
-                )
-                
-                assert result["status"] == "completed"
-    
-    @pytest.mark.asyncio
-    async def test_start_video_with_params(self):
-        """start_video_generation passes aspect_ratio and resolution."""
-        with patch('services.gemini_service.genai') as mock_genai:
-            with patch('services.gemini_service.types'):
-                from services.gemini_service import GeminiService
-                
-                mock_operation = MagicMock()
-                mock_operation.name = "operations/video-123"
-                mock_operation.done = False
-                
-                mock_genai.Client.return_value.models.generate_videos.return_value = mock_operation
-                
-                service = GeminiService(api_key="test-key")
-                await service.start_video_generation(
-                    base64_image="base64data",
-                    mime_type="image/jpeg",
-                    prompt="Animate",
-                    aspect_ratio="9:16",
-                    resolution="1080p",
-                    number_of_videos=2
-                )
-                
-                # Verify config was passed
-                call_args = mock_genai.Client.return_value.models.generate_videos.call_args
-                assert call_args is not None
-
-
-# =============================================================================
-# 8. Video Status Checking Tests
-# =============================================================================
-
-class TestVideoStatusChecking:
-    """Test check_video_status method."""
-    
-    @pytest.mark.asyncio
-    async def test_check_status_pending(self):
-        """check_video_status returns pending when not done."""
-        with patch('services.gemini_service.genai') as mock_genai:
-            from services.gemini_service import GeminiService
-            
-            mock_operation = MagicMock()
-            mock_operation.done = False
-            mock_operation.error = None
-            
-            mock_genai.Client.return_value.operations.get.return_value = mock_operation
-            
-            service = GeminiService(api_key="test-key")
-            result = await service.check_video_status("operations/video-123")
-            
-            assert result["done"] == False
-            assert result["status"] == "pending"
-    
-    @pytest.mark.asyncio
-    async def test_check_status_completed_with_url(self):
-        """check_video_status returns completed with video URL."""
-        with patch('services.gemini_service.genai') as mock_genai:
-            from services.gemini_service import GeminiService
-            
-            # Build nested mock structure
-            mock_video = MagicMock()
-            mock_video.uri = "https://storage.googleapis.com/video.mp4"
-            
-            mock_generated_video = MagicMock()
-            mock_generated_video.video = mock_video
-            
-            mock_result = MagicMock()
-            mock_result.generated_videos = [mock_generated_video]
-            
-            mock_operation = MagicMock()
-            mock_operation.done = True
-            mock_operation.error = None
-            mock_operation.result = mock_result
-            
-            mock_genai.Client.return_value.operations.get.return_value = mock_operation
-            
-            service = GeminiService(api_key="test-api-key")
-            result = await service.check_video_status("operations/video-123")
-            
-            assert result["done"] == True
-            assert result["status"] == "completed"
-            assert "video_url" in result
-            assert "test-api-key" in result["video_url"]  # API key appended
-    
-    @pytest.mark.asyncio
-    async def test_check_status_operation_error(self):
-        """check_video_status returns failed on operation error."""
-        with patch('services.gemini_service.genai') as mock_genai:
-            from services.gemini_service import GeminiService
-            
-            mock_error = MagicMock()
-            mock_error.message = "Content blocked by policy"
-            
-            mock_operation = MagicMock()
-            mock_operation.done = True
-            mock_operation.error = mock_error
-            
-            mock_genai.Client.return_value.operations.get.return_value = mock_operation
-            
-            service = GeminiService(api_key="test-key")
-            result = await service.check_video_status("operations/video-123")
-            
-            assert result["done"] == True
-            assert result["status"] == "failed"
-            assert "error" in result
-    
-    @pytest.mark.asyncio
-    async def test_check_status_404_expired(self):
-        """check_video_status handles 404 for expired operation."""
-        with patch('services.gemini_service.genai') as mock_genai:
-            from services.gemini_service import GeminiService
-            
-            mock_genai.Client.return_value.operations.get.side_effect = Exception("404 NOT_FOUND")
-            
-            service = GeminiService(api_key="test-key")
-            result = await service.check_video_status("operations/expired-123")
-            
-            assert result["done"] == True
-            assert result["status"] == "failed"
-            assert "expired" in result["error"].lower()
-    
-    @pytest.mark.asyncio
-    async def test_check_status_no_video_uri(self):
-        """check_video_status returns failed when no video URI."""
-        with patch('services.gemini_service.genai') as mock_genai:
-            from services.gemini_service import GeminiService
-            
-            mock_result = MagicMock()
-            mock_result.generated_videos = []  # Empty
-            
-            mock_operation = MagicMock()
-            mock_operation.done = True
-            mock_operation.error = None
-            mock_operation.result = mock_result
-            
-            mock_genai.Client.return_value.operations.get.return_value = mock_operation
-            
-            service = GeminiService(api_key="test-key")
-            result = await service.check_video_status("operations/video-123")
-            
-            assert result["status"] == "failed"
-            assert "safety filters" in result["error"].lower()
-
-
-# =============================================================================
-# 9. Video Download Tests
-# =============================================================================
-
-class TestVideoDownload:
-    """Test download_video method."""
-    
-    @pytest.mark.asyncio
-    async def test_download_video_saves_file(self):
-        """download_video saves file and returns filename."""
-        with patch('services.gemini_service.genai'):
-            from services.gemini_service import GeminiService, DOWNLOADS_DIR
-            
-            with patch('httpx.AsyncClient') as mock_client:
-                mock_response = MagicMock()
-                mock_response.content = b"fake video data"
-                mock_response.raise_for_status = MagicMock()
-                
-                mock_client_instance = AsyncMock()
-                mock_client_instance.get.return_value = mock_response
-                mock_client_instance.__aenter__.return_value = mock_client_instance
-                mock_client_instance.__aexit__.return_value = None
-                mock_client.return_value = mock_client_instance
-                
-                service = GeminiService(api_key="test-key")
-                
-                # Use temp directory
-                with tempfile.TemporaryDirectory() as temp_dir:
-                    with patch.object(
-                        __import__('services.gemini_service', fromlist=['DOWNLOADS_DIR']),
-                        'DOWNLOADS_DIR',
-                        temp_dir
-                    ):
-                        result = await service.download_video(
-                            "https://example.com/video.mp4",
-                            "test-op-123"
-                        )
-                        
-                        assert result == "test-op-123.mp4"
-    
-    @pytest.mark.asyncio
-    async def test_download_video_http_error(self):
-        """download_video raises error on HTTP failure."""
-        with patch('services.gemini_service.genai'):
-            from services.gemini_service import GeminiService
-            
-            with patch('httpx.AsyncClient') as mock_client:
-                mock_client_instance = AsyncMock()
-                mock_client_instance.get.side_effect = Exception("Connection refused")
-                mock_client_instance.__aenter__.return_value = mock_client_instance
-                mock_client_instance.__aexit__.return_value = None
-                mock_client.return_value = mock_client_instance
-                
-                service = GeminiService(api_key="test-key")
-                
-                with pytest.raises(ValueError, match="Failed to download"):
-                    await service.download_video(
-                        "https://example.com/video.mp4",
-                        "test-op-123"
-                    )
-    
-    @pytest.mark.asyncio
-    async def test_download_video_follows_redirects(self):
-        """download_video client is configured to follow redirects."""
-        with patch('services.gemini_service.genai'):
-            from services.gemini_service import GeminiService
-            
-            with patch('httpx.AsyncClient') as mock_client:
-                mock_response = MagicMock()
-                mock_response.content = b"video data"
-                mock_response.raise_for_status = MagicMock()
-                
-                mock_client_instance = AsyncMock()
-                mock_client_instance.get.return_value = mock_response
-                mock_client_instance.__aenter__.return_value = mock_client_instance
-                mock_client_instance.__aexit__.return_value = None
-                mock_client.return_value = mock_client_instance
-                
-                service = GeminiService(api_key="test-key")
-                
-                with tempfile.TemporaryDirectory() as temp_dir:
-                    with patch('services.gemini_service.DOWNLOADS_DIR', temp_dir):
-                        await service.download_video(
-                            "https://example.com/video.mp4",
-                            "redirect-test"
-                        )
-                
-                # Verify follow_redirects=True was passed
-                mock_client.assert_called_with(timeout=120.0, follow_redirects=True)
-
-
-# =============================================================================
-# 10. Error Handling Tests
-# =============================================================================
-
-class TestErrorHandling:
-    """Test _handle_api_error method."""
-    
-    def test_handle_api_error_404(self):
-        """_handle_api_error raises ValueError for 404."""
-        with patch('services.gemini_service.genai'):
-            from services.gemini_service import GeminiService
-            
-            service = GeminiService(api_key="test-key")
-            
-            with pytest.raises(ValueError, match="Model not found"):
-                service._handle_api_error(Exception("Error 404"), "test-model")
-    
-    def test_handle_api_error_not_found(self):
-        """_handle_api_error handles NOT_FOUND in message."""
-        with patch('services.gemini_service.genai'):
-            from services.gemini_service import GeminiService
-            
-            service = GeminiService(api_key="test-key")
-            
-            with pytest.raises(ValueError, match="Model not found"):
-                service._handle_api_error(Exception("NOT_FOUND: resource"), "test-model")
-    
-    def test_handle_api_error_entity_not_found(self):
-        """_handle_api_error handles 'Requested entity was not found'."""
-        with patch('services.gemini_service.genai'):
-            from services.gemini_service import GeminiService
-            
-            service = GeminiService(api_key="test-key")
-            
-            with pytest.raises(ValueError, match="Model not found"):
-                service._handle_api_error(
-                    Exception("Requested entity was not found"),
-                    "test-model"
-                )
-    
-    def test_handle_api_error_bracket_5_pattern(self):
-        """_handle_api_error handles [5, pattern."""
-        with patch('services.gemini_service.genai'):
-            from services.gemini_service import GeminiService
-            
-            service = GeminiService(api_key="test-key")
-            
-            with pytest.raises(ValueError, match="Model not found"):
-                service._handle_api_error(
-                    Exception("Response [5, 'NOT_FOUND']"),
-                    "test-model"
-                )
-    
-    def test_handle_api_error_reraises_other(self):
-        """_handle_api_error re-raises non-404 errors."""
-        with patch('services.gemini_service.genai'):
-            from services.gemini_service import GeminiService
-            
-            service = GeminiService(api_key="test-key")
-            
-            with pytest.raises(RuntimeError, match="Connection timeout"):
-                service._handle_api_error(
-                    RuntimeError("Connection timeout"),
-                    "test-model"
-                )
-
-
-# =============================================================================
-# 11. Downloads Directory Tests
-# =============================================================================
-
-class TestDownloadsDirectory:
-    """Test downloads directory handling."""
-    
-    def test_downloads_dir_exists(self):
-        """DOWNLOADS_DIR is created on module import."""
-        from services.gemini_service import DOWNLOADS_DIR
-        
-        assert os.path.exists(DOWNLOADS_DIR)
-        assert os.path.isdir(DOWNLOADS_DIR)
-    
-    def test_downloads_dir_is_in_project(self):
-        """DOWNLOADS_DIR is within project directory."""
-        from services.gemini_service import DOWNLOADS_DIR
-        
-        assert "downloads" in DOWNLOADS_DIR
-
-
-if __name__ == "__main__":
-    pytest.main([__file__, "-v"])