0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@zhangzhaoyu1204

asyncioで複数のYouTube動画を一括文字起こしする｜Semaphoreによる並列数制御と再開処理

0
Posted at

asyncioで複数のYouTube動画を一括文字起こしする｜Semaphoreによる並列数制御と再開処理

複数のYouTube動画をPythonで一括文字起こししたい場合、URLをfor文で順番に処理するだけでは時間がかかります。一方、すべてのリクエストを無制限に並列化すると、APIのレート制限に達して429 Too Many Requestsが発生しやすくなります。

この記事では、asyncio、httpx.AsyncClient、asyncio.Semaphoreを使い、複数のYouTube URLを安全に並列処理するバッチを実装します。

単に並列化するだけでなく、以下にも対応します。

  • CSVから複数のYouTube URLを読み込む
  • SemaphoreでAPIへの同時リクエスト数を制御する
  • 非同期の文字起こしタスクを個別にポーリングする
  • Retry-Afterを優先して429／500／503を再試行する
  • Idempotency-Keyでタスクの重複作成を防ぐ
  • JSONLへ進捗を追記し、途中終了後も再開する
  • 成功・失敗件数を最後に集計する

TL;DR

今回の構成は次のとおりです。

urls.csv
   ↓
asyncioで動画ごとの処理を起動
   ↓
Semaphoreで同時APIリクエスト数を制限
   ↓
文字起こしタスク作成
   ↓
retry_afterに従って非同期ポーリング
   ↓
results/{item_id}.jsonへ保存
   ↓
state.jsonlを使って中断位置から再開

重要なのは、Semaphoreを「動画処理全体」ではなく「HTTPリクエストを送る瞬間」にだけ適用することです。ポーリングの待機中まで枠を占有すると、他の動画が必要以上に待たされます。

想定するユースケース

  • YouTubeチャンネル内の技術動画をまとめて検索可能にする
  • 講義やウェビナーの文字起こしを保存する
  • 複数動画から字幕や要約を後続処理で作る
  • 動画アーカイブをJSONLやデータベースへ蓄積する
  • RAG用の元データを定期的に生成する

今回は公開URLを送るだけで処理できる複数の動画URLを同じ形式で処理できるTranscript APIを使用します。

このAPIは非同期方式です。POST /transcriptionsでタスクを作成し、返されたrequest_idを使って状態を確認し、完了後にGET /transcriptions/{request_id}/resultから結果を取得します。

動作環境

  • Python 3.11以上
  • httpx
  • python-dotenv

プロジェクトを作成します。

mkdir youtube-transcription-batch
cd youtube-transcription-batch

python -m venv .venv

# macOS / Linux
source .venv/bin/activate

# Windows PowerShell
.venv\Scripts\Activate.ps1

pip install httpx python-dotenv

ファイル構成は次のとおりです。

youtube-transcription-batch/
├── .env
├── .gitignore
├── urls.csv
├── batch_transcribe.py
├── state.jsonl
└── results/

.envへAPI Keyと最大同時リクエスト数を設定します。

VT_API_KEY=your_api_key
MAX_CONCURRENT_REQUESTS=3

API KeyをGitへ含めないようにします。

.env
.venv/
__pycache__/
state.jsonl
results/

入力CSVを用意する

urls.csvには、処理を識別するitem_idとYouTube URLを保存します。

item_id,source_url
video-001,https://www.youtube.com/watch?v=VIDEO_ID_1
video-002,https://youtu.be/VIDEO_ID_2
video-003,https://www.youtube.com/watch?v=VIDEO_ID_3

item_idは出力ファイル名と再開判定に使います。同じIDへ別のURLを割り当てないでください。

実装の全体像

先に完成版を掲載します。あとから重要な部分を分解して説明します。

# batch_transcribe.py
from __future__ import annotations

import asyncio
import csv
import hashlib
import json
import os
import random
import re
from dataclasses import dataclass
from pathlib import Path
from typing import Any

import httpx
from dotenv import load_dotenv

load_dotenv()

API_BASE = "https://videotranscriber.ai/openapi/v1"
INPUT_CSV = Path("urls.csv")
STATE_FILE = Path("state.jsonl")
RESULT_DIR = Path("results")

TERMINAL_STATUSES = {
    "succeeded",
    "partial_succeeded",
    "failed",
    "cancelled",
}
SUCCESS_STATUSES = {"succeeded", "partial_succeeded"}
RETRYABLE_STATUS_CODES = {429, 500, 503}


@dataclass(frozen=True)
class VideoItem:
    item_id: str
    source_url: str


class StateStore:
    """JSONLの最終行をitem_idごとの最新状態として扱う。"""

    def __init__(self, path: Path) -> None:
        self.path = path
        self.lock = asyncio.Lock()
        self.latest: dict[str, dict[str, Any]] = self._load()

    def _load(self) -> dict[str, dict[str, Any]]:
        latest: dict[str, dict[str, Any]] = {}
        if not self.path.exists():
            return latest

        for line in self.path.read_text(encoding="utf-8").splitlines():
            if not line.strip():
                continue
            record = json.loads(line)
            latest[record["item_id"]] = record

        return latest

    async def append(self, record: dict[str, Any]) -> None:
        # 複数コルーチンが同時に1行を書かないようLockで保護する
        async with self.lock:
            with self.path.open("a", encoding="utf-8") as file:
                file.write(json.dumps(record, ensure_ascii=False) + "\n")
            self.latest[record["item_id"]] = record


class TranscriptApiClient:
    def __init__(
        self,
        api_key: str,
        max_concurrent_requests: int,
    ) -> None:
        self.semaphore = asyncio.Semaphore(max_concurrent_requests)
        self.client = httpx.AsyncClient(
            base_url=API_BASE,
            headers={"Authorization": f"Bearer {api_key}"},
            timeout=httpx.Timeout(30.0),
        )

    async def __aenter__(self) -> "TranscriptApiClient":
        return self

    async def __aexit__(self, *args: Any) -> None:
        await self.client.aclose()

    @staticmethod
    def _retry_delay(response: httpx.Response, attempt: int) -> float:
        retry_after = response.headers.get("Retry-After")
        if retry_after:
            try:
                return max(float(retry_after), 1.0)
            except ValueError:
                pass

        # 1, 2, 4, 8...秒を上限30秒にし、同時再送を避けるjitterを加える
        return min(2 ** (attempt - 1), 30) + random.uniform(0, 0.5)

    async def request_json(
        self,
        method: str,
        path: str,
        *,
        max_attempts: int = 5,
        **kwargs: Any,
    ) -> tuple[dict[str, Any], httpx.Response]:
        for attempt in range(1, max_attempts + 1):
            try:
                # 待機時間ではなく、HTTP通信中だけSemaphoreを取得する
                async with self.semaphore:
                    response = await self.client.request(
                        method,
                        path,
                        **kwargs,
                    )
            except httpx.TransportError:
                if attempt == max_attempts:
                    raise
                await asyncio.sleep(
                    min(2 ** (attempt - 1), 30)
                    + random.uniform(0, 0.5)
                )
                continue

            if response.status_code in RETRYABLE_STATUS_CODES:
                if attempt == max_attempts:
                    response.raise_for_status()

                delay = self._retry_delay(response, attempt)
                print(
                    f"retry status={response.status_code} "
                    f"attempt={attempt} sleep={delay:.1f}s"
                )
                await asyncio.sleep(delay)
                continue

            response.raise_for_status()
            return response.json(), response

        raise RuntimeError("unreachable")

    async def create_task(
        self,
        item: VideoItem,
    ) -> dict[str, Any]:
        # 同じitem_idとURLなら再実行時も同じキーになる
        source = f"{item.item_id}:{item.source_url}".encode("utf-8")
        digest = hashlib.sha256(source).hexdigest()[:32]

        body, _ = await self.request_json(
            "POST",
            "/transcriptions",
            headers={"Idempotency-Key": f"batch-{digest}"},
            json={
                "source_url": item.source_url,
                "language": "auto",
                "speaker_diarization": False,
            },
        )
        return body

    async def get_status(
        self,
        request_id: str,
    ) -> tuple[dict[str, Any], httpx.Response]:
        return await self.request_json(
            "GET",
            f"/transcriptions/{request_id}",
        )

    async def get_result(self, request_id: str) -> dict[str, Any]:
        body, _ = await self.request_json(
            "GET",
            f"/transcriptions/{request_id}/result",
        )
        return body


def load_items(path: Path) -> list[VideoItem]:
    items: list[VideoItem] = []
    used_ids: set[str] = set()

    with path.open(newline="", encoding="utf-8-sig") as file:
        reader = csv.DictReader(file)

        for row in reader:
            item_id = (row.get("item_id") or "").strip()
            source_url = (row.get("source_url") or "").strip()

            if not re.fullmatch(r"[A-Za-z0-9_-]+", item_id):
                raise ValueError(f"Invalid item_id: {item_id!r}")
            if item_id in used_ids:
                raise ValueError(f"Duplicated item_id: {item_id}")
            if not source_url.startswith("https://"):
                raise ValueError(f"Invalid source_url: {source_url!r}")

            used_ids.add(item_id)
            items.append(VideoItem(item_id, source_url))

    return items


def polling_delay(
    status_body: dict[str, Any],
    response: httpx.Response,
) -> float:
    value = (
        status_body.get("retry_after")
        or response.headers.get("Retry-After")
        or 5
    )
    try:
        return max(float(value), 1.0)
    except (TypeError, ValueError):
        return 5.0


async def transcribe_one(
    item: VideoItem,
    api: TranscriptApiClient,
    state: StateStore,
) -> dict[str, Any]:
    RESULT_DIR.mkdir(parents=True, exist_ok=True)
    result_path = RESULT_DIR / f"{item.item_id}.json"
    previous = state.latest.get(item.item_id, {})

    # 結果保存まで完了していればAPIを呼ばない
    if (
        previous.get("status") in SUCCESS_STATUSES
        and result_path.exists()
    ):
        print(f"skip item_id={item.item_id}")
        return {
            "item_id": item.item_id,
            "status": "skipped",
            "result_path": str(result_path),
        }

    request_id = previous.get("request_id")

    if not request_id:
        task = await api.create_task(item)
        request_id = task["request_id"]

        await state.append({
            "item_id": item.item_id,
            "source_url": item.source_url,
            "request_id": request_id,
            "status": task.get("status", "queued"),
        })

        delay = max(float(task.get("retry_after", 5)), 1.0)
    else:
        # 前回保存したrequest_idからポーリングを再開する
        delay = 1.0

    while True:
        await asyncio.sleep(delay)
        status_body, status_response = await api.get_status(request_id)
        status = status_body["status"]

        print(
            f"item_id={item.item_id} "
            f"request_id={request_id} status={status}"
        )

        await state.append({
            "item_id": item.item_id,
            "source_url": item.source_url,
            "request_id": request_id,
            "status": status,
        })

        if status in TERMINAL_STATUSES:
            break

        delay = polling_delay(status_body, status_response)

    if status not in SUCCESS_STATUSES:
        return {
            "item_id": item.item_id,
            "status": status,
            "request_id": request_id,
        }

    result = await api.get_result(request_id)
    result_path.write_text(
        json.dumps(result, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )

    await state.append({
        "item_id": item.item_id,
        "source_url": item.source_url,
        "request_id": request_id,
        "status": status,
        "result_path": str(result_path),
    })

    return {
        "item_id": item.item_id,
        "status": status,
        "result_path": str(result_path),
    }


async def main() -> None:
    api_key = os.environ.get("VT_API_KEY")
    if not api_key:
        raise RuntimeError("VT_API_KEY is not set")

    max_concurrent = int(
        os.environ.get("MAX_CONCURRENT_REQUESTS", "3")
    )
    if max_concurrent < 1:
        raise ValueError("MAX_CONCURRENT_REQUESTS must be >= 1")

    items = load_items(INPUT_CSV)
    state = StateStore(STATE_FILE)

    async with TranscriptApiClient(
        api_key,
        max_concurrent,
    ) as api:
        results = await asyncio.gather(
            *(transcribe_one(item, api, state) for item in items),
            return_exceptions=True,
        )

    succeeded = 0
    failed = 0

    for item, result in zip(items, results, strict=True):
        if isinstance(result, Exception):
            failed += 1
            print(
                f"ERROR item_id={item.item_id} "
                f"type={type(result).__name__} detail={result}"
            )
        elif result["status"] in SUCCESS_STATUSES | {"skipped"}:
            succeeded += 1
        else:
            failed += 1

    print(
        f"finished total={len(items)} "
        f"succeeded={succeeded} failed={failed}"
    )


if __name__ == "__main__":
    asyncio.run(main())

実行します。

python batch_transcribe.py

処理が完了すると、動画ごとのJSONがresults/へ保存されます。

results/
├── video-001.json
├── video-002.json
└── video-003.json

asyncio.Semaphoreで並列数を制御する

asyncio.gather()だけで全動画を実行すると、動画数と同じだけ処理が同時に開始されます。

await asyncio.gather(
    *(transcribe_one(item) for item in items)
)

10本程度なら問題が見えなくても、数百本になると短時間にリクエストが集中します。

そこで、HTTP通信部分だけをSemaphoreで囲みます。

self.semaphore = asyncio.Semaphore(3)

async with self.semaphore:
    response = await self.client.request(method, path, **kwargs)

この書き方なら、動画単位のコルーチンは並行して動きつつ、外部APIへ同時送信されるリクエストは最大3件になります。

動画処理全体をSemaphoreで囲まない理由

次のように、タスク作成から完了までをすべてSemaphore内へ入れる方法もあります。

async with semaphore:
    await create_task()
    await poll_until_done()
    await get_result()

しかし文字起こし中の大部分は、API側の処理を待っている時間です。待機中の動画がSemaphoreを保持すると、空いている通信時間を他の動画が使えません。

今回は「同時処理動画数」ではなく「同時HTTPリクエスト数」を制限しています。

なお、契約上の同時処理タスク数にも制限がある場合は、別のSemaphoreをtranscribe_one()全体へ追加し、2段階で制御します。

429エラーとRetry-Afterを処理する

429 Too Many Requestsを受け取った直後に再送すると、さらに429が続く可能性があります。

優先順位は次のようにします。

  1. Retry-Afterがあれば、その時間だけ待つ
  2. なければ指数バックオフを使う
  3. 複数タスクの同時再送を避けるためjitterを加える
return min(2 ** (attempt - 1), 30) + random.uniform(0, 0.5)

指数バックオフだけでは、同時に失敗したタスクが1秒後、2秒後、4秒後に再び一斉送信されることがあります。小さな乱数を加えることで再送時刻をずらします。

POSTの再試行にはIdempotency-Keyが必要

ネットワークタイムアウトが発生したとき、サーバー側ではタスク作成に成功しているのに、クライアントだけがレスポンスを受け取れていない場合があります。

そのまま別リクエストとしてPOSTすると、同じ動画のタスクが二つ作成される可能性があります。

今回のコードでは、item_idとsource_urlから安定したキーを生成します。

source = f"{item.item_id}:{item.source_url}".encode("utf-8")
digest = hashlib.sha256(source).hexdigest()[:32]
idempotency_key = f"batch-{digest}"

同一の論理操作を再試行するときは同じキーを使い、意図的に新しい文字起こしを作るときはitem_idを変えます。

state.jsonlで途中から再開する

動画が多いと、PCの再起動やネットワーク障害によってバッチが途中で停止することがあります。

state.jsonlには状態変化を追記します。

{"item_id":"video-001","request_id":"tr_xxx","status":"queued"}
{"item_id":"video-001","request_id":"tr_xxx","status":"processing"}
{"item_id":"video-001","request_id":"tr_xxx","status":"succeeded","result_path":"results/video-001.json"}

再起動時にはitem_idごとの最後の行を読みます。

  • request_idがある：既存タスクのポーリングを再開
  • 成功済みで結果ファイルもある：スキップ
  • 状態がない：新しいタスクを作成

状態ファイルを毎回全体書き換えせず追記型にしているため、途中でプロセスが落ちても、それ以前の記録を残しやすくなります。

長期間運用する場合は、定期的に最新状態だけを別ファイルへ圧縮するか、SQLiteやPostgreSQLへ移行します。

gatherで一件の失敗を全体へ波及させない

デフォルトのasyncio.gather()では、一つの処理で例外が発生すると、呼び出し元でも例外になります。

バッチ処理では、一つの非公開動画や無効URLによって、ほかの正常な動画まで結果集計できなくなるのは困ります。

results = await asyncio.gather(
    *tasks,
    return_exceptions=True,
)

return_exceptions=Trueを指定して、例外を各動画の結果として回収します。最後に成功件数と失敗件数を集計すれば、失敗したURLだけを修正できます。

よくある失敗

ポーリングを1秒固定で繰り返す

タスクのretry_afterやHTTPのRetry-Afterを無視すると、不要なリクエストが増えます。APIが示した待機時間を優先します。

Semaphoreを作っただけで使っていない

Semaphoreはインスタンスを作成するだけでは制限されません。必ず通信部分をasync with semaphore:で囲みます。

リトライのたびにIdempotency-Keyを変更する

キーを毎回変えると、APIからは別の作成操作に見えます。同じPOSTの再送では、同じキーを再利用します。

API KeyをCSVやログへ出力する

API Keyは環境変数に置きます。Authorizationヘッダー、署名付きURL、認証情報を含むURLをログへ出さないようにします。

同じitem_idへ別URLを割り当てる

断点再開時に古いrequest_idと新しいURLが混ざります。item_idは処理対象に対して固定し、URLを変えるときは新しいIDを使います。

本番運用へ進めるなら

今回のJSONL方式は、1台のPCで動かす小規模バッチに向いています。本番環境では次の改善を検討できます。

  • 状態管理をSQLite／PostgreSQLへ移す
  • 結果JSONをS3互換ストレージへ保存する
  • Prometheusなどで429率と処理時間を監視する
  • URL単位の重複チェックを追加する
  • SIGINT／SIGTERMを受けたときに安全に終了する
  • 失敗理由ごとに再実行可否を分類する
  • 実行ごとにバッチIDを付ける

また、入力URLは公開HTTP／HTTPS URLである必要があります。ログインが必要な動画、期限切れ共有URL、社内ネットワーク限定URLは、外部APIから取得できない可能性があります。

まとめ

複数のYouTube動画をPythonで一括文字起こしするときは、単にasyncio.gather()で並列化するだけでは不十分です。

今回のポイントは次の5つです。

  1. httpx.AsyncClientでI/O待ちを並行化する
  2. asyncio.SemaphoreをHTTP通信部分へ適用する
  3. 429／500／503ではRetry-Afterと指数バックオフを使う
  4. Idempotency-KeyでPOSTの重複実行を防ぐ
  5. state.jsonlへ進捗を保存し、中断後に再開する

この構成なら、動画数が増えても無制限にリクエストを送らず、失敗した一件だけを切り分けられます。

まずはMAX_CONCURRENT_REQUESTS=2または3から始め、利用するAPIの制限と実際の429発生率を確認しながら調整するのがおすすめです。

※APIの仕様、利用制限、料金は変更される場合があります。実装時には最新の公式ドキュメントを確認してください。また、動画の利用条件と著作権を確認し、処理する権限のあるコンテンツで使用してください。

Qiita投稿時のタグ

Python asyncio YouTube API 文字起こし

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?