Seedance 2.5 is live — 30-second cinematic video with native audio & real-person references
用 Python 批量处理视频升分辨率:API 管道和成本预估
2026/08/14

用 Python 批量处理视频升分辨率:API 管道和成本预估

Python 批处理视频升分辨率的完整指南:80 行代码管道、速率限制预算、成本预估从 $0.002/s、无痛崩溃恢复。

要用 Python 大规模处理视频升分辨率,问题不在单个 API 调用。一个片段是一个 POST 加一个轮询循环,二十行代码,完成。问题在于百个片段的版本:保持在速率限制以下的同时轮询,提交前知道账单,以及恢复一个在第 61 个片段死掉的批次而不为前 60 个片段再次付费。

本指南针对 reAPI 的视频增强端点构建该管道,其中 Topaz Video Upscaler 在标准层级运行 $0.044/s 的源素材,enhance-video-1.0 从 $0.002054/s 起价[1][2]。下面的每个数字和合同细节都来自这些模型在 2026 年 8 月的实时页面。如果你想要工具对比而不是这个(开源、桌面、API),那是一份独立指南:如何用 AI 升分辨率视频

内容速览

  • API 合同本就为批处理而设计: 异步提交立即返回任务 id,计费按服务器探测的源素材秒数,失败的任务自动退款,所以一个崩溃的批次永远不会重复扣费[1]
  • 唯一硬约束是每用户每秒 5 个请求,轮询也算[1]。4 个工作线程每 2.5 秒轮询一次大约用掉那个预算的三分之一,为提交留下余量。
  • 提交前估算: 100 个一分钟片段通过 Topaz 标准层级成本 $264,通过 enhance-video-1.0 标准 1080p 成本 $24.64[1][2]。层级决策的价值是任何代码优化的十倍。
  • 把任务 id 持久化到磁盘成提交状态。 恢复则成本为零:完成的任务已经付费,其输出 URL 仍然可解析。
  • 完整管道大约 80 行,标准库加 requests

API 合同为批处理作业提供什么

三个合同细节塑造了整个设计,全部来自模型自己的参考[1]

提交是异步的。 POST /api/v1/videos/generations 立即返回 {"id", "status": "processing"};渲染在服务器侧发生,你轮询 GET /api/v1/tasks/<id> 直到 statuscompletedfailed。一个批次因此是一组你跟踪的开放任务,而不是你阻塞的队列。

计费跟随源片段,不是输出。 平台在服务器侧探测源视频长度并按其秒数计费:Topaz 升分辨率 $0.044/s 标准或 $0.077/s 最大,enhance-video 的标准层级从 $0.002054/s 到 $0.016429/s 取决于目标分辨率[1][2]。这使得成本成为素材时长的纯函数,也是为什么下面的估算器能工作。

失败自动退款。 信用在提交时保留,任务失败时全额退款[1]。客户端永远不需要补偿逻辑;它只需要记录什么失败了以及决定是否重新提交。

提交前估算账单

时长是唯一重要的输入。如果片段在上传前是本地的,ffprobe 一次调用读取;如果它们已托管,从你自己的元数据取时长。

import subprocess, json

def probe_seconds(path: str) -> float:
    out = subprocess.run(
        ["ffprobe", "-v", "quiet", "-print_format", "json",
         "-show_format", path],
        capture_output=True, text=True, check=True,
    )
    return float(json.loads(out.stdout)["format"]["duration"])

RATES = {                     # $ per second of source, August 2026
    "topaz-standard": 0.044,
    "topaz-max": 0.077,
    "enhance-720p": 0.002054,
    "enhance-1080p": 0.004107,
    "enhance-4k": 0.016429,
}

def estimate(paths, rate_key):
    total = sum(probe_seconds(p) for p in paths)
    return total, total * RATES[rate_key]

100 个一分钟片段的算术看起来像这样:

层级费率100 × 60 s
enhance-video-1.0,标准 720p$0.002054/s$12.32[2]
enhance-video-1.0,标准 1080p$0.004107/s$24.64[2]
enhance-video-1.0,标准 4K$0.016429/s$98.57[2]
Topaz Video Upscaler,标准$0.044/s$264.00[1]
Topaz Video Upscaler,最大$0.077/s$462.00[1]

最便宜行和最昂贵行之间的 37 倍差异是真正的优化表面。没有 Python 能像选择正确层级一样紧缩账单;选择逻辑足够简单可以编码为规则(干净源上升分辨率:enhance 层;降级源需要细节恢复:Topaz)。

用 Python 升分辨率视频:80 行批处理管道

设计:全部提交到前面(提交便宜快速),立即把任务映射持久化到磁盘,然后用小工作线程池轮询。状态存在一个按源 URL 加键的 JSON 文件中。

import json, pathlib, time
from concurrent.futures import ThreadPoolExecutor
import requests

API = "https://reapi.ai/api/v1"
HEADERS = {"Authorization": "Bearer rk_live_..."}
STATE = pathlib.Path("batch_state.json")
POLL_INTERVAL = 2.5   # 4 workers / 2.5s ≈ 1.6 req/s, well under the 5/s cap
WORKERS = 4

def load_state():
    return json.loads(STATE.read_text()) if STATE.exists() else {}

def save_state(state):
    STATE.write_text(json.dumps(state, indent=2))

def submit(video_url, state):
    if video_url in state:          # already submitted on a previous run
        return
    r = requests.post(f"{API}/videos/generations", headers=HEADERS, json={
        "model": "topaz-video-upscaler",
        "video_url": video_url,
        "upscale_factor": "2",
    }, timeout=30)
    r.raise_for_status()
    state[video_url] = {"task_id": r.json()["id"], "status": "processing"}
    save_state(state)               # persist before moving on

def poll_one(video_url, entry):
    while True:
        r = requests.get(f"{API}/tasks/{entry['task_id']}", headers=HEADERS)
        body = r.json()
        if body["status"] in ("completed", "failed"):
            return video_url, body
        time.sleep(POLL_INTERVAL)

def run(video_urls):
    state = load_state()
    for url in video_urls:
        submit(url, state)
        time.sleep(0.25)            # submits: 4/s, inside the budget

    open_items = [
        (u, e) for u, e in state.items() if e["status"] == "processing"
    ]
    with ThreadPoolExecutor(max_workers=WORKERS) as pool:
        for url, body in pool.map(lambda p: poll_one(*p), open_items):
            state[url]["status"] = body["status"]
            if body["status"] == "completed":
                state[url]["output"] = body["output"]
            else:
                state[url]["error"] = body.get("error")
            save_state(state)

    done = sum(1 for e in state.values() if e["status"] == "completed")
    failed = [u for u, e in state.items() if e["status"] == "failed"]
    print(f"{done} completed, {len(failed)} failed")
    for u in failed:
        print("FAILED:", u, state[u].get("error"))

run([
    "https://your-cdn.com/clip-001.mp4",
    "https://your-cdn.com/clip-002.mp4",
])

对 enhance-video-1.0 交换负载(tool_versionsceneresolution 代替 upscale_factor)且没什么别的改变[2]。源 URL 必须是公共 HTTPS;base64 上传被平台广泛拒绝。

速率限制预算,明确

平台将每个用户限制为每秒 5 个请求,轮询也算[1]。管道的支出:

  • 提交:在提交阶段限制到 4/s,那是短的。
  • 轮询:4 个工作线程 × 每 2.5 s 一个请求 = 稳定状态 1.6 req/s。
  • 余量:~3.4 req/s 留给第二个脚本、一个仪表板、或手工 curl 检查。

WORKERS 提高到 12 加 2.5 s 间隔会把轮询单独推到 4.8 req/s 并在任何别的东西接触 API 时开始返回 429。更多工作线程无论如何都不会更快完成渲染;渲染时间在服务器侧。工作线程只约束你多快察觉完成,察觉晚 2.5 秒成本为零。

处理失败不双重付费

平台的失败合同做了重活:一个失败的任务自动全额退款其保留信用[1]。管道的工作降低到簿记:

  1. 记录失败,包括带其 codemessagerequest_iderror 对象(错误代码参考存在于 API 文档[3])。
  2. 不盲目自动重新提交。 一个因源 URL 404 而失败的片段将以零成本再次失败,但重试墙烧掉你的速率预算。修复输入,然后重新运行脚本;状态文件跳过已完成的所有东西。
  3. 信任恢复。 完成的条目保持其输出 URL,指向不过期的 CDN 重托管文件[1],所以在第 61 个片段中断的批次以 60 个已付款结果完整重启且仅剩余部分未完成。

常见问题

我如何免费用 Python 升分辨率视频?

不是通过托管 API;渲染成本某人的 GPU 时间。免费路线是在你自己的 GPU 上运行像 video2x 或 Real-ESRGAN 这样的开源升分辨率器,它交换钱为硬件和设置,在我们的工具对比中涵盖。API 路线从 $0.002054 每源秒起价[2],注册信用涵盖第一次测试调用。

我可以多少个片段并行处理?

想提交多少就提交多少;约束是请求速率,不是开放任务。保持总请求吞吐量、提交加轮询、在每秒 5 以下[1]。4 个轮询工作线程在 2.5 秒间隔是舒适的稳定状态。

升分辨率 100 个视频成本多少?

时长决定。在每个片段一分钟:从 $12.32(enhance-video,720p)到 $462(Topaz 最大)[1][2]。在真实时长上运行估算器在提交前;它是四行 ffprobe。

我的脚本崩溃后我可以恢复批次吗?

是,如果任务 id 在提交时被持久化。完成的任务保持完成和已付费,它们的输出 URL 保持有效,失败的任务已被退款[1]。上面的状态文件模式使恢复成为默认行为而不是恢复功能。

我应该在批处理作业中升分辨率到 4K 吗?

仅当目标屏幕要求它时。4K 层级在 enhance-video 上成本是 720p 层级的 8 倍[2],社交源重新压缩上传无论如何。一个常见模式是成批处理一切在 1080p 并重新运行少数英雄片段在 4K。

我怎么知道平台将计费的源时长?

平台在服务器侧探测托管文件并按其真实长度计费[1]。同一文件上的本地 ffprobe 给出相同数字;差异意味着托管副本不同于本地,这值得在提交百个之前捕获。

运行第一个批次

从三个片段开始,不是百个:一个干净,一个降级,一个长。该运行验证状态文件,显示针对估算的真实每片段成本,当它们成本几美分时表面输入问题。然后指向脚本在完整列表和让退款合同和状态文件吸收无论什么出错。做这个用 Python 的全部要点是升分辨率视频用 Python 第二次,命令就是重新运行脚本,第二次运行仅为第一次未完成的付费。

参考文献

  1. reAPI. Topaz Video Upscaler — model page: live pricing, task lifecycle, rate limits. Retrieved August 2026 from reapi.ai/models/topaz-video-upscaler
  2. reAPI. Enhance Video 1.0 — model page: live per-second tier pricing. Retrieved August 2026 from reapi.ai/models/enhance-video-1-0
  3. reAPI. API error codes reference. Retrieved August 2026 from reapi.ai/docs/api/errors

深入阅读