十几条任务批量出片时，最容易出错的环节是参数散落在每条任务里，模型本身通常不是主因：分辨率在一处改了、另一处漏改，时长在页面手填过、脚本里又写死一份，跑完一批才发现画幅不对，事后连是哪条任务、哪一版参数都说不清。处理方向是先在页面单条跑通，把分辨率、时长、画幅抽成一份参数配置对象，提示词按一行一条放在独立列表里，脚本只负责组合、提交、记录任务标识和回收结果。判断标准很直接：新增任务只加一行提示词，改参数只改一处。

批量出片的关键在于参数只有一个来源，脚本写得复杂与否并不是重点。建议先在页面单条跑通一条任务并留下参数记录，再把分辨率、时长、画幅写成一份 JSON 配置，提示词按行拆成列表，脚本只做组合与回收。适用场景是一批几条到十几条、参数大体一致的出片任务；边界在于参数模板只覆盖整批一致的通用项，单条任务特有的内容仍要写在提示词行里。验证方式是随机抽两条，比对参数快照与页面单条产出一致。

先在页面单条跑通一条任务，确认参数项与产出

批量之前先跑一条，目的并不在于评价效果好坏，关键在于把人工流程固定成一份可照抄的记录。页面参数区建议整屏截图留一份，同时把各项手抄成一份 JSON：截图用来对照页面上的位置，JSON 用来给脚本读取；两者对不上时，以截图为准回头修正 JSON。

抄的时候顺手分两栏标注：哪些项每条任务都要改，哪些项整批保持不变。常见的情况是提示词、时长、输出文件名序号每次要改，分辨率、画幅、模型版本整批一致。一时分不清就先按“不同镜头可能不同就归到每次改”处理，宁可多标一项。

{
  "note": "单条验证任务，页面跑通后手工抄录，截图另存",
  "prompt": "海边日落，镜头缓慢推近",
  "config": {
    "resolution": "4K",
    "duration_seconds": 10,
    "aspect_ratio": "16:9"
  },
  "output": "out/001_sea-sunset_4K_10s_16x9.mp4",
  "change_per_task": ["prompt", "output"],
  "keep_fixed": ["resolution", "aspect_ratio"]
}

跑完后把页面显示的产出一项项核一遍：实际分辨率、实际时长、实际画幅，再对照这份记录。重点确认页面接受的取值写法——画幅是按 16:9 这种比例字符串填，还是按宽高数值填，时长是秒还是毫秒，这类差异只有单条跑完才看得清，也只有在写脚本之前确认才有意义。

抽出参数模板：分辨率、时长、画幅放进一个配置对象

把单条记录里的 config 提到一份独立配置文件，脚本只读这一份。这个动作的价值在于改参数时只需要动一个位置，不必在十几条任务里逐条查找替换。

{
  "resolution": "4K",
  "duration_seconds": 10,
  "aspect_ratio": "16:9",
  "output": {
    "dir": "./out",
    "filename_pattern": "{index:03d}_{prompt_slug}_{resolution}_{duration}s_{ratio}.mp4"
  }
}

几点说明。示例中的字段名只是内部命名，实际提交时以官方文档为准；脚本读取配置后加一层字段映射，把内部字段翻译成接口要求的键名，官方改名只改映射表一处。filename_pattern 用占位符拼文件名，序号、提示词摘要、分辨率、时长、画幅都进文件名，事后不打开文件就能认出产出属于哪条任务。

风险边界：这份配置只覆盖整批一致的参数。如果某条任务需要单独的分辨率或时长，不要临时改进模板，而是在提示词行旁边单列一份覆盖项，覆盖项优先于模板。否则模板会被改得越来越杂，模板和提示词的分工也就失效了。

提示词单独成列表，一行一条任务，与参数模板组合

内容与配置分开之后，新增任务只加一行，参数不动。提示词建议用纯文本，一行一条，空行和以 # 开头的行跳过，方便临时注释掉某条而不删掉原始内容。

# prompts.txt
海边日落，镜头缓慢推近
雨夜街道，霓虹倒映在湿路面，镜头横移
清晨山谷，云层流动，固定机位

读取与合并放在一个小函数里，只做拼装，不发请求，方便单独跑一遍看结果：

def load_prompts(path):
    with open(path, encoding="utf-8") as f:
        return [line.strip() for line in f
                if line.strip() and not line.lstrip().startswith("#")]

def build_jobs(prompts, config):
    jobs = []
    for i, prompt in enumerate(prompts, start=1):
        jobs.append({
            "index": i,
            "prompt": prompt,          # 内容只来自这一行
            "params": dict(config),    # 参数只来自配置对象
        })
    return jobs

需要按实际接口替换的位置集中在两处：一是字段映射，duration 是否要换成毫秒、画幅是否要拆成宽高两个字段；二是提示词是否需要拼接固定前后缀，如果需要，拼接在 build_jobs 里做，不要写回提示词文件，文件里的一行应当始终保持一条原始意图。

提交与回收：任务标识记录、失败重试、结果落盘

批量跑起来之后，只需要保证能回答两个问题：这条产物由哪条任务、哪份参数生成；失败的是哪一条、卡在哪一步。做法是每条任务提交前后都往同一个清单文件追加一行，中途中断直接看最后一行就知道停在哪。

import json, time

RETRY_LIMIT = 2              # 单条任务重试次数上限
CONSECUTIVE_FAIL_LIMIT = 3   # 连续失败达到该值中止整批
LEDGER = "run_ledger.jsonl"

def submit(job):
    # 提交接口：地址、鉴权方式、请求体字段名一律按官方文档填写
    payload = {
        "prompt": job["prompt"],
        # 以下三个键名需按官方文档替换
        "resolution": job["params"]["resolution"],
        "duration": job["params"]["duration_seconds"],
        "aspect_ratio": job["params"]["aspect_ratio"],
    }
    resp = client.create_task(payload)   # 替换为实际 SDK 或 HTTP 调用
    return resp["task_id"]               # 返回字段名按官方文档取值

def run(jobs):
    consecutive_fail = 0
    with open(LEDGER, "a", encoding="utf-8") as ledger:
        for job in jobs:
            snapshot = {"index": job["index"], "prompt": job["prompt"],
                        "params": job["params"], "task_id": None,
                        "status": "pending", "attempt": 0}
            for attempt in range(RETRY_LIMIT + 1):
                snapshot["attempt"] = attempt + 1
                try:
                    snapshot["task_id"] = submit(job)
                    snapshot["status"] = "submitted"
                    break
                except Exception as e:
                    snapshot["status"] = "submit_failed"
                    snapshot["error"] = repr(e)
                    time.sleep(2 * (attempt + 1))  # 退避间隔按实际限流调整
            ledger.write(json.dumps(snapshot, ensure_ascii=False) + "\n")
            ledger.flush()
            consecutive_fail = consecutive_fail + 1 if snapshot["status"] != "submitted" else 0
            if consecutive_fail >= CONSECUTIVE_FAIL_LIMIT:
                break

三处需要按环境调整：RETRY_LIMIT 是单条重试上限，接口限流严格时不要调大；CONSECUTIVE_FAIL_LIMIT 是中止阈值，用来避免整批白跑；退避间隔要结合账号的限流规则确认，代码里的 2 秒只是占位。清单用 JSONL 追加写，一行一条任务快照，含序号、提示词、参数快照、任务标识、状态和尝试次数，事后可以直接筛出失败行。

回收单独写一步：按清单里的任务标识拉取状态与产出地址，结果再追加一行并带上 index 与 task_id，靠这两个字段与提交行关联。不要把提交和回收塞进同一个循环，批量稍大就容易卡在某一条上拖住整批。

验证批量结果：抽样比对单条生成与批量生成是否一致

自动化最怕的情况在于参数被悄悄改了却不报错，报错本身反而不难发现。每批跑完抽两条比一下，随机抽，不要固定抽第一条——第一条往往是最容易被单独照顾的。

• 参数快照与页面单条设置：分辨率、时长、画幅逐项对，快照里缺字段按不一致处理。
• 提交请求体：脚本若打印了请求体，直接和快照对，键名和值都要看。
• 产出文件元数据：用 ffprobe 读宽高与时长，和快照对。
• 文件名：序号是否连续、占位符是否全部替换，有没有残留 {index} 这种没替换的片段。
• 提示词：含逗号、引号、换行的行最容易在拼接时被截断，抽到的两条优先挑这类。

ffprobe -v error -show_entries stream=width,height,duration -of default=noprint_wrappers=1 out/003_sample_4K_10s_16x9.mp4

不一致时按下面的顺序排查，每次都从更靠前的一环开始：

1. 先看快照本身对不对，快照就错了，问题在读配置或合并那一步。
2. 快照对、请求体不对，问题在字段映射，回去改映射表。
3. 请求体与快照都一致、产出不对，检查接口是否用默认值覆盖了未显式传入的字段。
4. 以上都对、只有个别条不一致，回到对应的提示词行，看是否被截断或拼接了额外前后缀。

发现不一致先停下，把整批标记为待复核，不要直接重跑覆盖原有清单。清单是排查依据，重跑前先把旧清单改名留档，再决定是改配置重跑还是逐条补跑。