Kling 系列

export AIO_API_KEY="sk-your-api-key"
export AIO_BASE_URL="https://api.aiearth.dev/v1"

curl "$AIO_BASE_URL/videos/image2video" \
  -H "Authorization: Bearer $AIO_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model_name": "kling-v3",
    "image": "https://example.com/source-image.jpg",
    "prompt": "让画面中的产品在柔和灯光下缓慢旋转，镜头轻微推进，突出金属质感和高级商业广告风格。",
    "mode": "std",
    "duration": "5",
    "sound": "on"
  }'

curl "$AIO_BASE_URL/videos/image2video/{task_id}" \
  -H "Authorization: Bearer $AIO_API_KEY"

curl "$AIO_BASE_URL/videos/omni-video" \
  -H "Authorization: Bearer $AIO_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model_name": "kling-v3-omni",
    "prompt": "让 <<<image_1>>> 中的人物走到 <<<image_2>>> 的展台前，镜头从中景缓慢推进，整体保持科技发布会风格。",
    "image_list": [
      {
        "image_url": "https://example.com/person.jpg"
      },
      {
        "image_url": "https://example.com/stage.jpg"
      }
    ],
    "duration": "5",
    "mode": "std",
    "aspect_ratio": "16:9",
    "callback_url": "",
    "external_task_id": ""
  }'

curl "$AIO_BASE_URL/videos/omni-video/{task_id}" \
  -H "Authorization: Bearer $AIO_API_KEY"

import os
import time
import requests

api_key = os.environ["AIO_API_KEY"]
base_url = os.environ.get("AIO_BASE_URL", "https://api.aiearth.dev/v1")
task_id = "task_xxx"

headers = {"Authorization": f"Bearer {api_key}"}

while True:
    response = requests.get(
        f"{base_url}/videos/image2video/{task_id}",
        headers=headers,
        timeout=60,
    )
    response.raise_for_status()
    data = response.json()
    status = (
        data.get("data", {}).get("task_status")
        or data.get("output", {}).get("task_status")
        or data.get("status")
    )
    print("status:", status)

    if status in {"succeed", "succeeded", "completed"}:
        print(data)
        break
    if status in {"failed", "failure", "canceled", "cancelled"}:
        raise RuntimeError(data)

    time.sleep(5)