google-gemini-omni-video

Google Gemini Omni 视频

将此技能用于 Google 第一方模型 gemini-omni-flash-preview。以下所有可用性、价格、限制和预览行为均视为已于 2026-07-10 核实；正式发布前请重新检查引用的第一方页面。

执行前明确范围

询问适用的后端、数据区域、源媒体、编辑模式、宽高比、所需时长、音频/对话、目标受众、权利状态和支出审批情况。不要擅自将 Omni 请求转换为其他产品。

Surface	Contract in scope	Do not conflate it with
Gemini Developer API	POST https://generativelanguage.googleapis.com/v1beta/interactions；API/身份验证密钥；Omni 仅限付费层级	消费者版 Gemini 应用、AI Studio UI 行为或 Google Cloud 数据控制
Gemini Enterprise Agent Platform	POST .../v1beta1/projects/{project}/locations/{location}/interactions；OAuth/IAM；Agent Platform API；具体主机取决于经核实的区域约定	可直接替换的 Developer API 端点或旧版 Veo 操作架构
Gemini Omni Flash	原生短视频生成、参考素材、音频生成和对话式编辑	Veo；它是一个专门的视频系列，在受支持的 Veo 版本中具有不同的端点和扩展/插帧等功能
Gemini 原生图像（“Nano Banana”）	独立的图像生成/编辑模型	Omni 视频输出
Gemini app、Flow、YouTube	拥有各自访问权限、设置和条款的消费者/创作者产品	关于 API 数据保留、架构、可用性或来源元数据的承诺
Third-party gateway/reseller	不在本技能范围内	Google 第一方的计费、隐私、安全或重试保证

事实： Gemini Omni Flash 是预览模型，但 Interactions API 本身已正式发布。Developer API 模型页面规定了文本/图像/视频输入、视频输出、1,048,576 个令牌的上下文，以及 3–10 秒、720p/24 FPS 的输出。Cloud 模型页面公布的输入上限则为 131,072 个令牌，输出上限为 57,920 个令牌。仅应用相应后端文档中规定的限制。

事实： Cloud 预览页面公布的停用日期为 2027 年 6 月 30 日。Developer API 页面并未承诺相同的生命周期。固定模型 ID，关注版本说明，并保留一项与提供方绑定的迁移测试。

有意选择后端

如需以最简方式进行第一方集成，并通过 API 密钥管理对话状态，请使用 Developer API。新的 AI Studio 密钥是绑定到服务账号的授权密钥；不受限制的标准密钥已于 2026 年 6 月 19 日停止工作，所有标准密钥计划于 2026 年 9 月起被拒绝。请将密钥保存在服务端、限制其权限，并使用 GEMINI_API_KEY 或 GOOGLE_API_KEY（若两者均存在，则以 GOOGLE_API_KEY 为准）。

如需 Cloud IAM、Cloud 合同、审计／安全控制、GCS 交付或 Cloud 数据治理，请使用 Gemini Enterprise Agent Platform。通过 Application Default Credentials 或短时有效的 OAuth 访问令牌进行身份验证。专用的 Omni 视频指南目前使用 location=global，该端点无法控制处理区域。不过，较新的 Cloud 部署页面将 Omni 列在 eu 和 us 多区域下，并称这些端点会将 ML 处理限制在所选司法管辖区内。请将此视为第一方路由文档之间的冲突：对于受监管的工作负载，只有在项目控制台、SDK 冒烟测试、合同或支持团队确认该 Omni 路由后，才使用 aiplatform.eu.rep.googleapis.com 或 aiplatform.us.rep.googleapis.com Interactions 路由；否则请停止操作，不要回退到 global。

不要根据 Developer 的可用性推断 Cloud 的可用性，反之亦然。截至验证日期，Cloud 模型页面本身存在矛盾：页面标注按需付费不受支持、固定配额受支持，却也在 global 下列出了标准 PayGo。在项目控制台或 Google 支持团队确认之前，请将 Cloud 消费模式、配额分配和价格视为未知。不要仅依据 Developer 价格提交 Cloud 创建请求。

应用当前能力约定

开发者 API 事实：

• 输出为一段带音频的 720p 视频，时长为 3–10 秒，帧率为 24 FPS；支持的宽高比为 16:9 和 9:16（默认横向）。
• generation_config.video_config.task 下的任务为 text_to_video、image_to_video、reference_to_video 和 edit；省略时由模型推断。
• 每次使用 previous_interaction_id 进行编辑都会生成一段新视频。必须保存父交互。
• 对于直接 REST 调用，请从 steps[].type == "model_output" 读取视频内容；output_video 是 SDK 提供的便捷字段。
• 生成的视频包含不可见的 SynthID。开发者 API 指南未承诺支持 C2PA。Cloud 文档另行承诺会自动为 Omni 输出进行 C2PA 签名；不支持 C2PA 的后期处理可能会使其失效。
• 评估仅针对英语；其他语言尚未评估。生成时间不定。
• 安全过滤器会检查提示词和输出，且因地区而异。Interactions 不支持自定义安全设置。

当前开发者 API 不支持：

• 上传音频参考、语音编辑、视频延长、首尾帧插值、多视频推理、系统指令、temperature、top_p、停止序列、专用的 negative_prompts、预配置吞吐量或 YouTube 媒体源。
• 最长 3 秒的视频参考虽可通过架构验证，但无法正确处理；请勿使用。
• 在欧洲经济区、瑞士和英国，无法编辑上传的视频；模型生成的视频在这些地区仍可编辑。
• 在欧洲经济区、瑞士和英国，不支持涉及未成年人的图像上传或编辑。其他地区对部分可识别人物也有更广泛的限制。

**文档冲突：**Google 的模型卡将其描述为可接受音频的原生多模态模型，开发者定价表也列出了音频输入价格，但当前 Developer Omni 指南明确指出不支持上传音频参考。请遵循 API 指南：用文本描述所需的语音、音乐、环境音和音效。Cloud 模型页面称音频仅用于输入，但发送请求前请核实相应的 Cloud 请求指南。

**文档冲突：**通用 Interactions 模型表（更新于 2026-07-09）未列出 Omni，但专门的 Omni 指南和更新日志要求使用 Interactions。请使用专门指南和模型端点，并保留冒烟测试，因为这是一个新的预览版。

规划费用、配额和审批

Developer Omni 没有免费层级。已公布的标准价格为：每百万输入文本／图像／视频／音频 token 收费 $1.50，每百万文本输出 token 收费 $9，每百万视频输出 token（包括思考过程）收费 $17.50。Google 按每秒 5,792 个 token 对 720p 输出计费，约为每秒 $0.10。仅计算视频的精确费用为每秒 $0.10136；10 秒输出在计入输入和任何文本输出前为 $1.0136。

每次创建或编辑前：

1. 统计请求的变体数和轮次；每次编辑都会生成另一份输出，并可能产生额外费用。
2. 提供保守估算：Developer API 的费用为 turns × ($1.0136 maximum video output + estimated inputs/text output)。请注明这是估算，而非服务商的硬性上限。
3. 明确取得批准，批准内容须注明后端、付费创建次数、预估最高费用、保留选项和源媒体。
4. 在 AI Studio 中检查项目的实时限额。限额按项目设置，因层级和模型而异；预览模型的限制更严格，且不作保证。不要硬编码 RPM 值。
5. 将 429 RESOURCE_EXHAUSTED 视为容量／配额信号。遵循任何重试提示，并对轮询／读取操作进行退避；若创建操作结果未知，请勿自动重放。

安全处理源文件和产物

• 要求获得源内容／品牌／性能相关权利及有记录的肖像同意；模型接受不代表已获许可。拒绝亲密影像滥用、冒充、侵犯生物特征／隐私、剥削儿童、监控和禁止用途；广告、公众人物、政治、敏感特征和重大决策相关内容需升级处理。
• 对源文件进行哈希处理，并在源文件－输出记录表中记录 MIME 类型／字节数。验证内容，读取／解码前先设定上限，拒绝可执行／多态文件，并将上传内容设为私有。切勿记录密钥、base64、签名 URI、含秘密的提示词或思维过程。
• Developer Files API 上传的单个文件上限为 2 GB，每个项目上限为 20 GB，文件会保留 48 小时，且无法通过该 API 下载，但可以手动删除。使用后应尽早删除。
• Developer 付费版 Interactions 默认存储 55 天，可配置为 7／14／28／55 天；按 ID 删除。store=false 会禁用父级状态／后台处理。滥用内容另行保留 55 天，且可能由获授权人员审核。付费内容不会用于改进产品，但仍会记录滥用日志；消费者／免费版条款不适用于此处。
• Cloud 异步内容最多保留 14 天。即使条款禁止训练，仍有滥用／Advanced-AI／日志记录例外；隔离内存缓存默认保留 24 小时，除非将其禁用。确认协议、例外、缓存以及 GCS IAM／数据驻留／加密／日志记录／生命周期设置。
• 验证 MP4 的时长、尺寸、FPS、音频、完整解码、大小和哈希；发布前检查画面、对白、文字、徽标、人脸、连续性、安全性以及事实／品牌声明。

构建可测试的提示词

明确主体、动作、环境、镜头、镜头／摄影机运动、照明、色调、节奏、物理效果和声音。若要单个镜头，请写明 single continuous unbroken shot; no scene cuts。用普通文字说明排除项：No dialogue. No extra sound effects.

使用自然语言描述时间或时间码：

Single continuous unbroken 9:16 shot; no scene cuts.
[0-3s] A paper kite rises above a quiet dawn beach; slow handheld push-in.
[3-6s] The kite crosses the sun and the camera tilts upward.
Audio: soft surf and light wind. No dialogue, no music.
Keep all lettering and logos out of frame.

时间码只是提示词指引，并不保证帧级精确的剪辑效果。请检查生成结果。

指定图像用途时，保持媒体顺序稳定，并明确绑定：

• <FIRST_FRAME> 将图像用作起始帧。
• <IMAGE_REF_N> 用于标识参考图像；编号从 0 开始。
• 存在歧义时，使用 Google 的明确源文件／参考前缀，例如 [# Sources <FIRST_FRAME>@Image1] 或 [# References <IMAGE_REF_0>@Image1]。

Google 展示了多张参考图像，但 Developer 指南未公布具体数量上限。不要自行编造上限。Cloud 模型页面另行公布每个提示词最多支持 10 张图像。

编辑时，每轮只做一项更改：Make the room lighting warmer. Keep everything else the same. 身份、布局、运动、文字和音频的一致性可能会在编辑过程中发生漂移。模型卡将跨编辑的一致性、复杂运动和完美文字列为已知限制；应进行验证，而不要声称这些内容一定会保持不变。

使用安全的 Developer API 工作流

对于超过 4 MB 的输出，优先使用 URI 交付。立即保存初始交互 ID 和 URI：目前，后续的 GET /interactions/{id} 会以内联 base64 返回内容，即使创建请求指定了 URI，因此只有在初始响应或 SSE 流中才能保证获得 URI。轮询对应的 File 资源，直到状态变为 ACTIVE；若状态为 FAILED 或达到截止时间，则失败；然后使用身份验证下载到临时文件，进行验证，并以原子方式重命名。

Omni 没有文档说明的创建幂等键。执行 POST 前，持久化一条尝试记录，其中包含客户端尝试 UUID、请求哈希、审批信息和 posting 状态。如果提交后连接失败，且没有可用的交互 ID，则将状态标记为 create_outcome_unknown，停止操作，并与操作员一起核对项目日志和账单。绝不要“以防万一”重试。GET、文件状态轮询和下载均为只读操作，可以使用有上限的抖动重试。

以下完整的 Python 示例涵盖文本、有状态对话，以及单独暂存的上传视频编辑。默认采用离线试运行，不会发起付费调用。仅在执行时安装 google-genai>=2.3.0。

#!/usr/bin/env python3
import argparse, base64, hashlib, http.client, ipaddress, json, os, random; import socket, ssl, subprocess, tempfile, threading, time, uuid; from decimal import Decimal, InvalidOperation; from fractions import Fraction
from pathlib import Path; from urllib.parse import parse_qs, urlsplit; MODEL = "gemini-omni-flash-preview"; ESTIMATED_MAX_USD = 1.10  # one 10s Developer turn; estimate, not a provider cap
MAX_VIDEO_BYTES = 80 * 1024 * 1024; MAX_TOOL_BYTES = 1024 * 1024; DOWNLOAD_HOST = "generativelanguage.googleapis.com"
def canonical(value) -> bytes:
    return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode()
def atomic_json(path: Path, value: dict) -> None:
    path.parent.mkdir(parents=True, exist_ok=True); fd, tmp = tempfile.mkstemp(prefix=path.name + ".", dir=path.parent)
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as f:
            json.dump(value, f, indent=2, sort_keys=True); f.flush(); os.fsync(f.fileno())
        os.replace(tmp, path)
    finally:
        if os.path.exists(tmp): os.unlink(tmp)
def claim_once(path: Path, value: dict) -> None:
    path.parent.mkdir(parents=True, exist_ok=True); fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
    with os.fdopen(fd, "w", encoding="utf-8") as f:
        json.dump(value, f, indent=2, sort_keys=True); f.flush(); os.fsync(f.fileno())
def read_record(path: Path) -> dict:
    if path.stat().st_size > MAX_TOOL_BYTES: raise RuntimeError("Attempt record size rejected")
    with path.open("rb") as source: raw = source.read(MAX_TOOL_BYTES + 1)
    if not raw or len(raw) > MAX_TOOL_BYTES: raise RuntimeError("Attempt record size rejected")
    value = json.loads(raw)
    if not isinstance(value, dict): raise RuntimeError("Attempt record must be an object")
    return value
def run_bounded(argv, timeout, cap=MAX_TOOL_BYTES):
    proc = subprocess.Popen(argv, shell=False, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE); out, err, lock, overflow = bytearray(), bytearray(), threading.Lock(), threading.Event()
    total = 0
    def drain(stream, bucket):
        nonlocal total
        while True:
            chunk = stream.read1(65536) if hasattr(stream, "read1") else stream.read(65536)
            if not chunk: break
            with lock:
                room = max(0, cap - total); bucket.extend(chunk[:room]); total += len(chunk)
                if total > cap: overflow.set()
            if overflow.is_set():
                try: proc.kill()
                except OSError: pass
    threads = [threading.Thread(target=drain, args=(proc.stdout, out), daemon=True), threading.Thread(target=drain, args=(proc.stderr, err), daemon=True)]
    for thread in threads: thread.start()
    timed_out = False
    try: code = proc.wait(timeout=timeout)
    except subprocess.TimeoutExpired:
        timed_out = True; proc.kill(); code = proc.wait()
    for thread in threads: thread.join()
    return {"code": code, "stdout": bytes(out), "stderr": bytes(err), "timeout": timed_out, "overflow": overflow.is_set()}
class PinnedHTTPSConnection(http.client.HTTPSConnection):
    def __init__(self, host, ip, timeout=30):
        super().__init__(host, 443, timeout=timeout, context=ssl.create_default_context()); self.ip = ip
    def connect(self):
        sock = socket.create_connection((self.ip, self.port), self.timeout); self.sock = self._context.wrap_socket(sock, server_hostname=self.host)
def generated_file_name(uri: str) -> str:
    parsed = urlsplit(uri)
    if (parsed.scheme != "https" or parsed.hostname != DOWNLOAD_HOST or parsed.username or parsed.password or parsed.fragment or parsed.port not in (None, 443)):
        raise RuntimeError("Generated file URI policy rejected")
    prefix = "/v1beta/files/"
    if not parsed.path.startswith(prefix) or not parsed.path.endswith(":download"):
        raise RuntimeError("Generated file URI path rejected")
    name = parsed.path[len(prefix):-len(":download")]
    if not name or not name.isalnum() or parse_qs(parsed.query) != {"alt": ["media"]}:
        raise RuntimeError("Generated file URI identifier/query rejected")
    return "files/" + name
def download_generated(uri: str, api_key: str, parent: Path) -> Path:
    name = generated_file_name(uri).split("/", 1)[1]; addresses = sorted({item[4][0] for item in socket.getaddrinfo(DOWNLOAD_HOST, 443, type=socket.SOCK_STREAM)})
    if not addresses or any(not ipaddress.ip_address(item).is_global for item in addresses):
        raise RuntimeError("Generated file host did not resolve publicly")
    conn, temp_path = PinnedHTTPSConnection(DOWNLOAD_HOST, addresses[0], timeout=60), None
    try:
        conn.request("GET", f"/v1beta/files/{name}:download?alt=media", headers={"x-goog-api-key": api_key, "Accept": "video/mp4"}); response = conn.getresponse()
        if 300 <= response.status < 400: raise RuntimeError("Generated file redirects are rejected")
        if response.status != 200: raise RuntimeError(f"Generated file HTTP {response.status}")
        mime = response.getheader("Content-Type", "").split(";", 1)[0].lower()
        if mime != "video/mp4": raise RuntimeError("Generated file MIME rejected")
        length = response.getheader("Content-Length")
        if length is not None and (not length.isdigit() or int(length) > MAX_VIDEO_BYTES):
            raise RuntimeError("Generated file declared size rejected")
        fd, name_on_disk = tempfile.mkstemp(prefix=".omni-", suffix=".mp4", dir=parent); temp_path = Path(name_on_disk); total = 0
        with os.fdopen(fd, "wb") as f:
            while True:
                chunk = response.read(1024 * 1024)
                if not chunk: break
                total += len(chunk)
                if total > MAX_VIDEO_BYTES: raise RuntimeError("Generated file crossed byte cap")
                f.write(chunk)
            f.flush(); os.fsync(f.fileno())
        return temp_path
    except BaseException:
        if temp_path is not None and temp_path.exists(): temp_path.unlink()
        raise
    finally:
        conn.close()
def inline_generated(data: str, parent: Path) -> Path:
    if len(data) > ((MAX_VIDEO_BYTES + 2) // 3) * 4 + 4:
        raise RuntimeError("Inline video encoding exceeds byte cap")
    blob = base64.b64decode(data, validate=True)
    if not blob or len(blob) > MAX_VIDEO_BYTES: raise RuntimeError("Inline video size rejected")
    fd, name = tempfile.mkstemp(prefix=".omni-", suffix=".mp4", dir=parent)
    with os.fdopen(fd, "wb") as f: f.write(blob); f.flush(); os.fsync(f.fileno())
    return Path(name)
def validate_and_publish(temp_path: Path, destination: Path) -> dict:
    try:
        with temp_path.open("rb") as f: header = f.read(12)
        if len(header) < 12 or header[4:8] != b"ftyp": raise RuntimeError("MP4 signature rejected")
        probe = run_bounded(["ffprobe", "-v", "error", "-show_entries", "format=duration,size:stream=index,codec_type,codec_name,width,height,r_frame_rate", "-of", "json", str(temp_path)], 30)
        if probe["timeout"] or probe["overflow"] or probe["code"] != 0: raise RuntimeError("ffprobe failed")
        metadata = json.loads(probe["stdout"]); streams = metadata.get("streams", []); videos = [item for item in streams if item.get("codec_type") == "video"]
        if len(videos) != 1 or not any(item.get("codec_type") == "audio" for item in streams):
            raise RuntimeError("Expected one video stream and generated audio")
        width, height = videos[0].get("width"), videos[0].get("height")
        if sorted((width, height)) != [720, 1280]: raise RuntimeError("Expected 720p dimensions")
        if abs(float(Fraction(videos[0].get("r_frame_rate", "0/1"))) - 24.0) > 0.1:
            raise RuntimeError("Expected 24 FPS")
        duration = float(metadata.get("format", {}).get("duration", 0))
        if not 2.9 <= duration <= 10.1: raise RuntimeError("Video duration outside 3-10 seconds")
        decode = run_bounded(["ffmpeg", "-v", "error", "-i", str(temp_path), "-f", "null", "-"], 300)
        if decode["timeout"] or decode["overflow"] or decode["code"] != 0: raise RuntimeError("Full decode failed")
        if destination.exists(): raise RuntimeError("Output already exists")
        digest = hashlib.sha256()
        with temp_path.open("rb") as f:
            for chunk in iter(lambda: f.read(1024 * 1024), b""): digest.update(chunk)
        artifact = {"path": str(destination), "bytes": temp_path.stat().st_size, "sha256": digest.hexdigest(), "probe": metadata}; os.replace(temp_path, destination); temp_path = None
        return artifact
    finally:
        if temp_path is not None and temp_path.exists(): temp_path.unlink()
def retry_after(exc) -> float | None:
    response = getattr(exc, "response", None); headers = getattr(response, "headers", {}) or {}; raw = headers.get("retry-after") or headers.get("Retry-After")
    try: value = float(raw)
    except (TypeError, ValueError): return None
    return value if 0 <= value <= 60 else None
def safe_read(call, deadline, attempts=5):
    for attempt in range(attempts):
        try: return call()
        except Exception as exc:
            if attempt + 1 == attempts or time.monotonic() >= deadline: raise
            delay = retry_after(exc) or min(20.0, 2 ** attempt) * (1 + random.random() * .25)
            if time.monotonic() + delay >= deadline: raise
            time.sleep(delay)
def find_video(interaction):
    convenience = getattr(interaction, "output_video", None)
    if convenience: return convenience
    for step in getattr(interaction, "steps", []) or []:
        if getattr(step, "type", None) == "model_output":
            for item in getattr(step, "content", []) or []:
                if getattr(item, "type", None) == "video": return item
    raise RuntimeError("No video output in completed interaction")
def delete_remote_files(client, items) -> list[dict]:
    results = []
    for role, name in items:
        try:
            client.files.delete(name=name)  # one cleanup attempt; ambiguity is recorded for an operator; results.append({"role": role, "name": name, "status": "deleted"})
        except Exception as exc:
            results.append({"role": role, "name": name, "status": "delete_outcome_unknown", "errorType": type(exc).__name__})
    return results
def load_source_manifest(path: Path, jurisdiction: str | None) -> dict:
    if jurisdiction in {"eea", "ch", "uk"}:
        raise RuntimeError("Uploaded-video editing is unavailable in this jurisdiction")
    if jurisdiction != "other":
        raise RuntimeError("--source-jurisdiction other is required for an uploaded-video edit")
    if path.is_symlink() or not path.is_file(): raise RuntimeError("Source manifest path rejected")
    if path.stat().st_size > MAX_TOOL_BYTES: raise RuntimeError("Source manifest size rejected")
    with path.open("rb") as source_file: raw = source_file.read(MAX_TOOL_BYTES + 1)
    if not raw or len(raw) > MAX_TOOL_BYTES: raise RuntimeError("Source manifest size rejected")
    value = json.loads(raw)
    if not isinstance(value, dict): raise RuntimeError("Source manifest must be an object")
    required = {"name", "uri", "sha256", "bytes", "mime", "state", "deleteAfterUse", "durationSeconds", "localFullDecodePassed", "rightsApproved"}
    if set(value) != required: raise RuntimeError("Source manifest schema rejected")
    name = value["name"]
    if not isinstance(name, str) or not name.startswith("files/") or not name[6:].isalnum():
        raise RuntimeError("Source File resource name rejected")
    parsed = urlsplit(value["uri"] if isinstance(value["uri"], str) else "")
    if (parsed.scheme != "https" or parsed.hostname != DOWNLOAD_HOST or parsed.username or parsed.password or parsed.port not in (None, 443) or parsed.query or parsed.fragment or parsed.path != "/v1beta/" + name):
        raise RuntimeError("Source File URI rejected")
    sha = value["sha256"]
    if not isinstance(sha, str) or len(sha) != 64 or any(c not in "0123456789abcdef" for c in sha):
        raise RuntimeError("Source SHA-256 rejected")
    if (type(value["bytes"]) is not int or not 0 < value["bytes"] <= 2 * 1024 * 1024 * 1024 or value["mime"] != "video/mp4" or value["state"] != "ACTIVE" or value["deleteAfterUse"] is not True or value["localFullDecodePassed"] is not True or value["rightsApproved"] is not True):
        raise RuntimeError("Source validation/lifecycle assertions rejected")
    try: duration = Decimal(str(value["durationSeconds"]))
    except InvalidOperation as exc: raise RuntimeError("Source duration rejected") from exc
    if not duration.is_finite() or not Decimal("0") < duration <= Decimal("10"):
        raise RuntimeError("Uploaded source must be no longer than 10 seconds")
    return value
def main() -> int:
    p = argparse.ArgumentParser(); p.add_argument("--prompt", required=True); p.add_argument("--aspect", choices=("16:9", "9:16"), default="16:9")
    p.add_argument("--stateful", action="store_true", help="store this turn so it can be edited later"); p.add_argument("--previous-id", help="stored parent interaction for a paid edit turn")
    p.add_argument("--source-manifest", type=Path, help="bounded manifest for an already ACTIVE, locally validated Developer File upload"); p.add_argument("--source-jurisdiction", choices=("other", "eea", "ch", "uk"))
    p.add_argument("--output", type=Path, default=Path("omni-output.mp4")); p.add_argument("--attempt-log", type=Path, default=Path("omni-attempt.json"))
    p.add_argument("--attempt-id", required=True, help="fresh canonical UUIDv4 used for this one paid turn"); p.add_argument("--execute", action="store_true")
    p.add_argument("--resume", action="store_true", help="read/query a claimed attempt; never create"); p.add_argument("--max-usd", default="1.10"); p.add_argument("--approval-sha256")
    p.add_argument("--show-full-plan", action="store_true"); a = p.parse_args()
    if not a.prompt.strip() or len(a.prompt.encode("utf-8")) > 8000:
        raise SystemExit("Prompt must be 1..8000 UTF-8 bytes")
    if a.output.suffix.lower() != ".mp4": raise SystemExit("Output must end in .mp4")
    try: attempt = uuid.UUID(a.attempt_id); maximum = Decimal(a.max_usd)
    except (ValueError, InvalidOperation) as exc: raise SystemExit("Attempt ID or maximum USD is invalid") from exc
    if attempt.version != 4 or str(attempt) != a.attempt_id.lower(): raise SystemExit("Attempt ID must be canonical UUIDv4")
    if not maximum.is_finite() or maximum <= 0 or maximum < Decimal("1.10"):
        raise SystemExit("Maximum USD must be finite, positive, and cover 1.10")
    output, attempt_log = a.output.resolve(), a.attempt_log.resolve()
    if output == attempt_log: raise SystemExit("Output and attempt record must differ")
    if a.source_jurisdiction and not a.source_manifest:
        raise SystemExit("--source-jurisdiction requires --source-manifest")
    if a.source_manifest and a.previous_id:
        raise SystemExit("Choose either an uploaded-video edit or a previous-interaction edit")
    source = load_source_manifest(a.source_manifest.resolve(), a.source_jurisdiction) if a.source_manifest else None
    request = { "model": MODEL, "input": a.prompt, "response_format": {"type": "video", "delivery": "uri", "aspect_ratio": a.aspect}, "background": False, "stream": False, "store": a.stateful or bool(a.previous_id), }
    if source:
        request["input"] = [{"type": "document", "uri": source["uri"]}, {"type": "text", "text": a.prompt}]; request["generation_config"] = {"video_config": {"task": "edit"}}
    if a.previous_id: request["previous_interaction_id"] = a.previous_id
    request_hash = hashlib.sha256(canonical(request)).hexdigest()
    envelope = {"backend": "gemini-developer-api", "endpoint": "/v1beta/interactions", "model": MODEL, "request": request, "sourceHashes": ([] if not source else [{"sha256": source["sha256"], "bytes": source["bytes"], "mime": source["mime"], "fileName": source["name"], "jurisdiction": a.source_jurisdiction, "deleteAfterUse": True}]), "attemptId": str(attempt), "attemptRecord": str(attempt_log), "outputPolicy": {"path": str(output), "mime": "video/mp4", "maxBytes": MAX_VIDEO_BYTES, "ffprobe": True, "fullDecode": True, "requireAudio": True}, "cost": {"estimatedMaxUsd": "1.10", "approvedMaxUsd": format(maximum, "f"), "priceAsOf": "2026-07-10", "paidCreates": 1}, "retention": {"storeInteraction": request["store"], "previousInteractionId": a.previous_id}}
    approval = hashlib.sha256(canonical(envelope)).hexdigest(); redacted_input = f"sha256:{hashlib.sha256(a.prompt.encode()).hexdigest()}"
    if source:
        redacted_input = [{"type": "document", "uriSha256": hashlib.sha256(source["uri"].encode()).hexdigest()}, {"type": "text", "sha256": hashlib.sha256(a.prompt.encode()).hexdigest()}]
    safe_envelope = {**envelope, "request": {**request, "input": redacted_input}}; print(json.dumps({"dry_run": not a.execute, "approval_sha256": approval, "plan": safe_envelope}, indent=2))
    if a.show_full_plan: print(json.dumps({"protected_full_envelope": envelope}, indent=2))
    if not a.execute: return 0
    if a.approval_sha256 != approval: raise SystemExit("Exact approval hash mismatch")
    api_key = os.getenv("GOOGLE_API_KEY") or os.getenv("GEMINI_API_KEY")
    if not api_key:
        raise SystemExit("Set GEMINI_API_KEY or GOOGLE_API_KEY server-side")
    from google import genai; from google.genai import types; create_client = genai.Client(api_key=api_key, http_options=types.HttpOptions( timeout=600000, retry_options=types.HttpRetryOptions(attempts=1)))
    read_client = genai.Client(api_key=api_key, http_options=types.HttpOptions( timeout=30000, retry_options=types.HttpRetryOptions(attempts=1)))
    if source:
        info = safe_read(lambda: read_client.files.get(name=source["name"]), time.monotonic() + 120); state = getattr(getattr(info, "state", None), "name", getattr(info, "state", None))
        if state != "ACTIVE" or getattr(info, "uri", None) != source["uri"]:
            raise RuntimeError("Approved source upload is not the same ACTIVE File resource")
    if a.resume:
        if not attempt_log.exists(): raise SystemExit("Resume record does not exist")
        record = read_record(attempt_log)
        if (record.get("attempt_id") != str(attempt) or record.get("approval_sha256") != approval or record.get("request_sha256") != request_hash):
            raise SystemExit("Resume record does not match this exact approval")
        interaction_id = record.get("interaction_id")
        if not isinstance(interaction_id, str) or not interaction_id:
            raise SystemExit("Create outcome is unknown without an interaction ID; do not replay")
        interaction = safe_read(lambda: read_client.interactions.get(interaction_id), time.monotonic() + 120)
    else:
        record = {"attempt_id": str(attempt), "backend": "gemini-developer-api", "model": MODEL, "approval_sha256": approval, "request_sha256": request_hash, "status": "posting", "approved_max_usd": format(maximum, "f"), "interaction_id": None, "source_file_name": source["name"] if source else None, "output_uri_sha256": None, "created_unix": int(time.time())}
        try: claim_once(attempt_log, record)
        except FileExistsError as exc: raise SystemExit("Attempt is already claimed; use --resume, never replay") from exc
        try:
            interaction = create_client.interactions.create(**request)  # exactly one network attempt
        except Exception as exc:
            record.update(status="create_outcome_unknown", error_type=type(exc).__name__, updated_unix=int(time.time())); atomic_json(attempt_log, record)
            raise RuntimeError("Create outcome unknown; reconcile logs/billing, do not retry") from exc
        interaction_id = getattr(interaction, "id", None)
        if not isinstance(interaction_id, str) or not interaction_id:
            record.update(status="create_outcome_unknown", error_type="MissingInteractionId"); atomic_json(attempt_log, record)
            raise RuntimeError("Create returned no interaction ID; do not retry")
        record.update(interaction_id=interaction_id, status="interaction_id_saved", updated_unix=int(time.time())); atomic_json(attempt_log, record)  # first action after create: preserve provider identity
    video = find_video(interaction); uri = getattr(video, "uri", None); data = getattr(video, "data", None)
    if uri:
        file_name = generated_file_name(uri); record.update(status="output_uri_saved", output_uri_sha256=hashlib.sha256(uri.encode()).hexdigest(), generated_file_name=file_name, updated_unix=int(time.time()))
        atomic_json(attempt_log, record)
    else: file_name = None
    if uri:
        deadline = time.monotonic() + 900
        while True:
            info = safe_read(lambda: read_client.files.get(name=file_name), deadline); state = getattr(getattr(info, "state", None), "name", getattr(info, "state", None))
            if state == "ACTIVE": break
            if state == "FAILED": raise RuntimeError("Provider file processing failed")
            if state not in {"PROCESSING", "PENDING", None}: raise RuntimeError(f"Unknown file state {state!r}")
            if time.monotonic() >= deadline: raise TimeoutError("File processing exceeded 15 minutes")
            time.sleep(5 + random.random())
        output.parent.mkdir(parents=True, exist_ok=True); temp_path = download_generated(uri, api_key, output.parent)
    elif data:
        output.parent.mkdir(parents=True, exist_ok=True); temp_path = inline_generated(data, output.parent)
    else:
        raise RuntimeError("Video has neither URI nor data")
    try:
        artifact = validate_and_publish(temp_path, output)
    except BaseException as exc:
        cleanup_items = ([] if not file_name else [("generated_output", file_name)])
        if source: cleanup_items.append(("uploaded_source", source["name"]))
        cleanup_results = delete_remote_files(read_client, cleanup_items); cleanup_unknown = any(item["status"] != "deleted" for item in cleanup_results)
        record.update(status="artifact_validation_failed_cleanup_pending" if cleanup_unknown else "artifact_validation_failed", error_type=type(exc).__name__, remote_file_cleanup=cleanup_results, cleanup_action=("Inspect/delete the recorded File resources; do not replay create" if cleanup_unknown else None), updated_unix=int(time.time()))
        atomic_json(attempt_log, record); raise
    cleanup_items = ([] if not file_name else [("generated_output", file_name)])
    if source: cleanup_items.append(("uploaded_source", source["name"]))
    record.update(status="artifact_saved_cleanup_pending" if cleanup_items else "artifact_saved", artifact=artifact, retention=envelope["retention"], rights_review="required", creative_qa="pending", synthetic_disclosure="required", synthid_expected=True, developer_c2pa="not_documented", updated_unix=int(time.time()))
    atomic_json(attempt_log, record)
    if cleanup_items:
        cleanup_results = delete_remote_files(read_client, cleanup_items); cleanup_unknown = any(item["status"] != "deleted" for item in cleanup_results)
        record.update(status="artifact_saved_cleanup_pending" if cleanup_unknown else "artifact_saved", remote_file_cleanup=cleanup_results, cleanup_action=("Inspect/delete the recorded File resources; do not replay create" if cleanup_unknown else None), updated_unix=int(time.time()))
        atomic_json(attempt_log, record)
        if cleanup_unknown:
            raise RuntimeError("Artifact saved, but remote File cleanup needs operator reconciliation")
    print(json.dumps(record, indent=2)); return 0
if __name__ == "__main__": raise SystemExit(main())

先进行试运行：

python omni_generate.py --attempt-id "7f8f06b7-9e68-4f0a-bf16-3eaf3ee9d6c4" --prompt "Single continuous shot of rain on a greenhouse roof. Audio: rain only. No dialogue."

获得审批后，恰好执行一次付费轮次：

python omni_generate.py --execute --approval-sha256 "<exact unchanged digest>" --max-usd 1.10 --attempt-id "7f8f06b7-9e68-4f0a-bf16-3eaf3ee9d6c4" --prompt "Single continuous shot of rain on a greenhouse roof. Audio: rain only. No dialogue."

如果你打算编辑某个首轮结果，请在试运行前添加 --stateful，并在执行时保持不变；默认的独立结果是无状态的，之后无法将其设为父项。在审批中记录保留选项。使用相同参数加上 --resume --execute 恢复已知交互；恢复操作绝不会进入创建分支。

对于图像输入，请将 input 替换为有序项目，并显式设置任务：

request["input"] = [
    {"type": "image", "data": base64.b64encode(image_bytes).decode(), "mime_type": "image/png"},
    {"type": "text", "text": "<FIRST_FRAME> single unbroken shot; the paper boat drifts forward. Keep its design unchanged. Audio: creek only."},
]
request["generation_config"] = {"video_config": {"task": "image_to_video"}}

替换图像前，请限制字节数、验证内容、计算哈希并审批相关权利。对于上传视频编辑，请单独审批暂存：拒绝符号链接和多格式文件，将大小限制为 2 GB，要求 MP4 时长不超过 10 秒，进行有界探测、完整解码并计算哈希，只上传一次，然后轮询 PROCESSING → ACTIVE|FAILED。记录 Google 的准确值：

{"name":"files/abc123","uri":"https://generativelanguage.googleapis.com/v1beta/files/abc123","sha256":"<64 lowercase hex>","bytes":1234567,"mime":"video/mp4","state":"ACTIVE","deleteAfterUse":true,"durationSeconds":8.0,"localFullDecodePassed":true,"rightsApproved":true}

在试运行和执行时，均原样传入 --source-manifest source.json --source-jurisdiction other；使用受保护的 --show-full-plan 检查确切提示词/URI。执行时会再次检查 ACTIVE URI，使用文档输入，并在验证后删除源 File 和生成的 File。记录创建或清理过程中出现的模糊结果，以便核对；绝不要重放操作。在 eea、ch 或 uk 中停止操作；绝不要使用有缺陷的时长不超过 3 秒的参考路径。

单独使用 Cloud 合约

Cloud 文档记录的同步请求支持 3s 到 10s 的显式 duration 字符串、GCS URI 交付，以及列表类型的 response_format。未经核对 Developer 参考文档，不要将这些字段复制到 Developer 代码中。

只有在核实确切的项目路由、配额模式和价格后，才能使用此故障关闭事务合约；在此之前，只准备试运行，不要提供通用的生产请求：

1. 要求提供规范的 UUIDv4 尝试 ID、确切的项目／位置／来源、私有 GCS 前缀、完整请求 JSON、输出路径／策略、当前价格凭据、有限且为正的估算值，以及有限且足以覆盖估算值的上限。将这些字段规范化并打印其 SHA-256；执行时必须使用未变更的摘要，而非布尔批准值。
2. 默认离线试运行：不查找 ADC、不创建目录、不调用 GCS，也不访问网络。执行时，使用 OAuth／ADC，并以原子方式创建新的 mode-0700 尝试目录。如果目录已存在，则失败；单独选择的恢复分支只能对持久化的已知交互 ID 执行 GET。
3. 发送一个有界 POST，不自动重试。将受保护的请求正文放入尝试目录中的 mode-0600 临时文件，避免提示词／媒体详情暴露在进程列表中。立即持久化返回的 ID。如果超时或解析失败，且没有持久化的 ID，则将结果标记为 create_outcome_unknown；应进行协调核对，而不是重放请求。
4. 限制 GET 重试次数并遵守 Retry-After。仅持久化受保护的交互 ID、请求／批准哈希、确切的清理 URI、响应哈希和经过净化的状态——绝不保存原始响应或模型思维内容。要求返回的视频 URI 恰好为一个，且位于批准的 GCS 前缀下。
5. 限制元数据、下载以及工具的耗时／输出。要求 video/mp4、已声明且实际观测到的字节数上限、MP4 ftyp 魔数、恰好一个 720p／约 24 FPS 视频流、生成的音频、3～10 秒时长、受限的 ffprobe、完整且受限的 ffmpeg 解码，以及当前版本 c2patool 验证成功。若目标位置已存在文件，则拒绝操作；通过原子重命名并计算哈希来发布。
6. 本地制品和受保护的清单持久化后，再通过有界调用删除 GCS 对象。如果删除结果不明确，则保留 artifact_saved_cleanup_pending 状态和确切 URI；绝不要为了完成清理而重新创建。任何发布转换后都要重新验证 C2PA。

对于 background:true，应用相同的创建／身份规则，并进行有界的只读轮询；Cloud 文档说明异步保留时间最长为 14 天。适用的 Cloud 协议、滥用监控例外、可选的请求／响应日志记录、24 小时内存缓存设置、存储桶 IAM／驻留区域／保留策略，以及交互／对象删除，仍是单独的批准字段。

按阶段处理故障

• **预检／架构／身份验证／区域：**修正配置；如果服务在接受请求前已明确拒绝，则未发生创建。
• **创建返回 ID：**持久化该 ID，并轮询／读取对应交互；不要仅因生成较慢就创建替代项。
• **创建连接失败且无 ID：**结果未知；停止并进行协调核对。没有文档记录的幂等键。
• **安全拦截：**不要削弱或规避控制。只有在请求合法且合规时才进行修改；遇到模糊的肖像权／版权问题时应升级处理。
• **文件处于 PROCESSING 状态：**使用抖动进行有界轮询；FAILED 为终止状态。无法找回 URI 时，不要假定 GET interaction 会返回该 URI。
• **制品验证失败：**隔离输出并保留元数据；不要发布。新一轮付费操作需要新的批准。
• **预览版变更或 404／不支持的字段：**重新查看模型页面、Omni 指南、变更日志和后端参考文档；不要猜测替代模型。

区分证据与判断

在计划和报告中使用以下标签：

• **FACT：**直接记录的当前 API 行为、模型限制、价格、政策或条款。
• **PROVIDER CLAIM：**Google 的定性陈述，或内部报告的基准测试／模型卡结果；不要将其当作独立证据。
• **OBSERVATION：**来自注明日期的本地试运行或付费测试的结果，包括后端、模型 ID、SDK 版本、区域和请求哈希。
• **HEURISTIC：**提示词建议，例如每轮只做一次编辑，或“其他内容保持不变”。
• **UNKNOWN：**未记录的 Developer API 最大参考数量、硬性延迟 SLA、创建操作的幂等性、Developer C2PA、Cloud 消耗模式冲突，或任何未经验证的 SDK 行为。

首先重新核实以下第一方来源

• Developer 指南和提示词语法：https://ai.google.dev/gemini-api/docs/omni
• Developer 模型详情：https://ai.google.dev/gemini-api/docs/models/gemini-omni-flash
• Interactions 状态／保留／最低 SDK 版本：https://ai.google.dev/gemini-api/docs/interactions-overview
• 定价和实时限制：https://ai.google.dev/gemini-api/docs/pricing and https://ai.google.dev/gemini-api/docs/rate-limits
• Files API：https://ai.google.dev/gemini-api/docs/files
• API 密钥：https://ai.google.dev/gemini-api/docs/api-key
• 发布说明：https://ai.google.dev/gemini-api/docs/changelog
• 条款、区域和滥用监控：https://ai.google.dev/gemini-api/terms, https://ai.google.dev/gemini-api/docs/available-regions, and https://ai.google.dev/gemini-api/docs/usage-policies
• 禁止用途：https://policies.google.com/terms/generative-ai/use-policy
• DeepMind 产品和模型卡：https://deepmind.google/models/gemini-omni/ and https://deepmind.google/models/model-cards/gemini-omni-flash/
• Cloud 模型／后端：https://docs.cloud.google.com/gemini-enterprise-agent-platform/models/gemini/omni-flash-preview and https://docs.cloud.google.com/gemini-enterprise-agent-platform/models/video/generate-videos-from-text
• Cloud C2PA 和端点驻留区域：https://docs.cloud.google.com/gemini-enterprise-agent-platform/models/content-credentials and https://docs.cloud.google.com/gemini-enterprise-agent-platform/resources/locations