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

Pythonで動画・YouTubeを文字起こしするREST APIクライアントを実装する（非同期処理・リトライ対応）

0
Posted at

Pythonで動画・YouTubeを文字起こしするREST APIクライアントを実装する（非同期処理・リトライ対応）

動画や音声を扱うアプリでは、文字起こしの本文だけでなく、タイムスタンプ、話者情報、字幕ファイルなどが必要になることがあります。

また、長い動画の文字起こしはすぐには完了しません。そのため、実際のアプリに組み込むには、次のような処理も必要です。

  • 非同期タスクの作成と状態管理
  • タイムアウト時の二重送信防止
  • APIが指定する間隔でのポーリング
  • レート制限や一時障害に対するリトライ
  • プロセス再起動後も結果を取得できる設計

この記事では、公開されている動画URLを文字起こしするREST APIクライアントをPythonで実装します。単にAPIを1回呼ぶだけでなく、非同期APIを実運用に組み込むときに考慮したい点まで扱います。

この記事で使用するAPIについて

実装例には、筆者側で開発している Video Transcriber AIのTranscript API を使用します。

自社・自作サービスの紹介だけにならないよう、この記事では特定サービスに限らず非同期REST APIで応用できる、冪等性、ポーリング、リトライ、状態管理を中心に説明します。

この記事は2026年9月時点の仕様に基づいています。最新のパラメータや制限事項は公式APIドキュメントを確認してください。

作るもの

次の流れを1本のPythonスクリプトとして実装します。

  1. 公開動画URLを指定して文字起こしタスクを作成する
  2. request_id を保存する
  3. サーバーが返す待機時間を使って状態を確認する
  4. 完了後に構造化された結果をJSONとして保存する

Transcript APIは、次の3つのエンドポイントを使う非同期APIです。

POST /transcriptions
  ↓ request_id
GET  /transcriptions/{request_id}
  ↓ succeeded / partial_succeeded
GET  /transcriptions/{request_id}/result

入力には、ログインなしで外部から取得できる公開HTTP/HTTPS URLを使用します。YouTube、TikTok、Instagram、Facebook、X、Bilibiliなどの対応プラットフォームURLのほか、公開されたMP3、MP4、M4A、WAV、WebMなどのURLを指定できます。

OpenAPIはファイルそのもののアップロードには対応していません。 ローカルファイルを処理したい場合は、まずアクセス可能な場所に配置し、その公開URLを source_url に渡す必要があります。

動作環境

この記事では次の環境を想定します。

  • Python 3.10以上
  • requests 2.31以上
  • Windows、macOS、またはLinux
  • Video Transcriber AIで発行したAPIキー
  • ログインなしで取得できる公開動画URL

動作確認用の仮想環境を作成し、requests をインストールします。

python -m venv .venv

macOSまたはLinuxでは、次のコマンドで仮想環境を有効にします。

source .venv/bin/activate
python -m pip install "requests>=2.31,<3"

Windows PowerShellの場合は次のとおりです。

.venv\Scripts\Activate.ps1
python -m pip install "requests>=2.31,<3"

APIキーはソースコードに直接書かず、環境変数に設定します。

macOSまたはLinux：

export VT_API_KEY="your_api_key"

Windows PowerShell：

$env:VT_API_KEY = "your_api_key"

まずはcurlでAPIの流れを確認する

Pythonを書く前に、APIの処理フローをcurlで確認します。

1. 文字起こしタスクを作成する

curl --request POST "https://videotranscriber.ai/openapi/v1/transcriptions" \
  --header "Authorization: Bearer $VT_API_KEY" \
  --header "Content-Type: application/json" \
  --header "Idempotency-Key: qiita-demo-0001" \
  --data '{
    "source_url": "https://example.com/media/demo.mp4",
    "language": "auto",
    "speaker_diarization": true
  }'

作成に成功すると、次のようにタスクIDと推奨待機時間が返ります。

{
  "request_id": "tr_01JEXAMPLE",
  "status": "queued",
  "poll_url": "/transcriptions/tr_01JEXAMPLE",
  "retry_after": 5
}

2. タスクの状態を確認する

curl --request GET \
  "https://videotranscriber.ai/openapi/v1/transcriptions/tr_01JEXAMPLE" \
  --header "Authorization: Bearer $VT_API_KEY"

タスクの主な状態は次のとおりです。

状態 意味
queued 処理待ち
processing 処理中
succeeded すべて成功
partial_succeeded 一部の機能だけ成功
failed 失敗
cancelled キャンセル済み

3. 結果を取得する

succeeded または partial_succeeded になったら、結果を取得します。

curl --request GET \
  "https://videotranscriber.ai/openapi/v1/transcriptions/tr_01JEXAMPLE/result" \
  --header "Authorization: Bearer $VT_API_KEY"

完全なPythonクライアント

次のコードを transcribe.py として保存します。

from __future__ import annotations

import argparse
import json
import os
import random
import time
import uuid
from pathlib import Path
from typing import Any

import requests


BASE_URL = "https://videotranscriber.ai/openapi/v1"
RETRYABLE_STATUS_CODES = {429, 500, 503}
SUCCESS_STATES = {"succeeded", "partial_succeeded"}
TERMINAL_STATES = SUCCESS_STATES | {"failed", "cancelled"}


class TranscriptApiError(RuntimeError):
    """Transcript APIの呼び出しに失敗したときの例外。"""


class TranscriptClient:
    def __init__(self, api_key: str, timeout: float = 30.0) -> None:
        self.timeout = timeout
        self.session = requests.Session()
        self.session.headers.update(
            {
                "Authorization": f"Bearer {api_key}",
                "Accept": "application/json",
            }
        )

    def _request(
        self,
        method: str,
        path: str,
        *,
        max_attempts: int = 5,
        **kwargs: Any,
    ) -> requests.Response:
        """429/500/503だけを指数バックオフ付きで再試行する。"""

        url = f"{BASE_URL}{path}"

        for attempt in range(max_attempts):
            try:
                response = self.session.request(
                    method,
                    url,
                    timeout=self.timeout,
                    **kwargs,
                )
            except (requests.Timeout, requests.ConnectionError):
                if attempt == max_attempts - 1:
                    raise

                delay = min(2**attempt, 30) + random.random()
                print(f"network error: retry in {delay:.1f}s")
                time.sleep(delay)
                continue

            if response.status_code not in RETRYABLE_STATUS_CODES:
                self._raise_for_api_error(response)
                return response

            if attempt == max_attempts - 1:
                self._raise_for_api_error(response)

            retry_after = response.headers.get("Retry-After")
            if retry_after is not None:
                try:
                    delay = float(retry_after)
                except ValueError:
                    delay = min(2**attempt, 30) + random.random()
            else:
                delay = min(2**attempt, 30) + random.random()

            print(
                f"HTTP {response.status_code}: "
                f"retry in {delay:.1f}s"
            )
            time.sleep(delay)

        raise TranscriptApiError("retry loop ended unexpectedly")

    @staticmethod
    def _raise_for_api_error(response: requests.Response) -> None:
        if response.ok:
            return

        try:
            body = response.json()
            error = body.get("error", {})
            code = error.get("code", "unknown_error")
            message = error.get("message", response.text)
        except ValueError:
            code = "invalid_json_response"
            message = response.text

        raise TranscriptApiError(
            f"HTTP {response.status_code} {code}: {message}"
        )

    def create_transcription(
        self,
        source_url: str,
        *,
        idempotency_key: str,
        language: str = "auto",
        speaker_diarization: bool = True,
    ) -> dict[str, Any]:
        payload = {
            "source_url": source_url,
            "language": language,
            "speaker_diarization": speaker_diarization,
        }

        response = self._request(
            "POST",
            "/transcriptions",
            headers={
                "Content-Type": "application/json",
                "Idempotency-Key": idempotency_key,
            },
            json=payload,
        )
        return response.json()

    def get_transcription(self, request_id: str) -> dict[str, Any]:
        response = self._request(
            "GET",
            f"/transcriptions/{request_id}",
        )
        data = response.json()

        # 次のポーリング間隔を決めるため、ヘッダーも保持する。
        data["_retry_after_header"] = response.headers.get("Retry-After")
        return data

    def wait_for_completion(
        self,
        request_id: str,
        *,
        initial_wait: float = 5.0,
        max_wait_seconds: float = 60 * 30,
    ) -> dict[str, Any]:
        started_at = time.monotonic()
        wait_seconds = max(initial_wait, 0.0)

        while True:
            if time.monotonic() - started_at >= max_wait_seconds:
                raise TimeoutError(
                    f"transcription did not finish within "
                    f"{max_wait_seconds:.0f}s: {request_id}"
                )

            time.sleep(wait_seconds)
            task = self.get_transcription(request_id)
            status = task.get("status")
            print(f"request_id={request_id} status={status}")

            if status in TERMINAL_STATES:
                task.pop("_retry_after_header", None)
                return task

            header_wait = task.pop("_retry_after_header", None)
            body_wait = task.get("retry_after")

            try:
                wait_seconds = float(header_wait or body_wait or 5)
            except (TypeError, ValueError):
                wait_seconds = 5.0

            # 不正な値や極端な値で待機し続けないようにする。
            wait_seconds = min(max(wait_seconds, 1.0), 60.0)

    def get_result(self, request_id: str) -> dict[str, Any]:
        response = self._request(
            "GET",
            f"/transcriptions/{request_id}/result",
        )
        return response.json()


def save_json(path: Path, data: dict[str, Any]) -> None:
    path.write_text(
        json.dumps(data, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Transcribe a public media URL"
    )
    parser.add_argument("source_url", help="Public video or audio URL")
    parser.add_argument(
        "--language",
        default="auto",
        help="Source language code (default: auto)",
    )
    parser.add_argument(
        "--output",
        type=Path,
        default=Path("transcript-result.json"),
        help="Output JSON path",
    )
    parser.add_argument(
        "--idempotency-key",
        help="Reuse this value when retrying the same create operation",
    )
    args = parser.parse_args()

    api_key = os.environ.get("VT_API_KEY")
    if not api_key:
        raise SystemExit("VT_API_KEY environment variable is required")

    # 本番では、この値をrequest_idと一緒にDB等へ保存する。
    idempotency_key = args.idempotency_key or str(uuid.uuid4())
    print(f"idempotency_key={idempotency_key}")

    client = TranscriptClient(api_key)
    created = client.create_transcription(
        args.source_url,
        idempotency_key=idempotency_key,
        language=args.language,
        speaker_diarization=True,
    )

    request_id = created["request_id"]
    print(f"created request_id={request_id}")

    try:
        initial_wait = float(created.get("retry_after", 5))
    except (TypeError, ValueError):
        initial_wait = 5.0

    task = client.wait_for_completion(
        request_id,
        initial_wait=initial_wait,
    )

    if task.get("status") not in SUCCESS_STATES:
        save_json(Path("transcript-task-error.json"), task)
        raise TranscriptApiError(
            f"transcription ended with status={task.get('status')}"
        )

    result = client.get_result(request_id)
    save_json(args.output, result)
    print(f"saved: {args.output}")


if __name__ == "__main__":
    main()

実行する

公開動画URLを引数にして実行します。

python transcribe.py "https://www.youtube.com/watch?v=VIDEO_ID"

言語を明示する場合は --language を指定します。

python transcribe.py \
  "https://www.youtube.com/watch?v=VIDEO_ID" \
  --language ja \
  --output result-ja.json

処理中は、次のように状態が表示されます。

idempotency_key=2f80fe19-xxxx-xxxx-xxxx-c8bc25390140
created request_id=tr_01JEXAMPLE
request_id=tr_01JEXAMPLE status=processing
request_id=tr_01JEXAMPLE status=succeeded
saved: result-ja.json

実装のポイント

Idempotency-Keyで二重タスクを防ぐ

POSTリクエストの直後にクライアント側でタイムアウトした場合、次のどちらが起きたのか分からないことがあります。

  1. サーバーにリクエストが届かなかった
  2. サーバーはタスクを作成したが、レスポンスだけ届かなかった

このとき、新しいリクエストとしてPOSTし直すと、同じメディアに対するタスクが二重に作成される可能性があります。

そのため、1つの論理操作には安定した Idempotency-Key を割り当てます。同じキーと同じ本文を再送すると、APIは元のタスクを返します。同じキーで異なる本文を送ると、409 idempotency_conflict になります。

サンプルでは毎回UUIDを生成していますが、実運用では次の情報をまとめて永続化するのが安全です。

自分のジョブID
source_url
Idempotency-Key
APIのrequest_id
現在のstatus
作成日時・更新日時

作成リクエストを再送するときは、新しいUUIDを生成せず、保存済みのキーを再利用します。

retry_afterを尊重してポーリングする

非同期処理だからといって、短い間隔でGETを繰り返す必要はありません。高頻度のポーリングは無駄な通信を増やし、レート制限にも到達しやすくなります。

この実装では、次の優先順位で待機時間を決定しています。

  1. HTTPレスポンスの Retry-After ヘッダー
  2. JSONレスポンスの retry_after
  3. どちらもない場合は5秒

さらに、異常値への対策として1〜60秒の範囲に制限しています。

リトライ可能なエラーを区別する

同じリクエストを再送してよいエラーと、入力や設定を修正すべきエラーを分けます。

HTTPステータス 原因の例 基本方針
400 不正な入力、未対応URL リトライせず入力を修正
401 APIキーがない、または無効 リトライせず認証を修正
409 冪等性キーの競合 キーと本文の対応を確認
429 レートまたはクォータ上限 Retry-After 後に再試行
500 / 503 一時的なサーバー障害 バックオフして再試行

error.message は人間が読むための文言で、将来変わる可能性があります。プログラムの分岐には、安定した error.code またはHTTPステータスを使います。

指数バックオフにjitterを加える

サーバー障害から復旧した瞬間に多数のクライアントが同時に再試行すると、再び負荷が集中します。

そこで、Retry-After がない場合は次のような待機時間を使います。

delay = min(2**attempt, 30) + random.random()

試行ごとに待機時間を増やしつつ、小さな乱数を加えることで再接続のタイミングを分散させます。

partial_succeededを正常系として扱う

文字起こし、話者分離、チャプター、翻訳などを同時に依頼した場合、一部だけ成功する可能性があります。

partial_succeeded を単純な失敗として捨てるのではなく、結果エンドポイントを呼び出し、必要な機能の結果が存在するかを個別に確認します。要求していない機能や失敗した機能は null になる場合があるため、レスポンスの各フィールドを無条件に参照しないようにします。

また、将来レスポンスにフィールドが追加されても壊れないよう、未知のフィールドは無視します。

結果JSONの扱い

結果には、検出言語、全文、メディアの長さ、時間情報付きセグメント、設定時の話者情報などが含まれます。

タイムラインの値は秒単位で、小数を含む場合があります。整数に丸めてから保存すると字幕の同期に影響するため、元の値を維持するのが安全です。

話者分離を無効にした場合や、信頼できる話者ラベルを取得できなかった場合、speaker は null になることがあります。

SRT/VTTのダウンロードURLは署名付きで、有効期間は最大1時間です。URL自体を恒久保存するのではなく、必要なファイルを期限内に取得して、自分のストレージへ保存します。

実運用に入れる前のチェックリスト

APIキーをブラウザに置かない

Bearer APIキーは、ReactやVueなどのフロントエンドコードに埋め込まず、自分のバックエンドからAPIを呼び出します。

次の場所にもキーを残さないよう注意します。

  • Gitリポジトリ
  • アプリケーションログ
  • エラー監視サービス
  • スクリーンショット
  • サポートへの問い合わせ本文

入力URLを検証する

APIに渡すURLは、外部からアクセス可能である必要があります。

  • ログインが必要なURL
  • Cookieが必要なURL
  • 社内ネットワーク内のURL
  • 期限切れのクラウド共有URL
  • 認証情報をクエリ文字列に含むURL

これらは避けます。ユーザーが任意のURLを指定できるサービスでは、自分の側でも許可するスキームやドメインを検証し、URLをログへそのまま出さない設計を検討します。

request_idを永続化する

サンプルコードは1プロセス内で完了まで待ちますが、長時間の処理をWebリクエスト内で待つ構成は避けた方が安全です。

本番では、例えば次のように分離できます。

Web API
  └─ 文字起こしタスクを作成してrequest_idをDBへ保存

Worker / Queue
  └─ 状態を定期確認
       ├─ processing → 次回確認を予約
       ├─ succeeded → 結果取得
       └─ failed → エラー情報を保存

プロセスが停止しても、保存済みの request_id から処理を再開できます。

保持期間を前提に結果を保存する

API上の文字起こし結果は生成から30日間取得できます。タスクレコードは作成月と、その翌月までUTC基準で保持されます。

恒久的に必要なデータは、処理完了後に自分のデータベースやオブジェクトストレージへ保存します。

利用量を切り上げで見積もる

基本の文字起こし利用量は、メディアの秒数を60秒単位で切り上げて計算されます。

billable_minutes = ceil(duration_seconds / 60)

例えば61秒の動画は2分として計算されます。チャプター生成や翻訳は追加のクォータを使用するため、大量処理では機能ごとに見積もります。

まとめ

公開動画URLを文字起こしするPythonクライアントを実装しました。

非同期の動画文字起こしAPIを安定して運用するうえで重要なのは、単にPOSTとGETを実装することではありません。

  • Idempotency-Key で二重タスクを防ぐ
  • request_id を保存して処理を再開可能にする
  • retry_after と Retry-After を尊重する
  • 400/401/409と429/500/503を分けて扱う
  • 指数バックオフとjitterで一時障害に対応する
  • partial_succeeded とフィールド単位の失敗を想定する
  • APIキー、署名付きURL、結果保持期間を意識する

今回は Video Transcriber AIのTranscript API を例にしました。対応URL、リクエストフィールド、レスポンススキーマ、エラーコードの最新情報は、Transcript API Documentationを参照してください。

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?