PythonAI算法
腾讯云 VOD AIGC 视频生成工具回调实现
腾讯云 VOD AIGC 视频生成工具 一个功能完整的腾讯云 VOD AIGC 视频生成工具库，支持**轮询模式**和**回调模式**两种获取结果方式。 目录结构 快速开始 配置凭证 确保项目根目录的 config.yaml 包含腾讯云 VOD 凭证： 或设置环境变量： 验证配置 输出示例： --- 方式一：轮询模式（同步等待） 适用场景：单任务测试、简单脚本、需要立即获取结果 基本用法 高级用法…
腾讯云 VOD AIGC 视频生成工具 一个功能完整的腾讯云 VOD AIGC 视频生成工具库，支持**轮询模式**和**回调模式**两种获取结果方式。 目录结构 快速开始 配置凭证 确保项目根目录的 config.yaml 包含腾讯云 VOD 凭证： 或设置环境变量： 验证配置 输出示例： --- 方式一：轮询模式（同步等待） 适用场景：单任务测试、简单脚本、需要立即获取结果 基本用法 高级用法…
一个功能完整的腾讯云 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 # 项目文档
确保项目根目录的 config.yaml 包含腾讯云 VOD 凭证：
tencent_vod:
secret_id: "AKIDxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
secret_key: "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
或设置环境变量：
export TENCENTCLOUD_SECRET_ID="AKIDxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
export TENCENTCLOUD_SECRET_KEY="xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
export TENCENTCLOUD_SUB_APP_ID="1234567890"
cd 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}")
# 自定义进度回调
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 ┌─────────────────┐
│ 你的回调服务│ ←────────────│ 腾讯云回调服务 │
└─────────────┘ (任务完成时) └─────────────────┘
# 查询当前配置
python setup_callback.py query
# 设置回调 URL
python setup_callback.py set "https://your-server.com/vod/callback"
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)
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}")
# 不需要等待，回调会自动发送
{
"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"
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"}
| 特性 | 轮询模式 | 回调模式 |
|---|---|---|
| 代码复杂度 | 简单 | 需要回调服务 |
| 获取结果 | 同步阻塞 | 异步通知 |
| API 调用次数 | N 次（轮询） | 0 次 |
| 适用场景 | 单任务/测试 | 批量/生产 |
| 资源占用 | 阻塞线程 | 无阻塞 |
| 额外配置 | 无 | NotificationUrl |
| 模型 | 文生视频 (T2V) | 图生视频 (I2V) | 首尾帧 | 多图输入 |
|---|---|---|---|---|
| GV (Google Veo) | ✓ | ✓ | ✓ | ✓ |
| KLING (可灵) | ✓ | ✓ | ✓ (2.1+) | ✗ |
| HUNYUAN (混元) | ✓ | ✓ | ✗ | ✗ |
| HAILUO (海螺) | ✗ | ✓ | ✗ | ✗ |
| VIDU | ✓ | ✓ | ✓ (q2-pro) | ✓ (1-7 张) |
| JIMENG (即梦) | ✓ | ✓ | ✗ | ✗ |
| OS (OpenAI Sora) | ✓ | ✗ | ✗ | ✗ |
| MINGMOU (明眸) | ✓ | ✗ | ✗ | ✗ |
| SEEDANCE | 部分 | ✓ | ✗ | ✗ |
A: 可以！两种方式互不影响。配置回调后，任务完成时会同时：
A: 不是。不传也会收到回调，只是回调里 SessionContext 字段为空。
A: StorageMode: Temporary 为 7 天，StorageMode: Persist 为永久（需要开通存储）。
A: 使用 https://webhook.site 获取临时 URL 测试。
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
使用加密算法（如AES、TripleDES、Rabbit或RC4）加密和解密文本明文。 在线工具，加密/解密文本在线工具，online
生成新的随机RSA私钥和公钥pem证书。 在线工具，RSA密钥对生成器在线工具，online
基于 Mermaid.js 实时预览流程图、时序图等图表，支持源码编辑与即时渲染。 在线工具，Mermaid 预览与可视化编辑在线工具，online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online