minimax · 生成

# 安装依赖：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"]}
    # 将所选模型的完整创建请求保存到此 JSON 文件，包含它要求的规格与扩展参数。
    body = json.loads(Path("video-request.json").read_text(encoding="utf-8"))
    if not body.get("model") or body["model"] == "MODEL_ID":
        raise ValueError("请在 video-request.json 中填写账户可用的模型与完整请求参数")
    # 创建只提交一次。网络超时时先核对任务记录，避免重复创建。
    created_response = requests.post(BASE_URL + "/v1/video/generations", headers=headers,
                                     json=body, timeout=(10, 120))
    created_response.raise_for_status()
    created = created_response.json()
    if created.get("error"):
        raise RuntimeError(created["error"])
    data = created.get("data") if isinstance(created.get("data"), dict) else {}
    task_id = created.get("task_id") or created.get("id") or data.get("task_id")
    if not isinstance(task_id, str) or not task_id:
        raise RuntimeError("响应没有任务 ID；保留提交记录，不自动重新创建。")
    Path("video-task-id.txt").write_text(task_id, encoding="utf-8")
    print("任务 ID：", task_id)

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

    result = poll_task(task_id, fetch, time.sleep, time.monotonic, total_timeout=600)
    url = normalize_task(result)["url"]
    if not url:
        raise RuntimeError("任务已成功但缺少结果地址；使用保存的任务 ID 查询。")
    # 返回的媒体 URL 可能在独立域名，下载时不发送平台 Authorization。
    print(download_video(url, "video.mp4"))

if __name__ == "__main__":
    main()