D-ID API 视频生成完成状态轮询与结果获取完整教程-Python教程-PHP中文网

D-ID API 视频生成完成状态轮询与结果获取完整教程

D-ID API 采用异步处理机制，POST 创建视频任务后需通过轮询 GET 接口检测 status 字段，直至其变为 "done" 才能安全获取 result_url；直接立即请求将返回 pending_url 和中间状态。

d-id api 采用异步处理机制，post 创建视频任务后需通过轮询 get 接口检测 `status` 字段，直至其变为 `"done"` 才能安全获取 `result_url`；直接立即请求将返回 `pending_url` 和中间状态。

D-ID API 的视频生成是典型的异步长任务流程：调用 /talks POST 接口仅提交任务并返回 201 Created 及临时 talk_id，实际渲染需后台处理数秒至数十秒。此时若立即 GET 查询，响应中 status 通常为 "created" → "started" → "processing" → "done"，而 video_url（即 result_url）仅在 status === "done" 时才被写入响应体。若过早读取，会得到 pending_url（S3 预签名路径，不可直接播放）或缺失 result_url 字段。

✅ 推荐实现方式：带退避的轮询（Polling with Backoff）
以下为生产就绪的 Flask 示例，已优化健壮性：

import os
import time
import json
import requests
from flask import Flask

app = Flask(__name__)

@app.route('/video')
def generate_video():
    bearer_token = os.getenv('BEARER_TOKEN')
    if not bearer_token:
        return "❌ Missing BEARER_TOKEN in environment", 500

    # Step 1: Submit talk generation task
    url = "https://api.d-id.com/talks"
    payload = {
        "source_url": "https://i.imghippo.com/files/wHD7943BS.jpg",
        "script": {
            "type": "text",
            "subtitles": False,
            "provider": {"type": "microsoft", "voice_id": "Sara"},
            "input": "Making videos is easy with D-ID"
        },
        "config": {"fluent": False, "pad_audio": "0.0"}
    }
    headers = {
        "accept": "application/json",
        "content-type": "application/json",
        "authorization": f"Bearer {bearer_token}"
    }

    response = requests.post(url, json=payload, headers=headers, timeout=30)
    if response.status_code != 201:
        return f"❌ Talk creation failed: {response.status_code} - {response.text}", 500

    talk_id = response.json()["id"]
    talk_url = f"https://api.d-id.com/talks/{talk_id}"

    # Step 2: Poll until status == "done" (with max retries & exponential backoff)
    max_retries = 30
    base_delay = 1.0  # seconds
    for attempt in range(max_retries):
        try:
            res = requests.get(talk_url, headers={"accept": "application/json", "authorization": f"Bearer {bearer_token}"}, timeout=15)
            res.raise_for_status()
            data = res.json()

            status = data.get("status")
            print(f"[Attempt {attempt+1}] Status: {status}")

            if status == "done":
                video_url = data.get("result_url")
                if not video_url:
                    return "❌ 'result_url' missing in 'done' response", 500
                return {"video_url": video_url, "talk_id": talk_id}

            elif status in ["created", "started", "processing"]:
                # Exponential backoff: 1s → 2s → 4s → ...
                time.sleep(base_delay * (2 ** attempt))
                continue

            else:
                return f"❌ Unexpected status: {status}", 500

        except requests.RequestException as e:
            print(f"Request failed on attempt {attempt+1}: {e}")
            if attempt == max_retries - 1:
                return f"❌ All polling attempts failed: {e}", 500
            time.sleep(base_delay * (2 ** attempt))

    return "❌ Video generation timed out after maximum retries", 500

⚠️ 关键注意事项：

禁止无限制轮询：必须设置 max_retries（建议 ≤30）和 timeout，避免阻塞主线程或耗尽 API 配额；
使用指数退避（Exponential Backoff）：初始延迟 1 秒，每次失败翻倍（如 1s → 2s → 4s），减轻服务端压力；
校验 result_url 而非 pending_url：pending_url 是内部 S3 路径，仅用于调试，不可用于前端播放；
错误处理必须覆盖：网络超时、HTTP 错误（4xx/5xx）、JSON 解析失败、字段缺失等场景；
生产环境建议改用 Webhook：D-ID 支持配置 webhook_url 在 /talks POST 中，服务端会在 status === "done" 时主动推送结果，彻底规避轮询。