可当 API 文档

OpenAI · 生成

创建兼容视频任务，再通过 /v1/videos/{id} 查询结果。

POST/v1/videos

OpenAI 兼容异步任务

查看候选模型 ID
模型 ID厂商类型说明上下文窗口最大输出 Token媒体规格价格状态
sora-2
OpenAI视频Sora 2 是我们新的强大媒体生成模型，可生成带有同步音频的视频。 它可以从自然语言或图像创建细节丰富的动态剪辑。—————

按公开标签归类，并补充模型广场资料；— 表示暂无明确资料。左右滑动查看完整表格。

进阶用法与完整流程 · 参数与透传

Headers请求头

左右滑动查看完整参数表

请求头
参数名类型必填说明默认值 / 示例
Authorizationstring必填

平台 API Key，使用 Bearer 认证。

示例
Bearer YOUR_API_KEY
Content-Typestring必填

由客户端自动生成 multipart/form-data 并附带 boundary；表中类型仅作说明，实际请求不要手工设置缺少 boundary 的请求头。

示例
multipart/form-data

Request Body请求体

multipart/form-data

左右滑动查看完整参数表

请求体
参数名类型必填说明默认值 / 示例
modelstring必填

完整视频模型 ID，本页参考协议针对 sora-2。

示例
sora-2
promptstring必填

视频内容提示词。

—
secondsstring可选

请求的视频秒数，使用字符串。sora-2 参考协议支持 4、8、12，最短为 4 秒。

可选值4812
示例
4
sizestring可选

输出像素尺寸；sora-2 参考协议的 720p 横向为 1280x720，纵向为 720x1280。

示例
1280x720

使用说明

  • 本页描述纯文本提交：model、prompt、seconds、size 四个 multipart/form-data 表单字段；请求头和 boundary 由客户端自动生成。
  • 创建成功后保存任务 ID，并 GET /v1/videos/{id} 查询；状态 completed 后用 GET /v1/videos/{id}/content 获取媒体字节流，不要将文件流当作 JSON 解析。
  • 模型可用范围以平台控制台为准；其他视频模型应使用各自的接口规格。
  • 创建超时不能证明服务端未创建任务；先检查已有任务 ID 或提交记录，避免重复生成。

创建 /v1/videos 任务

POST /v1/videos 使用 multipart/form-data；纯文本字段也必须按表单编码。model、prompt、seconds、size 的取值按此协议和模型配置填写。示例模型仍为 MODEL_ID，账户需开放相应接口。

展开 cURL：创建一次视频任务
cURL
curl --fail-with-body --max-time 120 --request POST \
  'https://ai.kedang.net/v1/videos' \
  --header "Authorization: Bearer ${KEDANG_API_KEY}" \
  --form-string 'model=MODEL_ID' \
  --form-string 'prompt=一朵白云缓慢飘过蓝天。' \
  --form-string 'seconds=4' \
  --form-string 'size=1280x720'

查询视频状态

创建请求只提交一次，并立即保存响应中的任务 ID。以下示例先等待 3 秒，再按 5、10、30 秒的间隔查询，之后保持 30 秒间隔；轮询预算为 600 秒。这些数值是客户端示例策略，不是平台频率限制或生成时长承诺。超时后保留任务 ID，后续继续查询同一任务。HTTP 错误、失败、取消及未知状态会停止本轮轮询。

GET /v1/videos/{id} 返回 JSON；常见状态为 queued、in_progress、completed 和 failed。只有完成后才下载内容；路径中的 ID 需要 URL 编码。

展开 cURL：按视频 ID 查询一次
cURL
export VIDEO_ID='VIDEO_TASK_ID'
ENCODED_VIDEO_ID="$(python3 -c 'import os; from urllib.parse import quote; print(quote(os.environ["VIDEO_ID"], safe=""))')"

curl --fail-with-body --max-time 30 --request GET \
  "https://ai.kedang.net/v1/videos/${ENCODED_VIDEO_ID}" \
  --header "Authorization: Bearer ${KEDANG_API_KEY}"
展开 Python：按视频 ID 查询一次
Python
# 安装依赖：python -m pip install requests
import json
import os
import tempfile
import time
from pathlib import Path
from urllib.parse import quote, urlsplit
import requests

BASE_URL = "https://ai.kedang.net"

def main():
    video_id = os.environ["VIDEO_ID"]  # 已保存的创建响应 id
    if not video_id:
        raise ValueError("VIDEO_ID 不能为空")
    path = "/v1/videos/" + quote(video_id, safe="")
    response = requests.get(BASE_URL + path,
        headers={"Authorization": "Bearer " + os.environ["KEDANG_API_KEY"]}, timeout=30)
    response.raise_for_status()
    result = response.json()
    if result.get("error"):
        raise RuntimeError(result["error"])
    print(json.dumps(result, ensure_ascii=False, indent=2))

if __name__ == "__main__":
    main()
展开 Python：创建、轮询与下载完整流程
Python
# 安装依赖：python -m pip install requests
import json
import os
import tempfile
import time
from pathlib import Path
from urllib.parse import quote, urlsplit
import requests

BASE_URL = "https://ai.kedang.net"

def normalize_task(payload):
    """读取任务状态；HTTP 200 或 code=success 仅表示查询成功。"""
    if not isinstance(payload, dict):
        raise ValueError("任务响应必须为 JSON 对象")
    if payload.get("error"):
        return {"state": "failed", "url": "", "reason": str(payload["error"])}
    code = payload.get("code")
    if code is not None and str(code).lower() not in ("0", "200", "success", "ok"):
        raise RuntimeError(payload.get("message") or str(code))
    nodes, layer = [payload], [payload]
    for _ in range(5):
        children = []
        for node in layer:
            for key in ("data", "output", "task", "content"):
                if isinstance(node.get(key), dict):
                    children.append(node[key])
        nodes.extend(children)
        layer = children
    # 外层任务状态优先；未提供时再读取嵌套状态。
    raw = next((str(node.get("status") or node.get("task_status")).lower()
                for node in nodes if node.get("status") or node.get("task_status")), "")
    states = {
        "queued": "queued", "pending": "queued", "not_start": "queued", "submitted": "queued",
        "in_progress": "running", "running": "running", "processing": "running",
        "success": "succeeded", "succeeded": "succeeded", "completed": "succeeded",
        "failure": "failed", "failed": "failed", "fail": "failed", "error": "failed", "expired": "failed",
        "cancelled": "cancelled", "canceled": "cancelled",
    }
    url = next((node[key] for node in nodes for key in ("result_url", "video_url", "url")
                if isinstance(node.get(key), str) and node[key]), "")
    reason = next((str(node[key]) for node in nodes for key in ("fail_reason", "error", "message")
                   if node.get(key)), "")
    return {"state": states.get(raw, "unknown"), "url": url, "reason": reason}


def poll_task(task_id, fetch, sleep, monotonic, total_timeout=600):
    """3 / 5 / 10 / 30 秒间隔与 600 秒预算均为客户端示例策略。"""
    if not isinstance(task_id, str) or not task_id:
        raise ValueError("任务 ID 不能为空")
    if total_timeout <= 0:
        raise ValueError("轮询预算必须大于零")
    encoded_id = quote(task_id, safe="")
    deadline = monotonic() + total_timeout
    delays, attempt = (3, 5, 10, 30), 0
    while True:
        remaining = deadline - monotonic()
        if remaining <= 0:
            raise TimeoutError("轮询超时，保留此 ID 后续查询：" + task_id)
        sleep(min(delays[min(attempt, len(delays) - 1)], remaining))
        remaining = deadline - monotonic()
        if remaining <= 0:
            raise TimeoutError("轮询超时，保留此 ID 后续查询：" + task_id)
        payload = fetch(encoded_id, min(30, remaining))
        if monotonic() >= deadline:
            raise TimeoutError("轮询超时，保留此 ID 后续查询：" + task_id)
        task = normalize_task(payload)
        if task["state"] == "succeeded":
            return payload
        if task["state"] in ("failed", "cancelled"):
            raise RuntimeError("任务 " + task_id + "：" + task["state"] + " " + task["reason"])
        if task["state"] not in ("queued", "running"):
            raise RuntimeError("未知任务状态，请按任务 ID 检查响应：" + task_id)
        attempt += 1

def download_video(url, target, headers=None):
    parsed = urlsplit(url)
    if parsed.scheme != "https" or not parsed.netloc:
        raise ValueError("下载地址必须为有效 HTTPS URL")
    target = Path(target)
    temporary = None
    try:
        with requests.get(url, headers=headers or {}, stream=True, timeout=(10, 120)) as response:
            response.raise_for_status()
            mime = response.headers.get("Content-Type", "").split(";", 1)[0].lower()
            if mime and not (mime.startswith("video/") or mime == "application/octet-stream"):
                raise ValueError("下载响应不是视频或二进制文件")
            with tempfile.NamedTemporaryFile(dir=target.parent, delete=False) as file:
                temporary = Path(file.name)
                size = 0
                for chunk in response.iter_content(chunk_size=1024 * 1024):
                    if chunk:
                        file.write(chunk)
                        size += len(chunk)
            if size == 0:
                raise ValueError("下载内容为空")
        temporary.replace(target)
    finally:
        if temporary is not None:
            temporary.unlink(missing_ok=True)
    return target

def main():
    headers = {"Authorization": "Bearer " + os.environ["KEDANG_API_KEY"]}
    # 这里是 /v1/videos 的参考协议；MODEL_ID 必须在账户中开放此接口。
    fields = {"model": "MODEL_ID", "prompt": "一朵白云缓慢飘过蓝天。",
              "seconds": "4", "size": "1280x720"}
    # (None, value) 使纯文本字段也按 multipart/form-data 编码。
    response = requests.post(BASE_URL + "/v1/videos", headers=headers,
        files={key: (None, value) for key, value in fields.items()}, timeout=(10, 120))
    response.raise_for_status()
    created = response.json()
    if created.get("error"):
        raise RuntimeError(created["error"])
    video_id = created.get("id")
    if not isinstance(video_id, str) or not video_id:
        raise RuntimeError("响应没有视频 ID；先核对记录，不自动重新创建。")
    Path("video-task-id.txt").write_text(video_id, encoding="utf-8")
    print("视频 ID：", video_id)

    def fetch(encoded_id, timeout):
        result = requests.get(BASE_URL + "/v1/videos/" + encoded_id,
                              headers=headers, timeout=timeout)
        result.raise_for_status()
        return result.json()

    poll_task(video_id, fetch, time.sleep, time.monotonic, total_timeout=600)
    path = "/v1/videos/" + quote(video_id, safe="") + "/content"
    print(download_video(BASE_URL + path, "video.mp4", headers))

if __name__ == "__main__":
    main()

下载视频内容

GET /v1/videos/{id}/content 返回媒体字节流；请保存为文件，不能调用 response.json()。下载使用平台认证。示例先写临时文件，成功后替换 video.mp4。

展开 cURL：保存已完成的视频
cURL
set -eu
export VIDEO_ID='VIDEO_TASK_ID'
ENCODED_VIDEO_ID="$(python3 -c 'import os; from urllib.parse import quote; print(quote(os.environ["VIDEO_ID"], safe=""))')"

TEMP_VIDEO="$(mktemp ./video.mp4.XXXXXX)"
trap 'rm -f "$TEMP_VIDEO"' EXIT
curl --fail-with-body --max-time 120 --request GET \
  "https://ai.kedang.net/v1/videos/${ENCODED_VIDEO_ID}/content" \
  --header "Authorization: Bearer ${KEDANG_API_KEY}" \
  --output "$TEMP_VIDEO"
mv "$TEMP_VIDEO" video.mp4
trap - EXIT
展开 Python：流式下载已完成的视频
Python
# 安装依赖：python -m pip install requests
import json
import os
import tempfile
import time
from pathlib import Path
from urllib.parse import quote, urlsplit
import requests

BASE_URL = "https://ai.kedang.net"

def download_video(url, target, headers=None):
    parsed = urlsplit(url)
    if parsed.scheme != "https" or not parsed.netloc:
        raise ValueError("下载地址必须为有效 HTTPS URL")
    target = Path(target)
    temporary = None
    try:
        with requests.get(url, headers=headers or {}, stream=True, timeout=(10, 120)) as response:
            response.raise_for_status()
            mime = response.headers.get("Content-Type", "").split(";", 1)[0].lower()
            if mime and not (mime.startswith("video/") or mime == "application/octet-stream"):
                raise ValueError("下载响应不是视频或二进制文件")
            with tempfile.NamedTemporaryFile(dir=target.parent, delete=False) as file:
                temporary = Path(file.name)
                size = 0
                for chunk in response.iter_content(chunk_size=1024 * 1024):
                    if chunk:
                        file.write(chunk)
                        size += len(chunk)
            if size == 0:
                raise ValueError("下载内容为空")
        temporary.replace(target)
    finally:
        if temporary is not None:
            temporary.unlink(missing_ok=True)
    return target

def main():
    video_id = os.environ["VIDEO_ID"]  # 先查询并确认任务 completed
    if not video_id:
        raise ValueError("VIDEO_ID 不能为空")
    path = "/v1/videos/" + quote(video_id, safe="") + "/content"
    headers = {"Authorization": "Bearer " + os.environ["KEDANG_API_KEY"]}
    print(download_video(BASE_URL + path, "video.mp4", headers))

if __name__ == "__main__":
    main()