当前位置：首页 > news >正文

调用 Google Veo 3.1 API 批量制作电商产品 UGC 视频

news 2025/10/27 7:27:22

在电商运营里，用户自己制作的视频内容（UGC）特别能促进销售。今天我来给大家演示一下，怎么用 Python 和 Google Veo AI 视频生成 API，把静态的产品照片批量变成各种实用的视频。

最终效果是这样的：

输入：产品图片 + 文字描述
输出：高质量的产品展示视频
批量处理：一次性处理几十个产品，每个都能自动生成对应的视频

开始前的准备

需要什么环境

Python 3.7 或更高版本
requests 库
可用的 Veo API 访问权限

获取 API 密钥

先去 DEFAPI平台注册个账号
在控制台拿到你的 API 密钥
重要提醒：别把密钥直接写在代码里，用环境变量来管理更安全

安装需要的包

pip install requests python-dotenv

API 接口说明

视频生成接口

接口地址: 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 - 任务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[str], prompt: str,model: str = "google/veo3.1-components",aspect_ratio: str = "9:16",callback_url: Optional[str] = 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}# 如果有回调URL就加上if callback_url:payload["callback_url"] = callback_urltry:# 发送视频生成请求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_idelse: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: 任务IDmax_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":# 任务成功了，返回视频URLvideo_url = task_data['result']['video_url']print(f"🎉 视频生成成功啦! URL: {video_url}")return video_urlelif status == "failed":# 任务失败了error_msg = task_data['status_reason'].get('message', '未知错误')raise Exception(f"视频生成失败: {error_msg}")elif status in ["pending", "submitted", "in_progress"]:# 任务还在进行中，等一会儿再查time.sleep(10)else:raise Exception(f"遇到未知的任务状态: {status}")else:raise Exception(f"状态查询API出错了: {status_result.get('message')}")raise Exception("等太久了，任务可能卡住了")

第四步：批量生成电商UGC视频

def batch_generate_ugc_videos(self, products_data: List[Dict]) -> Dict[str, str]:"""批量生成电商UGC视频参数:products_data: 产品数据列表，每个产品包含：- product_id: 产品ID- images: 产品图片URL列表- product_name: 产品名称- video_type: 要生成什么类型的视频 ('unboxing', 'demo', 'lifestyle')返回:字典：{产品ID: 视频URL}"""# 不同类型视频对应的描述模板prompt_templates = {'unboxing': "专业开箱视频，展示{product}的包装、开箱过程和产品亮相，画质高清",'demo': "专业产品演示视频，展示{product}的功能、使用方法和优势，场景真实",'lifestyle': "生活方式视频，展示{product}在日常生活中的使用场景，内容真实吸引人"}results = {}for product in products_data:product_id = product['product_id']product_name = product['product_name']video_type = product['video_type']print(f"\n🚀 开始处理产品: {product_name} ({video_type}类型)")try:# 生成视频描述文字prompt = prompt_templates[video_type].format(product=product_name)# 提交视频生成任务task_id = self.generate_video(product_images=product['images'],prompt=prompt,model="google/veo3.1-components",  # 用支持多图片的模型aspect_ratio="9:16"  # 适合手机看的竖屏比例)# 等着任务完成video_url = self.wait_for_task_completion(task_id)results[product_id] = video_urlprint(f"✅ 产品 {product_name} 的视频做好啦")except Exception as e:print(f"❌ 产品 {product_name} 的视频没做成: {str(e)}")results[product_id] = Nonereturn results

第五步：回调处理（可选，适合异步处理）

from flask import Flask, request, jsonifyapp = Flask(__name__)@app.route('/webhook/veo-callback', methods=['POST'])
def handle_veo_callback():"""处理Veo API的回调通知"""callback_data = request.jsontask_id = callback_data.get('task_id')status = callback_data.get('status')print(f"收到回调通知: 任务 {task_id} 现在状态是 {status}")if status == "success":video_url = callback_data['result']['video_url']consumed = callback_data['consumed']print(f"🎉 视频生成成功啦!")print(f"视频地址: {video_url}")print(f"用了 {consumed} 个点数")# 这里你可以加上自己的业务逻辑，比如更新数据库、发个通知什么的elif status == "failed":error_msg = callback_data['status_reason'].get('message')print(f"❌ 视频生成失败了: {error_msg}")return jsonify({"status": "收到啦"}), 200

使用示例

# 使用方法示例
if __name__ == "__main__":# 初始化视频生成器generator = VeoVideoGenerator()# 准备产品数据products = [{'product_id': '001','product_name': '无线蓝牙耳机','video_type': 'unboxing','images': ['https://example.com/earphone1.jpg','https://example.com/earphone2.jpg']},{'product_id': '002', 'product_name': '智能手表','video_type': 'demo','images': ['https://example.com/watch1.jpg','https://example.com/watch2.jpg']}]# 批量生成视频results = generator.batch_generate_ugc_videos(products)# 打印结果for product_id, video_url in results.items():if video_url:print(f"产品 {product_id} 的视频: {video_url}")else:print(f"产品 {product_id} 的视频生成失败")