Kie.ai Provider

Универсальный клиент для Kie.ai API (генерация музыки, видео и др.).

Возможности

• Универсальный API для всех Kie.ai сервисов
• Асинхронное ожидание результатов
• Автоматический retry
• Загрузка файлов

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

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

from kit.providers.kie import KieClient

client = KieClient(api_key="your_kie_api_key")

# Генерация музыки
task = await client.create_task(
    endpoint="music/generate",
    params={
        "prompt": "Epic orchestral trailer music",
        "duration": 30,
        "style": "cinematic"
    }
)

# Ожидание результата
result = await client.wait_for_result(task.task_id)
print(f"Music URL: {result.output_url}")

Доступные endpoints

# Генерация музыки
task = await client.create_task(
    endpoint="music/generate",
    params={"prompt": "...", "duration": 60}
)

# Генерация звуковых эффектов
task = await client.create_task(
    endpoint="sfx/generate",
    params={"prompt": "explosion sound effect"}
)

# Lip sync
task = await client.create_task(
    endpoint="video/lipsync",
    params={
        "video_url": "https://...",
        "audio_url": "https://..."
    }
)

# Voice cloning
task = await client.create_task(
    endpoint="voice/clone",
    params={
        "audio_url": "https://sample.mp3",
        "text": "Text to synthesize"
    }
)

Загрузка файлов

# Загрузка локального файла
file_url = await client.upload_file("./audio.mp3")

# Использование в задаче
task = await client.create_task(
    endpoint="video/lipsync",
    params={
        "video_url": video_url,
        "audio_url": file_url
    }
)

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

# Создание задачи
task = await client.create_task(...)

# Проверка статуса
status = await client.get_status(task.task_id)
print(f"Status: {status.state}")  # pending, processing, completed, failed
print(f"Progress: {status.progress}%")

# Ожидание с callback
async def on_progress(progress: float):
    print(f"Progress: {progress}%")

result = await client.wait_for_result(
    task.task_id,
    on_progress=on_progress
)

Batch операции

# Множественные задачи
tasks = []
for prompt in prompts:
    task = await client.create_task(
        endpoint="music/generate",
        params={"prompt": prompt, "duration": 30}
    )
    tasks.append(task.task_id)

# Ожидание всех
results = await client.wait_for_all(tasks)

API Reference

KieClient

class KieClient:
    def __init__(
        self,
        api_key: str,
        base_url: str = "https://api.kie.ai",
        timeout: float = 30.0,
        max_retries: int = 3
    )

    async def create_task(
        self,
        endpoint: str,
        params: dict
    ) -> KieTask

    async def get_status(self, task_id: str) -> KieStatus

    async def wait_for_result(
        self,
        task_id: str,
        timeout: float = 600,
        poll_interval: float = 5,
        on_progress: Callable = None
    ) -> KieResult

    async def wait_for_all(
        self,
        task_ids: List[str],
        timeout: float = 600
    ) -> List[KieResult]

    async def upload_file(
        self,
        file_path: str
    ) -> str  # Returns URL

    async def close(self) -> None

Data Classes

@dataclass
class KieTask:
    task_id: str
    endpoint: str
    status: str
    created_at: datetime

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

@dataclass
class KieResult:
    task_id: str
    output_url: str
    output_type: str  # audio, video, image
    metadata: dict = None

Примеры из production

Music Video Generator — генерация музыки

class MusicGenerator:
    def __init__(self):
        self.kie = KieClient(api_key=settings.kie_api_key)

    async def generate_background_music(
        self,
        mood: str,
        duration: int,
        style: str = "cinematic"
    ) -> str:
        """Генерация фоновой музыки для видео."""

        mood_prompts = {
            "happy": "upbeat cheerful positive energy",
            "sad": "melancholic emotional piano",
            "epic": "epic orchestral dramatic powerful",
            "calm": "peaceful ambient relaxing"
        }

        prompt = f"{mood_prompts.get(mood, mood)} {style} music"

        task = await self.kie.create_task(
            endpoint="music/generate",
            params={
                "prompt": prompt,
                "duration": duration,
                "style": style
            }
        )

        result = await self.kie.wait_for_result(task.task_id)
        return result.output_url

    async def generate_with_reference(
        self,
        reference_audio: str,
        duration: int
    ) -> str:
        """Генерация музыки похожей на референс."""

        # Загружаем референс
        ref_url = await self.kie.upload_file(reference_audio)

        task = await self.kie.create_task(
            endpoint="music/style-transfer",
            params={
                "reference_url": ref_url,
                "duration": duration
            }
        )

        result = await self.kie.wait_for_result(task.task_id)
        return result.output_url

Lip Sync для видео

class LipSyncProcessor:
    def __init__(self):
        self.kie = KieClient(api_key=settings.kie_api_key)

    async def sync_video(
        self,
        video_path: str,
        audio_path: str
    ) -> str:
        """Синхронизация губ в видео с аудио."""

        # Загружаем файлы
        video_url = await self.kie.upload_file(video_path)
        audio_url = await self.kie.upload_file(audio_path)

        task = await self.kie.create_task(
            endpoint="video/lipsync",
            params={
                "video_url": video_url,
                "audio_url": audio_url,
                "quality": "high"
            }
        )

        def progress_callback(progress):
            logger.info(f"Lip sync progress: {progress}%")

        result = await self.kie.wait_for_result(
            task.task_id,
            on_progress=progress_callback
        )

        return result.output_url

Voice Cloning

class VoiceCloner:
    def __init__(self):
        self.kie = KieClient(api_key=settings.kie_api_key)

    async def clone_and_speak(
        self,
        voice_sample: str,
        text: str
    ) -> str:
        """Клонирование голоса и синтез речи."""

        sample_url = await self.kie.upload_file(voice_sample)

        task = await self.kie.create_task(
            endpoint="voice/clone",
            params={
                "sample_url": sample_url,
                "text": text,
                "language": "ru"
            }
        )

        result = await self.kie.wait_for_result(task.task_id)
        return result.output_url