transcribe_qwen_asr_api.py ダウンロード/コピー

transcribe_qwen_asr_api.py をダウンロード

transcribe_qwen_asr_api.py
transcribe_qwen_asr_api.py
  1#!/usr/bin/env python3
  2"""
  3概要:
  4    ローカルのQwen3-ASRサーバーを使用して音声ファイルを文字起こしします。
  5
  6詳細説明:
  7    入力ファイルをFFmpegでデコードし、重複を持たせたWAVチャンクに分割して、OpenAI互換の音声認識APIエンドポイントに送信します。チャンク処理が完了するごとにチェックポイントを保存するため、長時間の文字起こしを中断後に再開することができます。
  8"""
  9
 10from __future__ import annotations
 11
 12import argparse
 13import json
 14import math
 15import subprocess
 16import sys
 17import tempfile
 18import time
 19import urllib.error
 20import urllib.request
 21import uuid
 22from pathlib import Path
 23from typing import Any
 24
 25DEFAULT_API_BASE = "http://192.168.27.18:8001/v1"
 26DEFAULT_MODEL = "Qwen/Qwen3-ASR-1.7B"
 27
 28
 29def run_command(command: list[str]) -> str:
 30    """
 31    概要:
 32        外部コマンドを実行し標準出力を返します。
 33
 34    詳細説明:
 35        subprocessを使用してコマンドを実行します。失敗した場合はエラーメッセージを含めた例外を発生させます。
 36
 37    引数:
 38        :param command: 実行するコマンドとその引数のリスト
 39        :type command: list
 40
 41    戻り値:
 42        :returns: コマンドの標準出力文字列
 43        :rtype: str
 44
 45    例外:
 46        :raises RuntimeError: コマンドが見つからないか実行に失敗した場合
 47    """
 48    try:
 49        completed = subprocess.run(
 50            command, check=True, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE
 51        )
 52    except FileNotFoundError as exc:
 53        raise RuntimeError(f"'{command[0]}' が見つかりません。FFmpeg を PATH に追加してください。") from exc
 54    except subprocess.CalledProcessError as exc:
 55        details = exc.stderr.strip() or exc.stdout.strip() or "詳細なエラー出力はありません。"
 56        rendered = subprocess.list2cmdline(command)
 57        raise RuntimeError(f"FFmpeg コマンドが失敗しました:\n{rendered}\n\n{details}") from exc
 58    return completed.stdout.strip()
 59
 60
 61def media_duration_seconds(path: Path) -> float:
 62    """
 63    概要:
 64        音声ファイルの長さを秒単位で取得します。
 65
 66    詳細説明:
 67        ffprobeコマンドを使用してメディアの長さを取得します。
 68
 69    引数:
 70        :param path: 音声ファイルのパス
 71        :type path: pathlib.Path
 72
 73    戻り値:
 74        :returns: メディアの長さの秒数
 75        :rtype: float
 76
 77    例外:
 78        :raises RuntimeError: 長さの取得に失敗した場合
 79    """
 80    text = run_command(
 81        [
 82            "ffprobe", "-v", "error", "-show_entries", "format=duration",
 83            "-of", "default=noprint_wrappers=1:nokey=1", str(path),
 84        ]
 85    )
 86    try:
 87        return float(text)
 88    except ValueError as exc:
 89        raise RuntimeError(f"音声長を取得できませんでした: {text!r}") from exc
 90
 91
 92def extract_chunk(source: Path, start: float, duration: float, destination: Path) -> None:
 93    """
 94    概要:
 95        音声ファイルの指定区間を抽出してWAV形式で保存します。
 96
 97    詳細説明:
 98        ffmpegを使用して、指定した開始位置から指定した長さの音声を16kHzのモノラルWAVファイルとして切り出します。
 99
100    引数:
101        :param source: 入力音声ファイルのパス
102        :type source: pathlib.Path
103        :param start: 切り出し開始位置の秒数
104        :type start: float
105        :param duration: 切り出す長さの秒数
106        :type duration: float
107        :param destination: 出力WAVファイルのパス
108        :type destination: pathlib.Path
109
110    戻り値:
111        :returns: なし
112        :rtype: None
113    """
114    run_command(
115        [
116            "ffmpeg", "-hide_banner", "-loglevel", "error", "-y",
117            "-ss", f"{start:.3f}", "-t", f"{duration:.3f}", "-i", str(source),
118            "-vn", "-map", "0:a:0", "-ac", "1", "-ar", "16000", "-c:a", "pcm_s16le",
119            str(destination),
120        ]
121    )
122
123
124def merge_overlap(previous: str, current: str, max_chars: int = 240) -> str:
125    """
126    概要:
127        重複する音声チャンクによって生じたテキストの重複部分を削除して結合します。
128
129    詳細説明:
130        前回のテキストの末尾と今回のテキストの先頭を比較し、一致する重複部分を取り除いた今回のテキストを返します。
131
132    引数:
133        :param previous: 前回の文字起こしテキスト
134        :type previous: str
135        :param current: 今回の文字起こしテキスト
136        :type current: str
137        :param max_chars: 重複を判定する最大文字数
138        :type max_chars: int
139
140    戻り値:
141        :returns: 重複部分を削除した今回のテキスト
142        :rtype: str
143    """
144    if not previous:
145        return current.strip()
146    current = current.strip()
147    if not current:
148        return ""
149
150    # Text is often Japanese without spaces, so compare characters rather than words.
151    limit = min(len(previous), len(current), max_chars)
152    for length in range(limit, 5, -1):
153        if previous[-length:] == current[:length]:
154            return current[length:].lstrip()
155    return current
156
157
158def save_json(path: Path, payload: dict[str, Any]) -> None:
159    """
160    概要:
161        辞書データをJSONファイルとして安全に保存します。
162
163    詳細説明:
164        一度一時ファイルとして書き出してから目的のパスにリネームすることで、保存中の中断によるデータ破損を防ぎます。
165
166    引数:
167        :param path: 保存先のJSONファイルのパス
168        :type path: pathlib.Path
169        :param payload: 保存する辞書データ
170        :type payload: dict
171
172    戻り値:
173        :returns: なし
174        :rtype: None
175    """
176    temporary = path.with_suffix(path.suffix + ".tmp")
177    temporary.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
178    temporary.replace(path)
179
180
181def transcribe_chunk(api_base: str, model: str, wav_path: Path, timeout: float) -> tuple[str, dict[str, Any]]:
182    """
183    概要:
184        WAVファイルを音声認識APIに送信し文字起こし結果を取得します。
185
186    詳細説明:
187        multipart/form-data形式でWAVファイルとモデル名をPOSTリクエストで送信し、結果をJSONとして受け取ります。
188
189    引数:
190        :param api_base: APIのエンドポイントのベースURL
191        :type api_base: str
192        :param model: 使用する音声認識モデルの名前
193        :type model: str
194        :param wav_path: 送信するWAVファイルのパス
195        :type wav_path: pathlib.Path
196        :param timeout: APIリクエストのタイムアウト秒数
197        :type timeout: float
198
199    戻り値:
200        :returns: 文字起こしテキストとAPI応答の辞書のタプル
201        :rtype: tuple
202
203    例外:
204        :raises RuntimeError: APIへの接続や応答処理に失敗した場合
205    """
206    endpoint = api_base.rstrip("/") + "/audio/transcriptions"
207    boundary = "----QwenASR" + uuid.uuid4().hex
208    with wav_path.open("rb") as audio:
209        audio_bytes = audio.read()
210    body = b"".join(
211        [
212            f"--{boundary}\r\nContent-Disposition: form-data; name=\"model\"\r\n\r\n{model}\r\n".encode(),
213            f"--{boundary}\r\nContent-Disposition: form-data; name=\"file\"; filename=\"{wav_path.name}\"\r\nContent-Type: audio/wav\r\n\r\n".encode(),
214            audio_bytes,
215            f"\r\n--{boundary}--\r\n".encode(),
216        ]
217    )
218    request = urllib.request.Request(
219        endpoint, data=body,
220        headers={"Content-Type": f"multipart/form-data; boundary={boundary}"}, method="POST",
221    )
222    try:
223        with urllib.request.urlopen(request, timeout=timeout) as response:
224            response_text = response.read().decode("utf-8", errors="replace")
225    except urllib.error.HTTPError as exc:
226        detail = exc.read().decode("utf-8", errors="replace")[:1000]
227        raise RuntimeError(f"ASR API エラー ({exc.code}): {detail}") from exc
228    except urllib.error.URLError as exc:
229        raise RuntimeError(f"ASR API に接続できません: {exc.reason}") from exc
230    try:
231        payload = json.loads(response_text)
232        text = payload["text"]
233    except (ValueError, KeyError, TypeError) as exc:
234        raise RuntimeError(f"ASR API の応答形式が不正です: {response_text[:1000]}") from exc
235    return str(text).strip(), payload
236
237
238def parse_args() -> argparse.Namespace:
239    """
240    概要:
241        コマンドライン引数を解析します。
242
243    詳細説明:
244        入力ファイルやAPI設定、チャンク設定などの引数を解析して返します。
245
246    戻り値:
247        :returns: 解析されたコマンドライン引数
248        :rtype: argparse.Namespace
249    """
250    parser = argparse.ArgumentParser(description="Qwen3-ASR API で WAV/MP3/MP4 を文字起こしします。")
251    parser.add_argument("input", type=Path, help="入力 WAV / MP3 / MP4 ファイル")
252    parser.add_argument("--api-base", default=DEFAULT_API_BASE, help=f"API の /v1 URL (既定: {DEFAULT_API_BASE})")
253    parser.add_argument("--model", default=DEFAULT_MODEL, help=f"ASR モデル名 (既定: {DEFAULT_MODEL})")
254    parser.add_argument("--output", type=Path, help="出力 TXT。省略時は入力名.qwen.txt")
255    parser.add_argument("--chunk-seconds", type=float, default=240, help="1 チャンクの長さ [s] (既定: 240)")
256    parser.add_argument("--overlap-seconds", type=float, default=10, help="隣接チャンクの重なり [s] (既定: 10)")
257    parser.add_argument("--timeout", type=float, default=1800, help="各 API 要求のタイムアウト [s] (既定: 1800)")
258    parser.add_argument("--resume", action="store_true", help="残っている .partial.json から再開")
259    parser.add_argument("--keep-wav", action="store_true", help="分割した一時 WAV を出力先の .chunks に残す")
260    return parser.parse_args()
261
262
263def main() -> int:
264    """
265    概要:
266        文字起こし処理のメインフローを実行します。
267
268    詳細説明:
269        コマンドライン引数を解析し、入力全体の長さを取得してチャンクごとに分割と文字起こしを実行します。途中で中断された場合でもチェックポイントファイルから再開できます。
270
271    戻り値:
272        :returns: 終了コード
273        :rtype: int
274    """
275    args = parse_args()
276    source = args.input.expanduser().resolve()
277    if not source.is_file():
278        print(f"入力ファイルがありません: {source}", file=sys.stderr)
279        return 2
280    if args.chunk_seconds <= 0 or args.overlap_seconds < 0 or args.overlap_seconds >= args.chunk_seconds:
281        print("--chunk-seconds は正、--overlap-seconds は 0 以上かつ chunk より小さくしてください。", file=sys.stderr)
282        return 2
283
284    output = (args.output or source.with_suffix(".qwen.txt")).expanduser().resolve()
285    output.parent.mkdir(parents=True, exist_ok=True)
286    checkpoint_path = output.with_suffix(output.suffix + ".partial.json")
287    segments_path = output.with_suffix(output.suffix + ".segments.json")
288
289    try:
290        duration = media_duration_seconds(source)
291    except RuntimeError as exc:
292        print(f"エラー: {exc}", file=sys.stderr)
293        return 1
294
295    starts: list[float] = []
296    step = args.chunk_seconds - args.overlap_seconds
297    start = 0.0
298    while start < duration - 0.01:
299        starts.append(start)
300        start += step
301
302    state: dict[str, Any] = {
303        "source": str(source), "duration_seconds": duration, "api_base": args.api_base,
304        "model": args.model, "chunk_seconds": args.chunk_seconds,
305        "overlap_seconds": args.overlap_seconds, "segments": [],
306    }
307    if args.resume and checkpoint_path.exists():
308        try:
309            saved = json.loads(checkpoint_path.read_text(encoding="utf-8"))
310            if Path(saved.get("source", "")).resolve() != source:
311                raise RuntimeError("チェックポイントの入力ファイルが異なります。")
312            state = saved
313            print(f"再開: {len(state['segments'])}/{len(starts)} チャンクは完了済みです。")
314        except (json.JSONDecodeError, RuntimeError) as exc:
315            print(f"エラー: チェックポイントを読めません: {exc}", file=sys.stderr)
316            return 1
317
318    completed_count = len(state["segments"])
319    if completed_count > len(starts):
320        print("エラー: チェックポイントのチャンク数が入力条件と一致しません。", file=sys.stderr)
321        return 1
322
323    chunk_dir: Path | None = output.with_suffix(output.suffix + ".chunks") if args.keep_wav else None
324    if chunk_dir:
325        chunk_dir.mkdir(exist_ok=True)
326
327    temporary_directory: tempfile.TemporaryDirectory[str] | None = None
328    if chunk_dir is None:
329        temporary_directory = tempfile.TemporaryDirectory(prefix="qwen-asr-")
330        work_dir = Path(temporary_directory.name)
331    else:
332        work_dir = chunk_dir
333
334    try:
335        for index, start in enumerate(starts[completed_count:], start=completed_count):
336            length = min(args.chunk_seconds, duration - start)
337            wav_path = work_dir / f"chunk_{index:04d}_{start:010.3f}.wav"
338            print(f"[{index + 1}/{len(starts)}] {start:.1f}–{start + length:.1f} s: WAV 変換中...", flush=True)
339            extract_chunk(source, start, length, wav_path)
340            print(f"[{index + 1}/{len(starts)}] ASR 実行中...", flush=True)
341            began = time.monotonic()
342            text, api_response = transcribe_chunk(args.api_base, args.model, wav_path, args.timeout)
343            elapsed = time.monotonic() - began
344            print(f"[{index + 1}/{len(starts)}] 完了 ({elapsed:.1f} s): {text[:100]}", flush=True)
345            state["segments"].append(
346                {
347                    "index": index, "start_seconds": start, "end_seconds": start + length,
348                    "text": text, "api_response": api_response,
349                }
350            )
351            save_json(checkpoint_path, state)
352            if not args.keep_wav:
353                wav_path.unlink(missing_ok=True)
354    except RuntimeError as exc:
355        print(f"エラー: {exc}\n途中結果は {checkpoint_path} に保存されています。--resume で再開できます。", file=sys.stderr)
356        return 1
357    finally:
358        if temporary_directory is not None:
359            temporary_directory.cleanup()
360
361    merged = ""
362    for segment in state["segments"]:
363        addition = merge_overlap(merged, segment["text"])
364        if addition:
365            merged += ("" if not merged else "\n") + addition
366    output.write_text(merged + ("\n" if merged else ""), encoding="utf-8")
367    save_json(segments_path, state)
368    print(f"\n完了: {output}")
369    print(f"区間・API 応答: {segments_path}")
370    return 0
371
372
373if __name__ == "__main__":
374    raise SystemExit(main())