Pythonで動画高画質化：バッチパイプラインと実コスト

Pythonで動画高画質化を本番スケールで行うとき、問題はAPI呼び出し自体ではありません。1本のクリップなら POST と ポーリングループで20行、完了です。問題は100本版：レート制限を守りながらポーリング、請求額を事前に知ってから送信、クリップ61で落ちたバッチを再開する際に1～60番目のクリップの重複課金を避ける、という3つです。

このガイドではreAPIの動画強化エンドポイントでそのパイプラインを構築します。Topaz Video Upscaler は標準ティアで$0.044/秒、enhance-video-1.0 は$0.002054/秒から始まります^[1][2]。以下のすべての数値と契約条項は2026年8月のモデル公式ページから取得したものです。ツール比較版（オープンソース、デスクトップ、API）をお探しでしたら、別ガイド 動画高画質化の手段を比較する をご覧ください。

TL;DR

• API契約はバッチ向けに設計されています： 非同期サブミットはタスクIDをすぐに返し、課金はサーバー検出の源動画秒数ベース、失敗時は自動返金なので、バッチ落下による二重請求はありません^[1]。
• たった一つの制約は5リクエスト/秒（ポーリング含む）です^[1]。4ワーカーで2.5秒ごとにポーリングすれば約1.6リクエスト/秒、予算の3分の1弱で、サブミット枠も残ります。
• 送信前に見積もります： 1分クリップ100本なら Topaz標準で$264、enhance-video-1.0の標準1080pで$24.64^[1][2]。ティア選択はコード最適化の10倍の価値があります。
• タスクIDをディスクに保存しながら送信します。 再開はタダです：完了済みタスクはすでに支払済みで、出力URLは生きています。
• 完全なパイプラインは約80行、標準ライブラリ + requests だけです。

APIが提供するバッチ向けの契約条項

設計全体を左右する3つの条項があります。すべてモデル公式リファレンスです^[1]：

サブミットは非同期です。 POST /api/v1/videos/generations は {"id", "status": "processing"} をすぐに返し、レンダリングはサーバー側で進行、あなたは GET /api/v1/tasks/<id> を status が completed または failed になるまでポーリングします。バッチは待ったキューではなく、あなたが追跡する開いたタスクの集合です。

課金は源動画に従います。 プラットフォームはホストされた源動画の長さをサーバーで検出し、その秒数で課金します。Topaz高画質化は標準$0.044/秒または最高$0.077/秒、enhance-videoの標準ティアは対象解像度に応じて$0.002054/秒～$0.016429/秒です^[1][2]。コストが映像尺の純粋な関数なのは、下の見積もりが動く理由です。

失敗は自動返金。 クレジットはサブミット時に予約され、タスク失敗時に全額返金されます^[1]。クライアントは償却ロジック不要、失敗を記録し再送するか決めるだけです。

送信前に請求額を見積もる

重要な入力は尺の長さだけです。クリップがローカルなら ffprobe で1回読み取り、すでにホストされていればメタデータから尺を取ります。

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]

1分クリップ100本での計算例：

最安行と最高行の37倍の差が、真の最適化面です。Pythonをどう工夫しても、ティア選択ほど請求額は変わりません。選択ロジックはシンプルで、ルール化できます（クリーンな源なら解像度アップで強化ティア、劣化源なら細部復元にTopaz）。

Python 80行のバッチパイプライン

設計：すべてを最初に送信（サブミットは安く速い）、タスクマップをディスクに即座に保存、小さなワーカープールでポーリングする。状態は源URLをキーにした1つの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_version、scene、resolution を upscale_factor の代わりに）だけで、他は変わりません^[2]。源URLは公開HTTPS必須、base64アップロードはプラットフォーム全体で拒否されます。

レート制限の予算を明示的に

プラットフォームは各ユーザーを5リクエスト/秒で頭打ち、ポーリングを含めます^[1]。パイプラインの消費：

• サブミット：送信フェーズ中4/秒にスロットル（フェーズは短い）。
• ポーリング：4ワーカー × 2.5秒ごと1リクエスト = 定常1.6リクエスト/秒。
• 余裕：~3.4リクエスト/秒残ります（別スクリプト、ダッシュボード、手動curlチェック用）。

WORKERS を12に、2.5秒間隔のままなら、ポーリングだけで4.8リクエスト/秒になり、他がAPI触れば直ぐ429エラー。ワーカー増加は映像処理を速くしません、サーバー側で走ります。ワーカーは完了検知を速くするだけで、2.5秒遅れはコストゼロです。

重複課金なしに失敗を扱う

プラットフォームの失敗契約が重い物を持ち上げます：失敗したタスクは予約クレジットを自動全額返金します^[1]。パイプラインの仕事は簿記に減ります：

1. 失敗を記録します。 エラーオブジェクト（code、message、request_id）と共に（エラーコード参照はAPIドキュメント^[3]）。
2. 盲目的に再送しません。 源URLが404の映像は同じ額（ゼロ）でまた失敗します、だが再試行の壁はレート予算を焼きます。入力を直してスクリプト再実行。状態ファイルは完了済みをスキップします。
3. 再開を信じます。 完了エントリは出力URLを保ちます。CDN再ホストファイルへのポインタで、有効期限はありません^[1]。クリップ61で中断されたバッチは、60件の支払い済み結果をそのまま再開、残りだけ現状不明です。

FAQ

Pythonで無料で動画高画質化できますか？

ホスト型APIではありません。レンダリングはどこかのGPU時間です。無料ルートは自GPU上でvideo2xやReal-ESRGANみたいなオープンソース高画質化ツール、ハードと手間をお金に交換します。ツール比較で詳しく。APIルートは源1秒$0.002054から^[2]、登録クレジットで初期テスト通ります。

並列に何クリップ処理できますか？

好きなだけ送信；制約はリクエストレート、開タスク数じゃありません。総リクエスト、サブミット+ポーリング、5/秒未満に保つ^[1]。4ポーリングワーカー2.5秒間隔は快適定常です。

100本の動画処理コストは？

尺で決まります。1分/本：$12.32（enhance-video 720p）～$462（Topaz最高）^[1][2]。実尺で見積もり実行してから送信、4行ffprobe分。

スクリプトクラッシュ後、バッチ再開できますか？

タスクIDがサブミット時に保存されてれば、はい。完了タスクは完了のまま支払済み、出力URLは生きて、失敗タスクは返金済みです^[1]。上の状態ファイルパターンは、再開を復帰機能でなく既定動作にします。

バッチジョブで4Kにアップスケールすべきですか？

出力画面が要求する場合だけ。4Kティアはenhance-videoで720pの8倍コスト^[2]、ソーシャルフィードは再圧縮。よくあるパターンは全1080p処理、ヒーロー数本を4Kで再実行です。

プラットフォームが請求する源尺をどう知りますか？

プラットフォームはホストファイルをサーバーで検出、実長で課金します^[1]。ローカルffprobe同ファイルなら同じ数字。ズレは、ホスト版がローカル版と異なる意味で、100本送信前にキャッチの価値があります。

初回バッチを実行する

100本でなく3本から：1本クリーンな、1本劣化した、1本長い。その実行は状態ファイル、実クリップあたりのコスト対見積もり、送信前の入力問題を表面化します。その後、スクリプトをフル一覧に指す、返金契約と状態ファイルが何か吸収。Pythonでやる全体のポイントは、Pythonで2回目実行、コマンドはスクリプト再実行で、2回目は1回目が終わらなかった分だけ支払うってことです。

References

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

Further reading

• reAPI. Upscale video with AI: open source, desktop, or API. reapi.ai/blog/upscale-video-with-ai
• reAPI. Topaz Video Upscaler API docs. reapi.ai/docs/topaz-video-upscaler