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 接口仅触发视频生成任务并立即返回（HTTP 201），不保证视频已就绪。此时资源状态为 "created" 或 "started"，响应中仅包含 pending_url（S3 预签名路径，不可直接访问）和中间字段（如 audio_url、source_url），真正的成品视频 URL（result_url）仅在处理完成后才写入响应体。

因此，客户端必须主动轮询——即周期性发送 GET 请求检查任务状态，直到 status === "done"。这是官方推荐且最通用的实现方式（相比 Webhook，轮询更易调试、无服务端回调依赖）。

以下为生产就绪的轮询实现要点与优化代码：

✅ 关键实践原则

状态驱动而非时间驱动：不依赖固定等待时长（如 time.sleep(5)），而是严格依据 response.json()["status"] 判断；
超时保护：避免无限循环，应设置最大重试次数或总耗时上限（D-ID 典型生成耗时为 5–30 秒）；
错误兜底：捕获网络异常、非 200 响应、缺失字段等边界情况；
使用 f-string 替代 .format()：提升可读性与安全性（避免格式化漏洞）；
移除冗余校验逻辑：len(did_api_key) != None 是错误写法（len() 不接受 None），应改为 if not did_api_key。

✅ 优化后的 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 "❌ Error: BEARER_TOKEN not found 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"❌ POST failed: {response.status_code} {response.text}", 500

    talk_id = response.json().get("id")
    if not talk_id:
        return "❌ Invalid response: missing 'id'", 500

    # Step 2: Poll until status == "done"
    talk_url = f"https://api.d-id.com/talks/{talk_id}"
    max_retries = 60  # ~2 min timeout (30s avg + buffer)
    for attempt in range(max_retries):
        try:
            resp = requests.get(talk_url, headers=headers, timeout=10)
            if resp.status_code != 200:
                return f"❌ GET failed: {resp.status_code} {resp.text}", 500

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

            if status == "done":
                result_url = data.get("result_url")
                if result_url:
                    return {"video_url": result_url}
                else:
                    return "❌ Response missing 'result_url'", 500
            elif status in ["created", "started", "processing"]:
                time.sleep(2)  # Brief pause before next poll
                continue
            else:
                return f"❌ Unexpected status: {status}", 500

        except requests.RequestException as e:
            return f"❌ Network error: {e}", 500

    return "❌ Timeout: Video generation did not complete within allowed retries", 500

⚠️ 注意事项

pending_url 是内部 S3 路径（s3://...），不可直接用于前端 <video></video> 标签；仅 result_url 是公开可访问的 HTTPS 地址；
若返回 status === "failed"，需检查 error 字段（如有）并记录日志；
生产环境建议添加重试退避（如指数退避 time.sleep(2 ** attempt)）以降低 API 压力；
对于高并发场景，可结合 Redis 缓存 talk_id → result_url 映射，避免重复轮询。