低遅延なリアルタイム文字起こしモデル「Soniox v3」を試す ①Get Started & 非同期API

import os
import time
import argparse
from typing import Optional
import requests
from requests import Session

SONIOX_API_BASE_URL = "https://api.soniox.com"


# Soniox STT の設定を取得
def get_config(
    audio_url: Optional[str], file_id: Optional[str], translation: Optional[str]
) -> dict:
    config = {
        # 使用するモデルを選択
        # 参考: soniox.com/docs/stt/models
        "model": "stt-async-v3",
        #
        # あらかじめ言語がわかっている場合は、言語ヒントで設定すると、認識精度が大幅に向上する
        # 参考: soniox.com/docs/stt/concepts/language-hints
        "language_hints": ["en", "es"],
        #
        # 言語識別を有効化。各トークンに "language" フィールドが含まれる。
        # 参考: soniox.com/docs/stt/concepts/language-identification
        "enable_language_identification": True,
        #
        # 話者ダイアライゼーション（話者識別）を有効化。各トークンに "speaker" フィールドが含まれる。
        # 参考: soniox.com/docs/stt/concepts/speaker-diarization
        "enable_speaker_diarization": True,
        #
        # コンテキストを設定すると、モデルがドメインを理解し、重要語句の認識や、カスタムの語彙・翻訳設定を適用できる
        # 参考: soniox.com/docs/stt/concepts/context
        "context": {
            "general": [
                {"key": "domain", "value": "Healthcare"},
                {"key": "topic", "value": "Diabetes management consultation"},
                {"key": "doctor", "value": "Dr. Martha Smith"},
                {"key": "patient", "value": "Mr. David Miller"},
                {"key": "organization", "value": "St John's Hospital"},
            ],
            "text": "Mr. David Miller visited his healthcare provider last month for a routine follow-up related to diabetes care. The clinician reviewed his recent test results, noted improved glucose levels, and adjusted his medication schedule accordingly. They also discussed meal planning strategies and scheduled the next check-up for early spring.",
            "terms": [
                "Celebrex",
                "Zyrtec",
                "Xanax",
                "Prilosec",
                "Amoxicillin Clavulanate Potassium",
            ],
            "translation_terms": [
                {"source": "Mr. Smith", "target": "Sr. Smith"},
                {"source": "St John's", "target": "St John's"},
                {"source": "stroke", "target": "ictus"},
            ],
        },
        #
        # このリクエストをトラッキングするための任意の識別子（クライアント定義）。
        # 参考: https://soniox.com/docs/stt/api-reference/transcriptions/create_transcription#request
        "client_reference_id": "MyReferenceId",
        #
        # 音声ソース （いずれか一つのみ指定可能）:
        # - 音声ファイルの公開 URL
        # - 以前にアップロードしたファイルのファイル ID
        # 参考: https://soniox.com/docs/stt/api-reference/transcriptions/create_transcription#request
        "audio_url": audio_url,
        "file_id": file_id,
    }

    # Webhook
    # 文字起こしが完了または失敗したときに通知を受け取るための Webhook を設定できます。
    # 参考: https://soniox.com/docs/stt/api-reference/transcriptions/create_transcription#request

    # 翻訳オプション
    # 参考: soniox.com/docs/stt/rt/real-time-translation#translation-modes
    if translation == "none":
        pass
    elif translation == "one_way":
        # すべての言語をターゲットとなる言語に翻訳
        config["translation"] = {
            "type": "one_way",
            "target_language": "es",
        }
    elif translation == "two_way":
        # language_a から language_b へ、そして language_b から language_a へ再翻訳
        config["translation"] = {
            "type": "two_way",
            "language_a": "en",
            "language_b": "es",
        }
    else:
        raise ValueError(f"Unsupported translation: {translation}")

    return config


def upload_audio(session: Session, audio_path: str) -> str:
    """音声ファイルをアップロードし、ファイル ID を取得する"""
    print("Starting file upload...")
    res = session.post(
        f"{SONIOX_API_BASE_URL}/v1/files",
        files={"file": open(audio_path, "rb")},
    )
    file_id = res.json()["id"]
    print(f"File ID: {file_id}")
    return file_id


def create_transcription(session: Session, config: dict) -> str:
    """文字起こしジョブを作成し、ジョブ ID を返す"""
    print("Creating transcription...")
    try:
        res = session.post(
            f"{SONIOX_API_BASE_URL}/v1/transcriptions",
            json=config,
        )
        res.raise_for_status()
        transcription_id = res.json()["id"]
        print(f"Transcription ID: {transcription_id}")
        return transcription_id
    except Exception as e:
        print("error here:", e)


def wait_until_completed(session: Session, transcription_id: str) -> None:
    """文字起こし完了するまでポーリングして待機する"""
    print("Waiting for transcription...")
    while True:
        res = session.get(f"{SONIOX_API_BASE_URL}/v1/transcriptions/{transcription_id}")
        res.raise_for_status()
        data = res.json()
        if data["status"] == "completed":
            return
        elif data["status"] == "error":
            raise Exception(f"Error: {data.get('error_message', 'Unknown error')}")
        time.sleep(1)


def get_transcription(session: Session, transcription_id: str) -> dict:
    """文字起こし結果を取得する"""
    res = session.get(
        f"{SONIOX_API_BASE_URL}/v1/transcriptions/{transcription_id}/transcript"
    )
    res.raise_for_status()
    return res.json()


def delete_transcription(session: Session, transcription_id: str) -> dict:
    """文字起こしを削除する"""
    res = session.delete(f"{SONIOX_API_BASE_URL}/v1/transcriptions/{transcription_id}")
    res.raise_for_status()


def delete_file(session: Session, file_id: str) -> dict:
    """音声ファイルを削除する"""
    res = session.delete(f"{SONIOX_API_BASE_URL}/v1/files/{file_id}")
    res.raise_for_status()


def delete_all_files(session: Session) -> None:
    """アップロード済みのすべての音声ファイルを削除する"""
    files: list[dict] = []
    cursor: str = ""

    while True:
        print("Getting files...")
        res = session.get(f"{SONIOX_API_BASE_URL}/v1/files?cursor={cursor}")
        res.raise_for_status()
        res_json = res.json()
        files.extend(res_json["files"])
        cursor = res_json["next_page_cursor"]
        if cursor is None:
            break

    total = len(files)
    if total == 0:
        print("No files to delete.")
        return

    print(f"Deleting {total} files...")
    for idx, file in enumerate(files):
        file_id = file["id"]
        print(f"Deleting file: {file_id} ({idx + 1}/{total})")
        delete_file(session, file_id)


def delete_all_transcriptions(session: Session) -> None:
    """完了またはエラー状態の文字起こしをすべて削除する"""
    transcriptions: list[dict] = []
    cursor: str = ""

    while True:
        print("Getting transcriptions...")
        res = session.get(f"{SONIOX_API_BASE_URL}/v1/transcriptions?cursor={cursor}")
        res.raise_for_status()
        res_json = res.json()
        for transcription in res_json["transcriptions"]:
            status = transcription["status"]
            # 完了またはエラーのステータスの文字起こしのみ削除
            if status in ("completed", "error"):
                transcriptions.append(transcription)
        cursor = res_json["next_page_cursor"]
        if cursor is None:
            break

    total = len(transcriptions)
    if total == 0:
        print("No transcriptions to delete.")
        return

    print(f"Deleting {total} transcriptions...")
    for idx, transcription in enumerate(transcriptions):
        transcription_id = transcription["id"]
        print(f"Deleting transcription: {transcription_id} ({idx + 1}/{total})")
        delete_transcription(session, transcription_id)


# トークン列を人間が読みやすい書き起こしに整形する。
def render_tokens(final_tokens: list[dict]) -> str:
    text_parts: list[str] = []
    current_speaker: Optional[str] = None
    current_language: Optional[str] = None

    # トークンを順に処理
    for token in final_tokens:
        text = token["text"]
        speaker = token.get("speaker")
        language = token.get("language")
        is_translation = token.get("translation_status") == "translation"

        # 話者が変わった場合 -> 話者タグを追加
        if speaker is not None and speaker != current_speaker:
            if current_speaker is not None:
                text_parts.append("\n\n")
            current_speaker = speaker
            current_language = None  # 話者が変わったら言語をリセット
            text_parts.append(f"Speaker {current_speaker}:")

        # 言語が変わった場合 -> 言語または翻訳タグを追加
        if language is not None and language != current_language:
            current_language = language
            prefix = "[Translation] " if is_translation else ""
            text_parts.append(f"\n{prefix}[{current_language}] ")
            text = text.lstrip()

        text_parts.append(text)

    return "".join(text_parts)


def transcribe_file(
    session: Session,
    audio_url: Optional[str],
    audio_path: Optional[str],
    translation: Optional[str],
) -> None:
    """音声ファイルを文字起こしして、結果を出力する"""
    if audio_url is not None:
        # 文字起こしする音声ファイルの公開 URL
        assert audio_path is None
        file_id = None
    elif audio_path is not None:
        # ローカルファイルをアップロードして、ファイル ID を取得
        assert audio_url is None
        file_id = upload_audio(session, audio_path)
    else:
        raise ValueError("Missing audio: audio_url or audio_path must be specified.")

    config = get_config(audio_url, file_id, translation)

    transcription_id = create_transcription(session, config)

    wait_until_completed(session, transcription_id)

    result = get_transcription(session, transcription_id)

    text = render_tokens(result["tokens"])
    print(text)

    delete_transcription(session, transcription_id)

    if file_id is not None:
        delete_file(session, file_id)


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--audio_url", help="Public URL of the audio file to transcribe."
    )
    parser.add_argument(
        "--audio_path", help="Path to a local audio file to transcribe."
    )
    parser.add_argument("--delete_all_files", action="store_true")
    parser.add_argument("--delete_all_transcriptions", action="store_true")
    parser.add_argument("--translation", default="none")
    args = parser.parse_args()

    api_key = os.environ.get("SONIOX_API_KEY")
    if not api_key:
        raise RuntimeError(
            "Missing SONIOX_API_KEY.\n"
            "1. Get your API key at https://console.soniox.com\n"
            "2. Run: export SONIOX_API_KEY=<YOUR_API_KEY>"
        )

    # 認証済みセッションを作成
    session = requests.Session()
    session.headers["Authorization"] = f"Bearer {api_key}"

    # アップロード済みのファイルをすべて削除
    if args.delete_all_files:
        delete_all_files(session)
        return

    # すべての文字起こしを削除
    if args.delete_all_transcriptions:
        delete_all_transcriptions(session)
        return

    # 削除しない場合、いずれかの音声ソース指定が必須
    if not (args.audio_url or args.audio_path):
        parser.error("Provide --audio_url or --audio_path (or use a delete flag).")

    transcribe_file(session, args.audio_url, args.audio_path, args.translation)


if __name__ == "__main__":
    main()