Интеграция Kling для генерации видео

Интеграция Kling для генерации видео

Kling от Kuaishou — одна из сильнейших коммерческих платформ видеогенерации с открытым API. Поддерживает text-to-video и image-to-video, до 30 секунд, 1080p. Текущий выбор для production-интеграций в рекламе и контент-продакшне.

REST API интеграция

import httpx
import asyncio
import jwt
import time

class KlingClient:
    def __init__(self, access_key: str, secret_key: str):
        self.access_key = access_key
        self.secret_key = secret_key
        self.base_url = "https://api.klingai.com"

    def _get_jwt_token(self) -> str:
        payload = {
            "iss": self.access_key,
            "exp": int(time.time()) + 1800,
            "nbf": int(time.time()) - 5
        }
        return jwt.encode(payload, self.secret_key, algorithm="HS256")

    async def create_text_to_video(
        self,
        prompt: str,
        negative_prompt: str = "",
        model: str = "kling-v1-5",  # kling-v1, kling-v1-5
        mode: str = "std",          # std, pro
        duration: str = "5",        # "5" или "10"
        aspect_ratio: str = "16:9", # 16:9, 9:16, 1:1
        cfg_scale: float = 0.5
    ) -> str:
        async with httpx.AsyncClient() as client:
            resp = await client.post(
                f"{self.base_url}/v1/videos/text2video",
                headers={"Authorization": f"Bearer {self._get_jwt_token()}"},
                json={
                    "model_name": model,
                    "prompt": prompt,
                    "negative_prompt": negative_prompt,
                    "cfg_scale": cfg_scale,
                    "mode": mode,
                    "aspect_ratio": aspect_ratio,
                    "duration": duration
                }
            )
            resp.raise_for_status()
            return resp.json()["data"]["task_id"]

    async def create_image_to_video(
        self,
        image_url: str,
        prompt: str = "",
        duration: str = "5",
        cfg_scale: float = 0.5
    ) -> str:
        async with httpx.AsyncClient() as client:
            resp = await client.post(
                f"{self.base_url}/v1/videos/image2video",
                headers={"Authorization": f"Bearer {self._get_jwt_token()}"},
                json={
                    "model_name": "kling-v1-5",
                    "image": image_url,
                    "prompt": prompt,
                    "cfg_scale": cfg_scale,
                    "duration": duration
                }
            )
            resp.raise_for_status()
            return resp.json()["data"]["task_id"]

    async def get_task_result(self, task_id: str) -> dict:
        async with httpx.AsyncClient() as client:
            resp = await client.get(
                f"{self.base_url}/v1/videos/text2video/{task_id}",
                headers={"Authorization": f"Bearer {self._get_jwt_token()}"}
            )
            return resp.json()["data"]

    async def wait_and_download(self, task_id: str, timeout: int = 300) -> bytes:
        for _ in range(timeout // 5):
            await asyncio.sleep(5)
            data = await self.get_task_result(task_id)

            if data["task_status"] == "succeed":
                video_url = data["task_result"]["videos"][0]["url"]
                async with httpx.AsyncClient() as client:
                    video_resp = await client.get(video_url, follow_redirects=True)
                    return video_resp.content

            elif data["task_status"] == "failed":
                raise RuntimeError(f"Kling generation failed")

        raise TimeoutError("Kling generation timeout")

FastAPI обёртка

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()
kling = KlingClient(KLING_ACCESS_KEY, KLING_SECRET_KEY)

class VideoRequest(BaseModel):
    prompt: str
    negative_prompt: str = ""
    duration: str = "5"
    mode: str = "std"

@app.post("/generate/text-to-video")
async def generate_video(req: VideoRequest):
    task_id = await kling.create_text_to_video(
        prompt=req.prompt,
        negative_prompt=req.negative_prompt,
        duration=req.duration,
        mode=req.mode
    )
    return {"task_id": task_id}

@app.get("/task/{task_id}")
async def check_task(task_id: str):
    return await kling.get_task_result(task_id)

Kling v1.5 std: ~$0.07 за 5-сек видео, pro: ~$0.14. В режиме std среднее время генерации — 1–3 минуты для 5 сек. Pro — 3–5 минут с значительно лучшим качеством движений. Сроки интеграции — 1–2 дня.