Vidu S2 接进现有剪辑流程 / 先跑通一次最短的调用链路

想把 Vidu S2 接进现有剪辑流程，先别急着写批量任务和目录监听。更稳的第一步，是把链路切成鉴权、提交任务、取回结果三段，每段各跑通一次，任何一段没验证过，后面写多长都可能在半路卡住。

适用场景：已有素材目录和剪辑脚本，只想确认 Vidu S2 能否被脚本调用。操作动作：凭据放环境变量，写一个只有鉴权、提交、取回的最小脚本，取回段可以先写死返回值。验证方式：看状态码、任务标识是否唯一、落盘文件属性是否符合预期。风险边界：接口路径、参数名、返回结构以官方文档为准，以下只给占位骨架；轮询间隔与重试次数需要结合自身额度和超时设置确认。

把流程切成三段：鉴权、提交任务、取回结果

先把三段的边界定死，每段只做一件事，输入输出写清楚再动代码。

鉴权段：输入是服务地址和凭据，输出是一个带请求头的客户端或请求头字典，不返回任何业务数据。
提交任务段：输入是请求头加生成参数，输出是任务标识，除此之外不要在这一段里处理结果。
取回结果段：输入是任务标识，输出是任务状态加结果地址或文件字节。

用占位签名把骨架固定下来，函数体可以先留空：

def build_client(api_base: str, api_key: str): ...
def submit_task(client, payload: dict) -> str: ...      # 返回 task_id
def fetch_result(client, task_id: str) -> dict: ...     # 返回 {"status": ..., "files": [...]}

最省时间的做法是把取回结果段先写死返回值，例如直接返回一个表示成功的本地样例结构，让主流程能从第一行跑到最后一行。鉴权段和提交段必须真实发出请求，写死它们等于没验证。

用环境变量管理凭据，先跑通一次鉴权请求

凭据不要写进脚本、不要提交到仓库，也不要打印到日志。先在本机设置：

export VIDU_API_KEY='替换为控制台生成的凭据'
export VIDU_API_BASE='替换为官方文档给出的服务地址'

脚本里只从环境变量取：

import os

api_key = os.environ["VIDU_API_KEY"]
api_base = os.environ["VIDU_API_BASE"].rstrip("/")

headers = {
    "Authorization": f"Bearer {api_key}",   # 头部名称以官方文档为准
    "Content-Type": "application/json",
}

验证方式很朴素：只打印响应状态码和响应体里的错误字段，绝不打印 headers 或 api_key。如果返回 401 或 403，先确认环境变量是否真的被当前进程读到，而不是去改请求体。

提交一次任务并落盘任务标识

提交段的职责只有一个：拿到后续可追踪的句柄。请求体字段按官方文档替换，下面只演示结构：

payload = {
    "model": "替换为文档中的模型标识",
    "prompt": "替换为你的生成描述",
}
resp = client.post(f"{api_base}{SUBMIT_PATH}", json=payload, headers=headers, timeout=30)
resp.raise_for_status()
task_id = extract_task_id(resp.json())   # 取值字段名按文档替换

with open("task_ids.log", "a", encoding="utf-8") as f:
    f.write(task_id + "\n")

日志里至少要留这几项：发出时间、任务标识、payload 摘要（去掉凭据）、响应状态码、响应里的请求 id（如果文档提供了）。验证方式：用同样的参数再提交一次，确认两次任务标识不同；如果相同，通常说明取值取错了字段，取到了固定值或上一个响应。

写轮询与超时分支，处理未完成和失败两种返回

没有超时和失败分支的轮询，最容易出现脚本无声挂住。间隔和上限可以先保守设置，后面按实际耗时调整：

import time

def wait_result(client, task_id, interval=5, max_attempts=60):
    for _ in range(max_attempts):
        r = client.get(f"{api_base}{QUERY_PATH}/{task_id}", headers=headers, timeout=30)
        r.raise_for_status()
        data = r.json()
        status = data.get("status")   # 状态取值以文档为准
        if status in ("succeeded", "success"):
            return data
        if status in ("failed", "error", "canceled"):
            raise RuntimeError(f"task {task_id} 结束于 {status}: {data.get('message')}")
        time.sleep(interval)
    raise TimeoutError(f"task {task_id} 在轮询上限内未完成")

失败和超时都要落到不同的错误分支里，不要用同一个 except 吞掉。验证方式：人为制造一次失败，比如把 max_attempts 设成 2 触发超时分支，或者故意传入一个不存在的任务标识，看脚本是抛错退出还是继续死等。

把结果落盘并接入剪辑流程的素材目录

最后一段决定这条链路能不能被重复调用。文件命名建议带日期、任务标识片段和序号，便于回溯：

{日期}-{task_id 前 8 位}-{序号}.{扩展名}
例：20250101-ab12cd34-01.mp4

下载时先写到临时目录，校验通过后再移动到剪辑流程的素材目录，避免剪辑软件监看到写了一半的文件。落盘后的核查点：文件名与扩展名是否符合命名规则、文件大小是否大于 0、能否用 ffprobe 读出时长和分辨率、下载时的校验值与落盘文件是否一致。

接入剪辑流程可以先从最土的方式开始：让脚本把文件放进素材目录，由剪辑软件或人工确认后再拖进时间线。整条链路跑通后，再做一次重跑对比——用同一个已完成的任务标识再执行一次，应该命中已存在的文件并跳过重复下载，而不是又生成一份同名副本。重跑结果与首次一致，说明这段可以固化下来，再考虑批量提交。