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

腾讯云VOD AIGC视频生成工具

一个功能完整的腾讯云VOD AIGC视频生成工具库，支持轮询模式和回调模式两种获取结果方式。

目录结构

test/vod/ ├── tencent_aigc_video.py # 核心库：API封装、任务管理 ├── config.py # 配置文件：从项目config.yaml读取凭证 ├── examples.py # 使用示例：各种生成场景 ├── callback_demo.py # 回调演示：带SessionContext的任务创建 ├── setup_callback.py # 回调配置：设置/查询回调URL └── README.md # 本文档 

快速开始

1. 配置凭证

确保项目根目录的 config.yaml 包含腾讯云VOD凭证：

secret_id:tencent_vod:"AKIDxxxxxxxxxxxxxxxxxxxxxxxxxxxx"secret_key:tencent_vod:"xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"

或设置环境变量：

exportTENCENTCLOUD_SECRET_ID="AKIDxxxxxxxxxxxxxxxxxxxxxxxxxxxx"exportTENCENTCLOUD_SECRET_KEY="xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"exportTENCENTCLOUD_SUB_APP_ID="1234567890"

2. 验证配置

cd /opt/xj_ai_suanfa/AIGC_INTERFACE_HSC/test/vod python config.py 

输出示例：

腾讯云VOD配置状态: secret_id: AKIDxxxx...xxxx secret_key: ******************** sub_app_id: 1234567890 状态: ✓ 配置完整 

方式一：轮询模式（同步等待）

适用场景：单任务测试、简单脚本、需要立即获取结果

基本用法

from tencent_aigc_video import AIGCVideoGenerator, AIGCModel # 创建生成器 generator = AIGCVideoGenerator()# 生成视频（同步等待，默认超时600秒） result = generator.generate_video( prompt="一只橘猫在阳光下打哈欠，毛发随风轻轻飘动", model=AIGCModel.GV, model_version="3.1-fast", aspect_ratio="16:9", timeout=600)# 获取结果print(f"任务ID: {result['task_id']}")print(f"状态: {result['status']}")print(f"视频URL: {result['video_urls']}")

高级用法（精细控制）

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}")# 自定义进度回调defprogress_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")asyncdefvod_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视频任务
• 查询任务详情
• 修改事件通知配置
• 事件通知综述