""" Pollinations Video Provider – generates videos from images/prompts via Pollinations.ai API. Supports multiple models: veo, seedance, seedance-pro, wan, klein, ltx-2, etc. Env vars: POLLINATIONS_API_KEY - API key from enter.pollinations.ai (optional for limited use) POLLINATIONS_VIDEO_MODEL - Model to use (default: seedance) Docs: https://gen.pollinations.ai/docs """ from __future__ import annotations import asyncio import base64 import logging import os from collections.abc import Callable from pathlib import Path from typing import Any from urllib.parse import quote import httpx from providers.ports import VideoProvider logger = logging.getLogger(__name__) POLLINATIONS_BASE_URL = "https://gen.pollinations.ai" # Available video models (from OpenAPI spec) VIDEO_MODELS = [ "veo", # Google Veo - text-to-video only (4-8 seconds) "seedance", # Text-to-video and image-to-video (2-10 seconds) "seedance-pro", # Higher quality seedance "wan", # Wan model "klein", # Klein model "klein-large", # Klein large "ltx-2", # LTX-2 model ] # Duration limits per model (from OpenAPI spec) DURATION_LIMITS = { "veo": (4, 8), # 4, 6, or 8 seconds only "seedance": (2, 10), # 2-10 seconds "seedance-pro": (2, 10), "wan": (2, 10), "klein": (2, 10), "klein-large": (2, 10), "ltx-2": (2, 10), } class PollinationsVideoProvider(VideoProvider): """Wraps the Pollinations.ai API for video generation (image-to-video).""" name = "pollinations" def __init__(self, api_key: str | None = None): """Initialize the Pollinations video provider. Args: api_key: Optional API key. If not provided, uses POLLINATIONS_API_KEY env var. Can work without key for limited usage. """ self.api_key = api_key or os.getenv("POLLINATIONS_API_KEY", "") self.base_url = POLLINATIONS_BASE_URL self.model = os.getenv("POLLINATIONS_VIDEO_MODEL", "seedance") if self.model not in VIDEO_MODELS: logger.warning( "Unknown video model '%s', using 'seedance'. Available: %s", self.model, VIDEO_MODELS, ) self.model = "seedance" def _upload_image_to_temp_url(self, image_path: Path) -> str: """Upload image to temporary hosting service and return URL. Uses 0x0.st for temporary file hosting. Args: image_path: Path to local image file Returns: Public URL of uploaded image """ logger.debug("Uploading image to temp hosting: %s", image_path) with open(image_path, "rb") as f: files = {"file": (image_path.name, f, "image/png")} response = httpx.post( "https://0x0.st", files=files, timeout=60, ) response.raise_for_status() url = response.text.strip() logger.debug("Image uploaded to: %s", url) return url def _image_to_data_url(self, image_path: Path) -> str: """Convert local image to base64 data URL. Args: image_path: Path to local image file Returns: Base64 data URL string """ with open(image_path, "rb") as f: image_data = f.read() # Detect mime type from extension ext = image_path.suffix.lower() mime_types = { ".png": "image/png", ".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".gif": "image/gif", ".webp": "image/webp", } mime_type = mime_types.get(ext, "image/png") b64_data = base64.b64encode(image_data).decode("utf-8") return f"data:{mime_type};base64,{b64_data}" async def generate( self, image_path: Path, video_prompt: str, output_path: Path, on_progress: Callable[[int], None] | None = None, **kwargs: Any, ) -> Path: """Async wrapper for image-to-video generation. Extra kwargs (model, duration, aspect_ratio, audio, seed, use_data_url) are passed to _generate_sync. """ return await asyncio.to_thread( self._generate_sync, image_path, video_prompt, output_path, kwargs.get('model'), kwargs.get('duration', 5), kwargs.get('aspect_ratio', '16:9'), kwargs.get('audio', False), kwargs.get('seed', -1), kwargs.get('use_data_url', False), ) def _generate_sync( self, image_path: str | Path, video_prompt: str, output_path: str | Path, model: str | None = None, duration: int = 5, aspect_ratio: str = "16:9", audio: bool = False, seed: int = -1, use_data_url: bool = False, ) -> Path: """Sync implementation of generate.""" image_path = Path(image_path) output_path = Path(output_path) output_path.parent.mkdir(parents=True, exist_ok=True) model = model or self.model # Validate model if model not in VIDEO_MODELS: logger.warning("Unknown model '%s', using 'seedance'", model) model = "seedance" # Validate duration min_dur, max_dur = DURATION_LIMITS.get(model, (2, 10)) if duration < min_dur or duration > max_dur: logger.warning( "Duration %d out of range for %s (%d-%d), clamping", duration, model, min_dur, max_dur ) duration = max(min_dur, min(max_dur, duration)) # Get image URL if use_data_url: image_url = self._image_to_data_url(image_path) else: image_url = self._upload_image_to_temp_url(image_path) # URL encode the prompt encoded_prompt = quote(video_prompt, safe="") # Build URL with query parameters url = f"{self.base_url}/image/{encoded_prompt}" params = { "model": model, "image": image_url, "duration": duration, "aspectRatio": aspect_ratio, "seed": seed, } # Add model-specific params if model == "veo" and audio: params["audio"] = "true" # Build headers headers = {"Accept": "video/*"} if self.api_key: headers["Authorization"] = f"Bearer {self.api_key}" logger.info( "Generating video via Pollinations: model=%s duration=%ds → %s", model, duration, output_path, ) logger.debug("Prompt: %s", video_prompt[:100]) # Make request with very long timeout (video generation takes time) with httpx.Client(timeout=600) as client: response = client.get(url, params=params, headers=headers) response.raise_for_status() # Save video output_path.write_bytes(response.content) logger.info("Video saved: %s (%d bytes)", output_path, len(response.content)) return output_path @staticmethod def list_models() -> list[str]: """Return list of available video models.""" return VIDEO_MODELS.copy()