Video Providers

Генерация видео через Kling и Fal.ai.

Возможности

• Text-to-Video генерация
• Image-to-Video (анимация изображений)
• Асинхронное ожидание результата
• Различные стили и параметры

Kling Provider

Базовое использование

from kit.providers.video import KlingProvider

provider = KlingProvider(
    access_key="your_access_key",
    secret_key="your_secret_key"
)

# Text to Video
task = await provider.text_to_video(
    prompt="Красивый закат над океаном, волны бьются о скалы",
    duration=5,  # секунды
    aspect_ratio="16:9"
)

print(f"Task ID: {task.task_id}")

# Ожидание результата
result = await provider.wait_for_completion(task.task_id)
print(f"Video URL: {result.video_url}")

Image to Video

# Анимация статичного изображения
task = await provider.image_to_video(
    image_path="landscape.jpg",
    prompt="Камера медленно движется вперёд, облака плывут",
    duration=5
)

result = await provider.wait_for_completion(task.task_id)

Параметры генерации

task = await provider.text_to_video(
    prompt="Космический корабль летит сквозь астероидное поле",
    duration=10,
    aspect_ratio="16:9",
    mode="high_quality",  # standard, high_quality
    cfg_scale=0.5,        # 0-1, adherence to prompt
    seed=42               # для воспроизводимости
)

Проверка статуса

# Ручная проверка статуса
status = await provider.get_status(task.task_id)
print(f"Status: {status.state}")  # pending, processing, completed, failed

if status.state == "processing":
    print(f"Progress: {status.progress}%")

Fal.ai Provider

Базовое использование

from kit.providers.video import FalProvider

provider = FalProvider(api_key="your_fal_key")

# Генерация через модель
result = await provider.generate(
    model="fal-ai/fast-svd-lcm",
    input={
        "image_url": "https://example.com/image.jpg",
        "motion_bucket_id": 127,
        "fps": 25
    }
)

print(f"Video: {result.video_url}")

Доступные модели

# SVD (Stable Video Diffusion)
result = await provider.generate(
    model="fal-ai/fast-svd-lcm",
    input={"image_url": "...", "motion_bucket_id": 180}
)

# AnimateDiff
result = await provider.generate(
    model="fal-ai/animatediff-v2v",
    input={
        "video_url": "...",
        "prompt": "anime style, high quality"
    }
)

API Reference

KlingProvider

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

    async def text_to_video(
        self,
        prompt: str,
        duration: int = 5,
        aspect_ratio: str = "16:9",
        mode: str = "standard",
        negative_prompt: str = None,
        cfg_scale: float = 0.5,
        seed: int = None
    ) -> VideoTask

    async def image_to_video(
        self,
        image_path: str,
        prompt: str,
        duration: int = 5,
        mode: str = "standard"
    ) -> VideoTask

    async def get_status(self, task_id: str) -> VideoStatus
    async def wait_for_completion(
        self,
        task_id: str,
        timeout: float = 600,
        poll_interval: float = 5
    ) -> VideoResult

    async def close(self) -> None

FalProvider

class FalProvider:
    def __init__(self, api_key: str)

    async def generate(
        self,
        model: str,
        input: dict,
        timeout: float = 300
    ) -> VideoResult

    async def close(self) -> None

Data Classes

@dataclass
class VideoTask:
    task_id: str
    status: str
    created_at: datetime

@dataclass
class VideoStatus:
    task_id: str
    state: str  # pending, processing, completed, failed
    progress: float = 0
    error: str = None

@dataclass
class VideoResult:
    task_id: str
    video_url: str
    duration: float
    width: int
    height: int

Примеры из production

Autoshorts — генерация клипов

class VideoGenerator:
    def __init__(self):
        self.kling = KlingProvider(
            access_key=settings.kling_access_key,
            secret_key=settings.kling_secret_key
        )

    async def generate_scene(self, scene: dict) -> str:
        """Генерация видео для одной сцены."""
        task = await self.kling.text_to_video(
            prompt=scene["visual_prompt"],
            duration=scene["duration"],
            aspect_ratio="9:16",  # Вертикальное для shorts
            mode="high_quality"
        )

        result = await self.kling.wait_for_completion(
            task.task_id,
            timeout=600
        )

        return result.video_url

    async def generate_all_scenes(self, scenes: List[dict]) -> List[str]:
        """Генерация всех сцен параллельно."""
        tasks = []
        for scene in scenes:
            task = await self.kling.text_to_video(
                prompt=scene["visual_prompt"],
                duration=scene["duration"]
            )
            tasks.append(task.task_id)

        # Ожидание всех
        results = []
        for task_id in tasks:
            result = await self.kling.wait_for_completion(task_id)
            results.append(result.video_url)

        return results

Music Video Generator — анимация обложек

class CoverAnimator:
    def __init__(self):
        self.kling = KlingProvider(...)

    async def animate_cover(
        self,
        cover_path: str,
        music_mood: str,
        duration: int = 10
    ) -> str:
        """Анимация обложки альбома под настроение музыки."""

        mood_prompts = {
            "energetic": "dynamic camera movement, pulsing energy, vibrant",
            "calm": "slow gentle movement, peaceful atmosphere, soft flow",
            "dark": "mysterious shadows, dramatic lighting, intense"
        }

        task = await self.kling.image_to_video(
            image_path=cover_path,
            prompt=f"Album cover animation, {mood_prompts.get(music_mood, '')}",
            duration=duration,
            mode="high_quality"
        )

        result = await self.kling.wait_for_completion(task.task_id)
        return result.video_url