用AI批量生成产品视频！Python+Google Veo 3.1 API让电商转化率飙升

今天跟大家分享一个超实用的电商运营技巧：如何用Python和Google Veo 3.1 AI，把枯燥的产品图片批量变成生动的营销视频。用户原创内容（UGC）现在特别火，对销售的拉动效果非常明显，有了这个工具，你也能轻松制作大量高质量视频内容！
先看看最终效果：

• 你只需要准备：产品图片 + 文字描述
  
• AI帮你产出：专业级产品展示视频
  
• 最棒的是：一次性处理几十个产品，全自动生成对应视频
  

准备工作

运行环境要求

• Python 3.7 或更新版本
  
• requests 库
  
• 有效的 Veo API 访问权限
  

获取API密钥

• 访问 DEFAPI平台 注册账号
  
• 在控制台获取你的专属 API 密钥
  
• 安全提醒：千万别把密钥直接硬编码在代码里，用环境变量管理更稳妥
  

安装必要的包

1. 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 - 任务唯一标识
  

代码实战

第一步：基础配置和导入

1. import os
2. import requests
3. import time
4. import json
5. from typing import List, Dict, Optional
6. from dotenv import load_dotenv
8. # 加载环境配置
9. load_dotenv()
11. class VeoVideoGenerator:
12.     def __init__(self):
13.         # 从环境变量安全读取API密钥
14.         self.api_key = os.getenv('VEO_API_KEY')
15.         if not self.api_key:
16.             raise ValueError("记得先配置 VEO_API_KEY 环境变量哦")
17.         
18.         self.base_url = "https://api.defapi.org"
19.         self.headers = {
20.             "Authorization": f"Bearer {self.api_key}",
21.             "Content-Type": "application/json"
22.         }

复制代码

第二步：核心视频生成功能

1. def generate_video(self,
2.                   product_images: List[str],
3.                   prompt: str,
4.                   model: str = "google/veo3.1-components",
5.                   aspect_ratio: str = "9:16",
6.                   callback_url: Optional[str] = None) -> str:
7.     """
8.     启动视频生成任务
9.    
10.     参数说明:
11.         product_images: 产品图片URL列表
12.         prompt: 视频描述提示语
13.         model: 指定Veo模型版本
14.         aspect_ratio: 视频比例
15.         callback_url: 回调URL（选填）
16.    
17.     返回:
18.         task_id: 任务ID，用于后续查询进度
19.     """
20.    
21.     # 组装请求数据
22.     payload = {
23.         "model": model,
24.         "prompt": prompt,
25.         "images": product_images,
26.         "aspect_ratio": aspect_ratio
27.     }
28.    
29.     # 可选的回调地址
30.     if callback_url:
31.         payload["callback_url"] = callback_url
32.    
33.     try:
34.         # 发起视频生成请求
35.         response = requests.post(
36.             f"{self.base_url}/api/google/veo/generate",
37.             headers=self.headers,
38.             json=payload,
39.             timeout=30
40.         )
41.         
42.         # 检查响应状态
43.         if response.status_code == 200:
44.             result = response.json()
45.             if result.get('code') == 0:
46.                 task_id = result['data']['task_id']
47.                 print(f"✅ 视频任务已创建: {task_id}")
48.                 return task_id
49.             else:
50.                 raise Exception(f"API返回异常: {result.get('message')}")
51.         else:
52.             # 处理错误响应
53.             error_msg = response.json().get('message', '未知错误')
54.             raise Exception(f"请求失败 (HTTP {response.status_code}): {error_msg}")
55.             
56.     except requests.exceptions.RequestException as e:
57.         raise Exception(f"网络请求异常: {str(e)}")

复制代码

第三步：任务进度跟踪

[code]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"
来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！