利用Veo3 API批量生成ASMR切水果视频

pip install requests

import requests
import time
import json
import random

class VeoVideoGenerator:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://api.defapi.org"
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }

    def generate_video(self, prompt, image_url=""):
        """生成视频任务"""
        url = f"{self.base_url}/api/google/veo/generate"
        payload = {
            "model": "google/veo3-fast",
            "prompt": prompt,
            "image": image_url
        }

        response = requests.post(url, headers=self.headers, json=payload)
        return response.json()

    def query_task(self, task_id):
        """查询任务状态"""
        url = f"{self.base_url}/api/task/query"
        params = {"task_id": task_id}

        response = requests.get(url, headers=self.headers, params=params)
        return response.json()

    def wait_for_completion(self, task_id, max_wait_time=300):
        """等待任务完成"""
        start_time = time.time()

        while time.time() - start_time < max_wait_time:
            result = self.query_task(task_id)

            if result["code"] == 0:
                status = result["data"]["status"]
                print(f"任务 {task_id} 状态: {status}")

                if status == "success":
                    return result["data"]["result"]["video_url"]
                elif status == "failed":
                    print(f"任务失败: {result['data']['status_reason']['message']}")
                    return None

            time.sleep(10)  # 每10秒检查一次

        print("任务超时")
        return None

class ASMRFruitPrompts:
    def __init__(self):
        # 水果类型
        self.fruits = [
            "苹果", "橙子", "柠檬", "西瓜", "哈密瓜", "猕猴桃",
            "草莓", "蓝莓", "葡萄", "梨子", "桃子", "芒果"
        ]

        # 切割动作描述
        self.cutting_actions = [
            "缓慢地切片", "轻柔地切块", "精细地切丁", "小心地对半切开",
            "优雅地切成薄片", "温柔地切成小块", "仔细地去皮切片"
        ]

        # 环境和氛围
        self.environments = [
            "在安静的厨房里", "在温馨的家中", "在明亮的料理台上",
            "在自然光线下", "在整洁的操作台上", "在舒适的环境中"
        ]

        # 视觉效果
        self.visual_effects = [
            "水果汁液缓慢滴落", "果肉纹理清晰可见", "切面光滑平整",
            "颜色鲜艳诱人", "质感丰富细腻", "画面清新自然"
        ]

        # ASMR特定描述
        self.asmr_elements = [
            "发出清脆的切割声", "产生令人放松的声音", "创造舒缓的氛围",
            "声音清晰悦耳", "节奏缓慢平稳", "动作轻柔优雅"
        ]

    def generate_prompt(self):
        """生成随机的ASMR切水果提示词"""
        fruit = random.choice(self.fruits)
        action = random.choice(self.cutting_actions)
        environment = random.choice(self.environments)
        visual = random.choice(self.visual_effects)
        asmr = random.choice(self.asmr_elements)

        prompt = f"""
        ASMR视频：{environment}，一双手正在{action}{fruit}，{asmr}。
        画面特点：{visual}，镜头聚焦在切割过程，
        整体氛围宁静舒缓，适合放松和冥想。
        高清画质，柔和光线，专业ASMR视频风格。
        """.strip().replace('\n        ', ' ')

        return prompt, fruit

def batch_generate_asmr_videos(api_key, video_count=5):
    """批量生成ASMR切水果视频"""

    generator = VeoVideoGenerator(api_key)
    prompt_maker = ASMRFruitPrompts()

    results = []

    print(f"开始批量生成 {video_count} 个ASMR切水果视频...")

    for i in range(video_count):
        print(f"\n=== 生成第 {i+1} 个视频 ===")

        # 生成提示词
        prompt, fruit = prompt_maker.generate_prompt()
        print(f"水果类型: {fruit}")
        print(f"提示词: {prompt}")

        # 提交生成任务
        response = generator.generate_video(prompt)

        if response["code"] == 0:
            task_id = response["data"]["task_id"]
            print(f"任务已提交，ID: {task_id}")

            # 等待完成
            video_url = generator.wait_for_completion(task_id)

            if video_url:
                result = {
                    "index": i + 1,
                    "fruit": fruit,
                    "prompt": prompt,
                    "task_id": task_id,
                    "video_url": video_url,
                    "status": "success"
                }
                print(f"✅ 视频生成成功: {video_url}")
            else:
                result = {
                    "index": i + 1,
                    "fruit": fruit,
                    "prompt": prompt,
                    "task_id": task_id,
                    "video_url": None,
                    "status": "failed"
                }
                print("❌ 视频生成失败")
        else:
            result = {
                "index": i + 1,
                "fruit": fruit,
                "prompt": prompt,
                "task_id": None,
                "video_url": None,
                "status": "error",
                "error": response.get("message", "未知错误")
            }
            print(f"❌ 提交任务失败: {response.get('message')}")

        results.append(result)

        # 避免请求过于频繁
        if i < video_count - 1:
            print("等待5秒后继续...")
            time.sleep(5)

    return results

def save_results(results, filename="asmr_videos_results.json"):
    """保存生成结果"""
    with open(filename, 'w', encoding='utf-8') as f:
        json.dump(results, f, ensure_ascii=False, indent=2)
    print(f"\n结果已保存到 {filename}")

def print_summary(results):
    """打印生成摘要"""
    total = len(results)
    success = len([r for r in results if r["status"] == "success"])
    failed = total - success

    print(f"\n=== 生成摘要 ===")
    print(f"总数: {total}")
    print(f"成功: {success}")
    print(f"失败: {failed}")
    print(f"成功率: {success/total*100:.1f}%")

    print(f"\n=== 成功的视频 ===")
    for result in results:
        if result["status"] == "success":
            print(f"{result['index']}. {result['fruit']} - {result['video_url']}")

def main():
    # 替换为你的API密钥
    API_KEY = "your-api-key-here"

    try:
        # 批量生成5个ASMR切水果视频
        results = batch_generate_asmr_videos(API_KEY, video_count=5)

        # 保存结果
        save_results(results)

        # 打印摘要
        print_summary(results)

    except Exception as e:
        print(f"程序执行出错: {e}")

if __name__ == "__main__":
    main()