用AI批量生成产品视频！Python+Google Veo 3.1 API让电商转化率飙升
今天跟大家分享一个超实用的电商运营技巧：如何用Python和Google Veo 3.1 AI，把枯燥的产品图片批量变成生动的营销视频。用户原创内容（UGC）现在特别火，对销售的拉动效果非常明显，有了这个工具，你也能轻松制作大量高质量视频内容！先看看最终效果：
[*]你只需要准备：产品图片 + 文字描述
[*]AI帮你产出：专业级产品展示视频
[*]最棒的是：一次性处理几十个产品，全自动生成对应视频
准备工作
运行环境要求
[*]Python 3.7 或更新版本
[*]requests 库
[*]有效的 Veo API 访问权限
获取API密钥
[*]访问 DEFAPI平台 注册账号
[*]在控制台获取你的专属 API 密钥
[*]安全提醒：千万别把密钥直接硬编码在代码里，用环境变量管理更稳妥
安装必要的包
pip install requests python-dotenvAPI接口概览
视频生成接口
[*]请求地址: POST https://api.defapi.org/api/google/veo/generate
[*]认证方式: Authorization: Bearer YOUR_API_KEY
[*]核心参数:
[*]model: 选择合适的 Veo 模型版本
[*]prompt: 视频描述文案
[*]images: 产品图片的 URL 集合
[*]aspect_ratio: 视频画幅比例 (16:9 或 9:16)
[*]callback_url: 结果回调地址（选填）
任务状态查询
[*]请求地址: GET https://api.defapi.org/api/task/query
[*]参数: task_id - 任务唯一标识
代码实战
第一步：基础配置和导入
import os
import requests
import time
import json
from typing import List, Dict, Optional
from dotenv import load_dotenv
# 加载环境配置
load_dotenv()
class VeoVideoGenerator:
def __init__(self):
# 从环境变量安全读取API密钥
self.api_key = os.getenv('VEO_API_KEY')
if not self.api_key:
raise ValueError("记得先配置 VEO_API_KEY 环境变量哦")
self.base_url = "https://api.defapi.org"
self.headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}第二步：核心视频生成功能
def generate_video(self,
product_images: List,
prompt: str,
model: str = "google/veo3.1-components",
aspect_ratio: str = "9:16",
callback_url: Optional = None) -> str:
"""
启动视频生成任务
参数说明:
product_images: 产品图片URL列表
prompt: 视频描述提示语
model: 指定Veo模型版本
aspect_ratio: 视频比例
callback_url: 回调URL（选填）
返回:
task_id: 任务ID，用于后续查询进度
"""
# 组装请求数据
payload = {
"model": model,
"prompt": prompt,
"images": product_images,
"aspect_ratio": aspect_ratio
}
# 可选的回调地址
if callback_url:
payload["callback_url"] = callback_url
try:
# 发起视频生成请求
response = requests.post(
f"{self.base_url}/api/google/veo/generate",
headers=self.headers,
json=payload,
timeout=30
)
# 检查响应状态
if response.status_code == 200:
result = response.json()
if result.get('code') == 0:
task_id = result['data']['task_id']
print(f"✅ 视频任务已创建: {task_id}")
return task_id
else:
raise Exception(f"API返回异常: {result.get('message')}")
else:
# 处理错误响应
error_msg = response.json().get('message', '未知错误')
raise Exception(f"请求失败 (HTTP {response.status_code}): {error_msg}")
except requests.exceptions.RequestException as e:
raise Exception(f"网络请求异常: {str(e)}")第三步：任务进度跟踪
def query_task_status(self, task_id: str) -> Dict: """ 查询任务当前状态 参数: task_id: 任务ID 返回: 任务状态详情 """ try: response = requests.get( f"{self.base_url}/api/task/query", headers=self.headers, params={"task_id": task_id}, timeout=10 ) if response.status_code == 200: return response.json() else: raise Exception(f"状态查询失败: HTTP {response.status_code}") except requests.exceptions.RequestException as e: raise Exception(f"查询时网络异常: {str(e)}")def wait_for_task_completion(self, task_id: str, max_wait: int = 600) -> str: """ 等待任务完成并返回视频URL 参数: task_id: 任务ID max_wait: 最长等待时间（秒） 返回: 视频访问URL """ start_time = time.time() while time.time() - start_time < max_wait: # 查询当前任务状态 status_result = self.query_task_status(task_id) if status_result.get('code') == 0: task_data = status_result['data'] status = task_data['status'] print(f"当前任务状态: {status}") if status == "success": # 任务成功完成 video_url = task_data['result']['video_url'] print(f"
来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！ 很好很强大我过来先占个楼 待编辑 感谢分享，下载保存了，貌似很强大 热心回复！
页:
[1]