解决 Google Flow 无权访问报错的 3 种方案 – Veo 3.1 API 替代方案完整指南

import requests
import time
import os

class Veo31VideoGenerator:
    """Veo 3.1 视频生成器 - 支持原生音频"""

    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://api.apiyi.com/v1"  # API易 统一接口
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }

    def text_to_video(self, prompt, aspect_ratio="16:9", with_audio=True):
        """
        文生视频 (TEXT_2_VIDEO)

        Args:
            prompt: 视频描述，可包含对话 (使用引号)
            aspect_ratio: "16:9" 横屏 或 "9:16" 竖屏
            with_audio: 是否生成原生音频

        Returns:
            dict: 包含 video_url 和 status 的结果
        """
        payload = {
            "model": "veo-3.1",
            "mode": "TEXT_2_VIDEO",
            "prompt": prompt,
            "aspect_ratio": aspect_ratio,
            "generate_audio": with_audio,
            "duration": 8
        }

        return self._submit_and_wait(payload)

    def first_last_frame_to_video(self, prompt, first_frame_url, last_frame_url):
        """
        首尾帧控制生成 (FIRST_AND_LAST_FRAMES_2_VIDEO)

        Args:
            prompt: 视频过渡描述
            first_frame_url: 第一帧图片 URL
            last_frame_url: 最后一帧图片 URL

        Returns:
            dict: 生成结果
        """
        payload = {
            "model": "veo-3.1",
            "mode": "FIRST_AND_LAST_FRAMES_2_VIDEO",
            "prompt": prompt,
            "first_frame": first_frame_url,
            "last_frame": last_frame_url,
            "duration": 8
        }

        return self._submit_and_wait(payload)

    def reference_to_video(self, prompt, reference_images):
        """
        参考图生成 (REFERENCE_2_VIDEO)

        Args:
            prompt: 视频描述
            reference_images: 参考图 URL 列表 (1-3 张)

        Returns:
            dict: 生成结果
        """
        if len(reference_images) > 3:
            raise ValueError("最多支持 3 张参考图")

        payload = {
            "model": "veo-3.1",
            "mode": "REFERENCE_2_VIDEO",
            "prompt": prompt,
            "reference_images": reference_images,
            "duration": 8
        }

        return self._submit_and_wait(payload)

    def extend_video(self, previous_video_url, extension_prompt):
        """
        场景扩展 - 基于上一个视频的最后一帧继续生成

        Args:
            previous_video_url: 上一个视频的 URL
            extension_prompt: 扩展场景描述

        Returns:
            dict: 新视频结果
        """
        payload = {
            "model": "veo-3.1",
            "mode": "SCENE_EXTENSION",
            "previous_video": previous_video_url,
            "prompt": extension_prompt,
            "duration": 8
        }

        return self._submit_and_wait(payload)

    def _submit_and_wait(self, payload, max_wait=300):
        """提交任务并等待完成"""

        # 提交任务
        response = requests.post(
            f"{self.base_url}/videos/generations",
            headers=self.headers,
            json=payload
        )

        if response.status_code != 200:
            raise Exception(f"提交失败: {response.text}")

        task_id = response.json()["id"]
        print(f"✅ 任务提交成功，ID: {task_id}")

        # 轮询等待
        start_time = time.time()
        while time.time() - start_time < max_wait:
            result = requests.get(
                f"{self.base_url}/videos/generations/{task_id}",
                headers=self.headers
            ).json()

            status = result.get("status")

            if status == "completed":
                print(f"🎬 视频生成完成!")
                return {
                    "status": "success",
                    "video_url": result["video_url"],
                    "duration": result.get("duration", 8),
                    "resolution": result.get("resolution", "1080p")
                }
            elif status == "failed":
                raise Exception(f"生成失败: {result.get('error')}")
            else:
                progress = result.get("progress", 0)
                print(f"⏳ 生成中... {progress}%")

            time.sleep(5)

        raise TimeoutError("生成超时，请稍后查询结果")


# ========== 使用示例 ==========

if __name__ == "__main__":
    # 初始化生成器
    generator = Veo31VideoGenerator(api_key="YOUR_API_KEY")

    # 示例 1: 文生视频 (带对话音频)
    result = generator.text_to_video(
        prompt='''
        一位年轻女性站在咖啡店柜台前，微笑着说："我要一杯拿铁"。
        咖啡师点头回应："好的，稍等"。
        背景是温馨的咖啡店氛围，轻柔的爵士乐。
        ''',
        aspect_ratio="16:9",
        with_audio=True
    )
    print(f"视频地址: {result['video_url']}")

    # 示例 2: 首尾帧控制 (精确构图)
    result = generator.first_last_frame_to_video(
        prompt="镜头从特写逐渐拉远，展示整个城市天际线",
        first_frame_url="https://example.com/closeup.jpg",
        last_frame_url="https://example.com/skyline.jpg"
    )

    # 示例 3: 生成长视频 (多次扩展)
    clips = []

    # 生成第一个片段
    clip1 = generator.text_to_video("日出时分，海面波光粼粼，一艘帆船缓缓驶来")
    clips.append(clip1["video_url"])

    # 扩展后续片段
    clip2 = generator.extend_video(clip1["video_url"], "帆船靠近港口，渔民开始忙碌")
    clips.append(clip2["video_url"])

    clip3 = generator.extend_video(clip2["video_url"], "渔民们满载而归，脸上洋溢着笑容")
    clips.append(clip3["video_url"])

    print(f"已生成 3 个连续片段: {clips}")

API易 平台通过官方渠道接入 Veo 3.1 能力，技术上与 Google 官方 API 完全一致。主要区别在于:

1. 访问便捷性: 无需科学上网，国内直连访问
2. 付费方式: 按量付费，无需订阅 Google AI Pro
3. 接口格式: 提供 OpenAI 兼容格式，便于集成
4. 额度灵活: 不受 Google 每月 credits 限制

通过 API易 apiyi.com 可以获取免费测试额度，快速验证效果。

所有通过 Veo 3.1 生成的视频都会包含 SynthID 隐形水印，这是 Google 用于标识 AI 生成内容的技术。这个水印:

• 肉眼不可见
• 不影响视频画质
• 可被专门检测工具识别
• 符合 AI 内容标注规范

这是 Google 的安全措施，通过任何渠道调用 Veo 3.1 都会包含此水印。

优化提示词是关键，建议包含以下要素:

• Google 官方: 生成后 2 天内需下载，否则会被删除
• API易 平台: 视频永久保存在你的账户中，可随时下载

根据 Google 服务条款，使用 Veo 3.1 生成的内容可用于商业用途，但需要:

1. 确保提示词和参考图不侵犯他人知识产权
2. 不生成违规内容 (暴力、色情等)
3. 遵守当地法律法规

API易 平台同样遵循这些规范。