腾讯云 VOD AIGC 视频生成工具回调实现

高级用法（精细控制）

from tencent_aigc_video import AIGCVideoManager, AIGCModel, TencentVODConfig

# 创建配置和管理器
config = TencentVODConfig.from_env()
manager = AIGCVideoManager(config)

# 创建任务
task = manager.create_video_task(
    model=AIGCModel.GV,
    model_version="3.1-fast",
    prompt="海上日落，金色阳光洒在波光粼粼的海面上",
    output_config={
        "StorageMode": "Temporary",  # 临时存储，7 天有效
        "AspectRatio": "16:9"
    }
)
print(f"任务已创建：{task.task_id}")

# 自定义进度回调
def progress_callback(t):
    status_map = {
        "WAIT": "等待中",
        "RUN": "生成中",
        "DONE": "完成",
        "FAIL": "失败"
    }
    print(f"  状态：{status_map.get(t.status.value, t.status.value)}")

# 等待完成
completed = manager.wait_for_completion(
    task,
    timeout=600,
    poll_interval=10,  # 每 10 秒查询一次
    progress_callback=progress_callback
)

# 获取结果
if completed.video_urls:
    print(f"视频 URL: {completed.video_urls[0]}")

运行示例

python examples.py  # 选择 1-7 测试不同功能

方式二：回调模式（异步通知）

适用场景：批量任务、生产环境、无需阻塞等待

原理

┌─────────────┐ 1.创建任务 ┌─────────────────┐
│ 你的代码    │ ───────────→ │ 腾讯云 VOD API  │
└─────────────┘ (立即返回) └────────┬────────┘
                                   │ 2.后台生成
                                   ↓
┌─────────────┐ 3.HTTP POST ┌─────────────────┐
│ 你的回调服务│ ←────────────│ 腾讯云回调服务  │
└─────────────┘ (任务完成时) └─────────────────┘

步骤 1：配置回调 URL（一次性）

方法 A：使用工具脚本

# 查询当前配置
python setup_callback.py query

# 设置回调 URL
python setup_callback.py set "https://your-server.com/vod/callback"

方法 B：代码配置

from tencent_aigc_video import TencentVODClient, TencentVODConfig

config = TencentVODConfig.from_env()
client = TencentVODClient(config)

# 设置回调
params = {
    "SubAppId": config.sub_app_id,
    "Mode": "PUSH",
    "NotificationUrl": "https://your-server.com/vod/callback",
    "UploadMediaCompleteEventSwitch": "ON",
    "DeleteMediaCompleteEventSwitch": "ON"
}
client._make_request("ModifyEventConfig", params)

步骤 2：创建任务

from tencent_aigc_video import AIGCVideoManager, AIGCModel, TencentVODConfig
import json
import uuid
from datetime import datetime

config = TencentVODConfig.from_env()
manager = AIGCVideoManager(config)

# 生成业务 ID
business_id = f"order_{uuid.uuid4().hex[:8]}"

# SessionContext：透传业务信息，回调时原样返回
session_context = json.dumps({
    "business_id": business_id,
    "user_id": "user_123",
    "created_at": datetime.now().isoformat()
}, ensure_ascii=False)

# 创建任务（立即返回，不阻塞）
task = manager.create_video_task(
    model=AIGCModel.GV,
    model_version="3.1-fast",
    prompt="一只橘猫在阳光下打哈欠",
    output_config={
        "StorageMode": "Temporary",
        "AspectRatio": "16:9"
    },
    session_id=business_id,      # 去重 ID
    session_context=session_context  # 透传信息
)
print(f"任务已提交：{task.task_id}")
print(f"业务 ID: {business_id}")
# 不需要等待，回调会自动发送

步骤 3：接收回调

回调数据示例

{
  "EventType": "AigcVideoTaskComplete",
  "AigcVideoTaskCompleteEvent": {
    "TaskId": "1234567890-AigcVideoTask-xxxxxxxxxxxx",
    "Status": "FINISH",
    "Progress": 100,
    "Input": {
      "Prompt": "一只橘猫在阳光下打哈欠",
      "ModelName": "GV",
      "ModelVersion": "3.1-fast"
    },
    "Output": {
      "FileInfos": [
        {
          "FileUrl": "http://xxx.vod2.myqcloud.com/.../aigcVideoGenFile.mp4",
          "ExpireTime": "2026-01-13T01:50:26Z"
        }
      ]
    },
    "SessionContext": "{\"business_id\": \"order_abc123\", \"user_id\": \"user_123\"}",
    "SessionId": "order_abc123"
  }
}

回调处理示例（FastAPI）

from fastapi import FastAPI, Request
import json

app = FastAPI()

@app.post("/vod/callback")
async def vod_callback(request: Request):
    data = await request.json()
    event_type = data.get("EventType")
    
    if event_type == "AigcVideoTaskComplete":
        event = data["AigcVideoTaskCompleteEvent"]
        
        # 解析透传的业务信息
        ctx = json.loads(event.get("SessionContext", "{}"))
        business_id = ctx.get("business_id")
        user_id = ctx.get("user_id")
        
        # 获取结果
        status = event["Status"]
        if status == "FINISH":
            video_url = event["Output"]["FileInfos"][0]["FileUrl"]
            # 更新数据库、通知用户等
            print(f"任务完成：{business_id}, 视频：{video_url}")
        else:
            error_msg = event.get("Message", "未知错误")
            print(f"任务失败：{business_id}, 错误：{error_msg}")
    
    return {"code": 0, "message": "success"}

两种方式对比

支持的模型

常见问题

Q: 回调配置后，原来的轮询代码还能用吗？

A: 可以！两种方式互不影响。配置回调后，任务完成时会同时：

• 轮询代码正常返回结果
• 腾讯云发送回调到配置的 URL

Q: SessionContext 是必须的吗？

A: 不是。不传也会收到回调，只是回调里 SessionContext 字段为空。

Q: 视频 URL 有效期多久？

A: StorageMode: Temporary 为 7 天，StorageMode: Persist 为永久（需要开通存储）。

Q: 如何测试回调？

A: 使用 https://webhook.site 获取临时 URL 测试。

API 文档

• 创建 AIGC 视频任务
• 查询任务详情
• 修改事件通知配置
• 事件通知综述