Add Higgsfield provider for Kling 3.0, Sora 2, Veo 3.1

- Created HiggsFieldProvider using higgsfield-client SDK
- Added Kling 3.0 Pro and standard models for video generation
- Added Sora 2 and Veo 3.1 via Higgsfield platform
- Updated UI with premium model options
- Added higgsfield-client to requirements

Requires HIGGSFIELD_API_KEY or HF_API_KEY env var to use.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

requirements.txt

@@ -2,15 +2,14 @@ fastapi>=0.109.0
 uvicorn[standard]>=0.27.0
 aiohttp>=3.9.0
 sqlalchemy>=2.0.0
+alembic>=1.13.0
 aiosqlite>=0.19.0
 pydantic>=2.5.0
 pydantic-settings>=2.1.0
 jinja2>=3.1.0
 Pillow>=10.2.0
+apscheduler>=3.10.0
 httpx>=0.26.0
 pyyaml>=6.0
 python-multipart>=0.0.6
-python-dotenv>=1.0.0
-runpod>=1.6.0
-paramiko>=3.4.0
-wavespeed>=0.1.0
+higgsfield-client>=0.1.0

src/content_engine/api/routes_video.py

@@ -21,14 +21,20 @@ router = APIRouter(prefix="/api/video", tags=["video"])
 # Video jobs tracking
 _video_jobs: dict[str, dict] = {}
 
-# WaveSpeed provider (initialized from main.py)
+# Cloud providers (initialized from main.py)
 _wavespeed_provider = None
+_higgsfield_provider = None
 
 def init_wavespeed(provider):
     """Initialize WaveSpeed provider for cloud video generation."""
     global _wavespeed_provider
     _wavespeed_provider = provider
 
+def init_higgsfield(provider):
+    """Initialize Higgsfield provider for Kling 3.0, Sora 2, etc."""
+    global _higgsfield_provider
+    _higgsfield_provider = provider
+
 # Pod state is shared from routes_pod
 def _get_pod_state():
     from content_engine.api.routes_pod import _pod_state
@@ -145,11 +151,22 @@ async def generate_video_cloud(
     num_frames: int = Form(81),
     fps: int = Form(24),
     seed: int = Form(-1),
+    backend: str = Form("wavespeed"),  # wavespeed or higgsfield
 ):
-    """Generate a video using WaveSpeed cloud API (Kling, WAN I2V, Veo, etc)."""
+    """Generate a video using cloud API (WaveSpeed or Higgsfield)."""
     import random
     import httpx
 
+    # Route to Higgsfield for Kling 3.0 models
+    if backend == "higgsfield" or model.startswith("kling-3"):
+        return await generate_video_higgsfield(
+            image=image,
+            prompt=prompt,
+            model=model,
+            duration=max(3, num_frames // 24),  # Convert frames to seconds
+            seed=seed,
+        )
+
     if not _wavespeed_provider:
         raise HTTPException(500, "WaveSpeed API not configured")
 
@@ -185,6 +202,134 @@ async def generate_video_cloud(
     }
 
 
+@router.post("/generate/higgsfield")
+async def generate_video_higgsfield(
+    image: UploadFile = File(...),
+    prompt: str = Form("smooth cinematic motion"),
+    model: str = Form("kling-3.0"),
+    duration: int = Form(5),
+    resolution: str = Form("720p"),
+    enable_audio: bool = Form(False),
+    seed: int = Form(-1),
+):
+    """Generate a video using Higgsfield (Kling 3.0, Sora 2, Veo 3.1)."""
+    import random
+
+    if not _higgsfield_provider:
+        raise HTTPException(500, "Higgsfield API not configured - set HIGGSFIELD_API_KEY")
+
+    job_id = str(uuid.uuid4())[:8]
+    seed = seed if seed >= 0 else random.randint(0, 2**32 - 1)
+
+    # Read the image
+    image_bytes = await image.read()
+
+    # Create job entry
+    _video_jobs[job_id] = {
+        "status": "running",
+        "seed": seed,
+        "started_at": time.time(),
+        "duration": duration,
+        "model": model,
+        "backend": "higgsfield",
+        "message": "Starting Higgsfield video generation...",
+    }
+
+    logger.info("Higgsfield video generation started: %s (model=%s)", job_id, model)
+
+    # Start background task
+    asyncio.create_task(_generate_higgsfield_video(
+        job_id, image_bytes, prompt, model, duration, resolution, enable_audio, seed
+    ))
+
+    return {
+        "job_id": job_id,
+        "status": "running",
+        "seed": seed,
+        "model": model,
+        "backend": "higgsfield",
+        "estimated_time": f"~{duration * 10}-{duration * 20} seconds",
+    }
+
+
+async def _generate_higgsfield_video(
+    job_id: str,
+    image_bytes: bytes,
+    prompt: str,
+    model: str,
+    duration: int,
+    resolution: str,
+    enable_audio: bool,
+    seed: int,
+):
+    """Background task to generate video via Higgsfield API."""
+    try:
+        _video_jobs[job_id]["message"] = "Uploading image to Higgsfield..."
+
+        # Upload image to temp URL
+        image_url = await _higgsfield_provider._upload_temp_image(image_bytes) if hasattr(_higgsfield_provider, '_upload_temp_image') else None
+
+        if not image_url:
+            # Fall back to base64 data URL
+            import base64
+            image_b64 = base64.b64encode(image_bytes).decode("utf-8")
+            image_url = f"data:image/png;base64,{image_b64}"
+
+        _video_jobs[job_id]["message"] = f"Generating video with {model}..."
+
+        # Generate video
+        result = await _higgsfield_provider.generate_video(
+            prompt=prompt,
+            model=model,
+            duration=duration,
+            resolution=resolution,
+            enable_audio=enable_audio,
+            image_url=image_url,
+        )
+
+        video_url = result.get("video_url")
+        if not video_url:
+            _video_jobs[job_id]["status"] = "failed"
+            _video_jobs[job_id]["error"] = "No video URL in response"
+            return
+
+        # Download the video
+        _video_jobs[job_id]["message"] = "Downloading generated video..."
+        import httpx
+        async with httpx.AsyncClient(timeout=120) as client:
+            video_resp = await client.get(video_url)
+
+            if video_resp.status_code != 200:
+                _video_jobs[job_id]["status"] = "failed"
+                _video_jobs[job_id]["error"] = "Failed to download video"
+                return
+
+            # Save to local output directory
+            from content_engine.config import settings
+            output_dir = settings.paths.output_dir / "videos"
+            output_dir.mkdir(parents=True, exist_ok=True)
+
+            ext = ".mp4"
+            if video_url.endswith(".webm"):
+                ext = ".webm"
+
+            local_path = output_dir / f"video_{job_id}{ext}"
+            local_path.write_bytes(video_resp.content)
+
+            _video_jobs[job_id]["status"] = "completed"
+            _video_jobs[job_id]["output_path"] = str(local_path)
+            _video_jobs[job_id]["completed_at"] = time.time()
+            _video_jobs[job_id]["filename"] = local_path.name
+            _video_jobs[job_id]["message"] = "Video generated successfully!"
+
+            logger.info("Higgsfield video saved: %s", local_path)
+
+    except Exception as e:
+        logger.error("Higgsfield video generation failed: %s", e)
+        _video_jobs[job_id]["status"] = "failed"
+        _video_jobs[job_id]["error"] = str(e)
+
+
 async def _poll_wavespeed_video(poll_url: str, api_key: str, job_id: str, max_attempts: int = 120, interval: float = 3.0) -> str | None:
     """Poll the WaveSpeed async video job URL until outputs are ready.
 

src/content_engine/api/ui.html

@@ -990,10 +990,17 @@ select { cursor: pointer; }
             <div id="video-cloud-model-select">
               <label>Video Model</label>
               <select id="video-cloud-model">
+                <optgroup label="⭐ Kling 3.0 (Higgsfield) - Best Quality">
+                  <option value="kling-3.0-pro">Kling 3.0 Pro (15s, Audio)</option>
+                  <option value="kling-3.0">Kling 3.0 (Up to 15s)</option>
+                </optgroup>
+                <optgroup label="⭐ Premium (Higgsfield)">
+                  <option value="sora-2-hf">Sora 2 (OpenAI via Higgsfield)</option>
+                  <option value="veo-3.1-hf">Veo 3.1 (Google via Higgsfield)</option>
+                </optgroup>
                 <optgroup label="Higgsfield (Cinematic Motion)">
                   <option value="higgsfield-dop">Higgsfield DoP (5s Cinematic)</option>
                   <option value="higgsfield-dop-turbo">Higgsfield DoP Turbo (Fast)</option>
-                  <option value="higgsfield-dop-lite">Higgsfield DoP Lite</option>
                 </optgroup>
                 <optgroup label="WAN 2.6 I2V (Alibaba)">
                   <option value="wan-2.6-i2v-pro" selected>WAN 2.6 I2V Pro (Best)</option>
@@ -1007,8 +1014,8 @@ select { cursor: pointer; }
                   <option value="wan-2.2-i2v-1080p">WAN 2.2 I2V 1080p</option>
                   <option value="wan-2.2-i2v-720p">WAN 2.2 I2V 720p</option>
                 </optgroup>
-                <optgroup label="Kling (Kuaishou)">
-                  <option value="kling-o3-pro">Kling O3 Pro (Highest Quality)</option>
+                <optgroup label="Kling (WaveSpeed)">
+                  <option value="kling-o3-pro">Kling O3 Pro</option>
                   <option value="kling-o3">Kling O3</option>
                 </optgroup>
                 <optgroup label="Seedance (ByteDance)">

src/content_engine/main.py

@@ -129,6 +129,19 @@ async def lifespan(app: FastAPI):
     else:
         logger.info("WaveSpeed not configured — cloud generation disabled")
 
+    # Initialize Higgsfield cloud provider if API key is set
+    higgsfield_provider = None
+    higgsfield_key = os.environ.get("HIGGSFIELD_API_KEY") or os.environ.get("HF_API_KEY")
+    if higgsfield_key:
+        try:
+            from content_engine.services.cloud_providers.higgsfield_provider import HiggsFieldProvider
+            higgsfield_provider = HiggsFieldProvider()
+            logger.info("Higgsfield cloud provider initialized (Kling 3.0, Sora 2, Veo 3.1)")
+        except Exception as e:
+            logger.warning("Failed to initialize Higgsfield provider: %s", e)
+    else:
+        logger.info("Higgsfield not configured — set HIGGSFIELD_API_KEY for Kling 3.0/Sora 2")
+
     # Initialize route dependencies
     routes_generation.init_routes(
         local_worker, template_engine, variation_engine, character_profiles,
@@ -138,9 +151,11 @@ async def lifespan(app: FastAPI):
     routes_catalog.init_routes(catalog)
     routes_system.init_routes(comfyui_client, catalog, template_engine, character_profiles)
 
-    # Initialize video routes with WaveSpeed provider for cloud video generation
+    # Initialize video routes with cloud providers
     if wavespeed_provider:
         routes_video.init_wavespeed(wavespeed_provider)
+    if higgsfield_provider:
+        routes_video.init_higgsfield(higgsfield_provider)
 
     # Initialize LoRA trainer (local)
     from content_engine.services.lora_trainer import LoRATrainer

src/content_engine/services/cloud_providers/higgsfield_provider.py

@@ -0,0 +1,314 @@
+"""Higgsfield.ai cloud provider — access to Kling 3.0, Sora 2, Veo 3.1, and more.
+
+Higgsfield provides a unified API for multiple AI video and image generation models
+including Kling 3.0 (Kuaishou), Sora 2 (OpenAI), Veo 3.1 (Google), WAN 2.5 (Alibaba),
+and their own Higgsfield Soul for character consistency.
+
+SDK: pip install higgsfield-client
+Docs: https://cloud.higgsfield.ai/
+"""
+
+from __future__ import annotations
+
+import asyncio
+import base64
+import logging
+import os
+import time
+import uuid
+from typing import Any
+
+import httpx
+
+from content_engine.services.cloud_providers.base import CloudGenerationResult, CloudProvider
+
+logger = logging.getLogger(__name__)
+
+# Model IDs for Higgsfield platform
+# Format: provider/model/task
+TEXT_TO_IMAGE_MODELS = {
+    "seedream-4": "bytedance/seedream/v4/text-to-image",
+    "seedream-4.5": "bytedance/seedream/v4.5/text-to-image",
+    "nano-banana-pro": "google/nano-banana-pro/text-to-image",
+    "flux-2": "black-forest-labs/flux-2/text-to-image",
+    "gpt-image": "openai/gpt-image/text-to-image",
+    "default": "bytedance/seedream/v4/text-to-image",
+}
+
+TEXT_TO_VIDEO_MODELS = {
+    "kling-3.0": "kuaishou/kling/v3.0/text-to-video",
+    "kling-3.0-pro": "kuaishou/kling/v3.0-pro/text-to-video",
+    "sora-2": "openai/sora-2/text-to-video",
+    "veo-3.1": "google/veo-3.1/text-to-video",
+    "wan-2.5": "alibaba/wan-2.5/text-to-video",
+    "seedance-pro": "bytedance/seedance/pro/text-to-video",
+    "default": "kuaishou/kling/v3.0/text-to-video",
+}
+
+IMAGE_TO_VIDEO_MODELS = {
+    "kling-3.0": "kuaishou/kling/v3.0/image-to-video",
+    "kling-3.0-pro": "kuaishou/kling/v3.0-pro/image-to-video",
+    "sora-2": "openai/sora-2/image-to-video",
+    "veo-3.1": "google/veo-3.1/image-to-video",
+    "wan-2.5": "alibaba/wan-2.5/image-to-video",
+    "higgsfield-dop": "higgsfield/dop/image-to-video",
+    "default": "kuaishou/kling/v3.0/image-to-video",
+}
+
+IMAGE_EDIT_MODELS = {
+    "higgsfield-soul": "higgsfield/soul/image-to-image",
+    "seedream-4-edit": "bytedance/seedream/v4/edit",
+    "default": "higgsfield/soul/image-to-image",
+}
+
+
+class HiggsFieldProvider(CloudProvider):
+    """Cloud provider using Higgsfield.ai for Kling 3.0, Sora 2, Veo 3.1, etc."""
+
+    def __init__(self, api_key: str = None, api_secret: str = None):
+        """Initialize with Higgsfield credentials.
+
+        Can use either:
+        - Combined key (HF_KEY env var)
+        - Separate key/secret (HF_API_KEY/HF_API_SECRET env vars)
+        """
+        self._api_key = api_key or os.getenv("HIGGSFIELD_API_KEY") or os.getenv("HF_API_KEY")
+        self._api_secret = api_secret or os.getenv("HIGGSFIELD_API_SECRET") or os.getenv("HF_API_SECRET")
+        self._combined_key = os.getenv("HF_KEY")
+
+        self._http_client = httpx.AsyncClient(timeout=300)
+        self._client = None
+
+        # Try to initialize SDK if available
+        try:
+            from higgsfield_client import HiggsFieldClient
+            if self._combined_key:
+                self._client = HiggsFieldClient()
+            elif self._api_key and self._api_secret:
+                self._client = HiggsFieldClient(api_key=self._api_key, api_secret=self._api_secret)
+            logger.info("Higgsfield SDK initialized")
+        except ImportError:
+            logger.warning("higgsfield-client not installed, using direct API")
+        except Exception as e:
+            logger.warning("Failed to init Higgsfield SDK: %s", e)
+
+    @property
+    def name(self) -> str:
+        return "higgsfield"
+
+    async def is_available(self) -> bool:
+        """Check if Higgsfield API is configured."""
+        return bool(self._client or self._api_key)
+
+    def _resolve_t2i_model(self, model_name: str | None) -> str:
+        """Resolve friendly name to Higgsfield model ID for text-to-image."""
+        if model_name and model_name in TEXT_TO_IMAGE_MODELS:
+            return TEXT_TO_IMAGE_MODELS[model_name]
+        if model_name:
+            return model_name
+        return TEXT_TO_IMAGE_MODELS["default"]
+
+    def _resolve_t2v_model(self, model_name: str | None) -> str:
+        """Resolve friendly name to Higgsfield model ID for text-to-video."""
+        if model_name and model_name in TEXT_TO_VIDEO_MODELS:
+            return TEXT_TO_VIDEO_MODELS[model_name]
+        if model_name:
+            return model_name
+        return TEXT_TO_VIDEO_MODELS["default"]
+
+    def _resolve_i2v_model(self, model_name: str | None) -> str:
+        """Resolve friendly name to Higgsfield model ID for image-to-video."""
+        if model_name and model_name in IMAGE_TO_VIDEO_MODELS:
+            return IMAGE_TO_VIDEO_MODELS[model_name]
+        if model_name:
+            return model_name
+        return IMAGE_TO_VIDEO_MODELS["default"]
+
+    def _resolve_edit_model(self, model_name: str | None) -> str:
+        """Resolve friendly name to Higgsfield model ID for image editing."""
+        if model_name and model_name in IMAGE_EDIT_MODELS:
+            return IMAGE_EDIT_MODELS[model_name]
+        if model_name:
+            return model_name
+        return IMAGE_EDIT_MODELS["default"]
+
+    async def generate_image(
+        self,
+        prompt: str,
+        model: str | None = None,
+        resolution: str = "2K",
+        aspect_ratio: str = "16:9",
+        **kwargs,
+    ) -> CloudGenerationResult:
+        """Generate an image using Higgsfield text-to-image models."""
+        start = time.time()
+        model_id = self._resolve_t2i_model(model)
+
+        if self._client:
+            # Use SDK
+            try:
+                result = self._client.subscribe(
+                    model_id,
+                    {
+                        "prompt": prompt,
+                        "resolution": resolution,
+                        "aspect_ratio": aspect_ratio,
+                        **kwargs,
+                    }
+                )
+
+                # Extract image URL
+                images = result.get("images", [])
+                if not images:
+                    raise RuntimeError(f"No images in Higgsfield response: {result}")
+
+                image_url = images[0].get("url") if isinstance(images[0], dict) else images[0]
+
+                # Download image
+                resp = await self._http_client.get(image_url)
+                resp.raise_for_status()
+
+                return CloudGenerationResult(
+                    job_id=str(uuid.uuid4()),
+                    image_bytes=resp.content,
+                    generation_time_seconds=time.time() - start,
+                )
+            except Exception as e:
+                logger.error("Higgsfield image generation failed: %s", e)
+                raise
+        else:
+            raise RuntimeError("Higgsfield SDK not initialized")
+
+    async def generate_video(
+        self,
+        prompt: str,
+        model: str | None = None,
+        duration: int = 5,
+        resolution: str = "720p",
+        aspect_ratio: str = "16:9",
+        enable_audio: bool = False,
+        image_url: str | None = None,
+        **kwargs,
+    ) -> dict:
+        """Generate a video using Higgsfield models (Kling 3.0, Sora 2, Veo 3.1, etc.).
+
+        Args:
+            prompt: Text description of desired video
+            model: Model to use (kling-3.0, sora-2, veo-3.1, etc.)
+            duration: Video duration in seconds (3-15 for Kling 3.0)
+            resolution: Output resolution (720p, 1080p)
+            aspect_ratio: Aspect ratio (16:9, 9:16, 1:1)
+            enable_audio: Enable audio generation (Kling 3.0 supports this)
+            image_url: Reference image URL for image-to-video
+
+        Returns:
+            Dict with job_id, status, and video_url when complete
+        """
+        start = time.time()
+
+        # Choose model based on whether we have an image
+        if image_url:
+            model_id = self._resolve_i2v_model(model)
+        else:
+            model_id = self._resolve_t2v_model(model)
+
+        if self._client:
+            try:
+                payload = {
+                    "prompt": prompt,
+                    "resolution": resolution,
+                    "aspect_ratio": aspect_ratio,
+                    "duration": duration,
+                }
+
+                if enable_audio:
+                    payload["enable_audio"] = True
+
+                if image_url:
+                    payload["image"] = image_url
+
+                payload.update(kwargs)
+
+                # Submit and wait for result
+                result = self._client.subscribe(model_id, payload)
+
+                # Extract video URL
+                video_url = None
+                if "video" in result:
+                    video_url = result["video"]
+                elif "outputs" in result and result["outputs"]:
+                    video_url = result["outputs"][0]
+                elif "output" in result:
+                    video_url = result["output"]
+
+                if not video_url:
+                    raise RuntimeError(f"No video URL in Higgsfield response: {result}")
+
+                return {
+                    "job_id": str(uuid.uuid4()),
+                    "status": "completed",
+                    "video_url": video_url,
+                    "generation_time": time.time() - start,
+                }
+
+            except Exception as e:
+                logger.error("Higgsfield video generation failed: %s", e)
+                raise
+        else:
+            raise RuntimeError("Higgsfield SDK not initialized")
+
+    async def submit_generation(
+        self,
+        positive_prompt: str,
+        negative_prompt: str = "",
+        model: str | None = None,
+        width: int = 1024,
+        height: int = 1024,
+        seed: int = -1,
+        **kwargs,
+    ) -> str:
+        """Submit an image generation job. Returns job ID."""
+        # For now, use synchronous generation
+        result = await self.generate_image(
+            prompt=positive_prompt,
+            model=model,
+            **kwargs,
+        )
+
+        # Cache result for get_result
+        self._last_result = {
+            "job_id": result.job_id,
+            "result": result,
+            "timestamp": time.time(),
+        }
+
+        return result.job_id
+
+    async def check_status(self, job_id: str) -> str:
+        """Check job status."""
+        if hasattr(self, '_last_result') and self._last_result.get("job_id") == job_id:
+            return "completed"
+        return "unknown"
+
+    async def get_result(self, job_id: str) -> CloudGenerationResult:
+        """Get the generation result."""
+        if not hasattr(self, '_last_result') or self._last_result.get("job_id") != job_id:
+            raise RuntimeError(f"No cached result for job {job_id}")
+        return self._last_result["result"]
+
+    async def generate(
+        self,
+        positive_prompt: str,
+        negative_prompt: str = "",
+        model: str | None = None,
+        width: int = 1024,
+        height: int = 1024,
+        seed: int = -1,
+        **kwargs,
+    ) -> CloudGenerationResult:
+        """Convenience method: generate image in one call."""
+        return await self.generate_image(
+            prompt=positive_prompt,
+            model=model,
+            **kwargs,
+        )