OpenAI Whisperによる日本語の音声・動画ファイルやマイクの文字起こしツール

# 音声・動画ファイル文字起こしツール
# 特徴技術名: Faster-Whisper
# 出典: SYSTRAN. (2023). faster-whisper: Faster Whisper transcription with CTranslate2. GitHub. https://github.com/SYSTRAN/faster-whisper
# 特徴機能: Faster Whisper batching - 複数の音声ストリームを同時に処理し、スループットを向上させる機能。大規模な音声データセットを必要とするアプリケーションに高いスケーラビリティを提供
# 学習済みモデル: OpenAI Whisper large-v3, medium等のCTranslate2変換モデル（Hugging Face Hub自動ダウンロード対応）。高精度な多言語音声認識を実現し、日本語音声の文字起こしに最適化
# 特徴技術および学習済モデルの利用制限: MIT License - 学術研究・商用利用可能
# 方式設計:
#   関連利用技術:
#     - WebRTC VAD: Googleが開発した音声活動検出技術。リアルタイム音声処理において無音区間を検出し、効率的な音声認識を実現
#     - PyAudio: クロスプラットフォーム音声入出力ライブラリ。リアルタイムマイク入力の取得に使用
#     - Tkinter: Python標準GUIライブラリ。ユーザーインターフェースの構築に使用
#     - PyTorch: 深層学習フレームワーク。GPU/CPU自動選択による最適化された推論処理
#   入力と出力: 入力: 音声・動画ファイル、リアルタイムマイク入力、出力: 文字起こし結果（SRT/TXT/CSV形式）
#   処理手順:
#     1. 音声入力（ファイルまたはマイク）の取得
#     2. VADによる音声活動検出とセグメンテーション
#     3. Faster-Whisperによるバッチ処理での音声認識
#     4. 結果の整形と指定形式での出力
#   前処理、後処理:
#     前処理: 16kHzサンプリングレート正規化、VADによる無音区間除去
#     後処理: タイムスタンプ付与、SRT形式変換、文字列整形
#   追加処理:
#     - 長時間発話時のVAD感度動的調整: 7秒以上の発話時に無音判定フレーム数を450msから210msに短縮し、認識精度を向上
#     - GPU/CPU自動選択: PyTorchによるデバイス自動検出により最適な計算環境を選択
#   調整を必要とする設定値:
#     - RT_VAD_AGGRESSIVENESS(現在値:3): VAD積極性レベル（0-3）。値が大きいほど音声検出が厳格
#     - RT_LONG_SPEECH_THRESHOLD_S(現在値:7.0): 長時間発話判定閾値（秒）。この値を超えるとVAD感度が向上
# 将来方策: VAD積極性の動的調整機能 - 環境ノイズレベルに応じてRT_VAD_AGGRESSIVENESSを自動調整する機能の実装
# その他の重要事項: Windows環境対応、リアルタイム処理とバッチ処理の両方をサポート
# 前準備:
#   pip install -U torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu126
#   pip install faster-whisper tkinter pyaudio webrtcvad numpy

import sys
import os
import threading
import time
from pathlib import Path
import tkinter as tk
from tkinter import filedialog, messagebox, scrolledtext, ttk
import torch
from faster_whisper import WhisperModel
import csv
import queue
import pyaudio
import webrtcvad
import numpy as np

# 対応ファイル形式
VIDEO_EXTENSIONS = {'.mp4', '.avi', '.mov', '.mkv', '.flv', '.m4v'}
AUDIO_EXTENSIONS = {'.wav', '.mp3', '.aac', '.ogg', '.wma', '.flac', '.webm', '.3gp'}
SUPPORTED_EXTENSIONS = VIDEO_EXTENSIONS | AUDIO_EXTENSIONS

# リアルタイム処理用の定数
RT_CHUNK_SAMPLES = 480
RT_VAD_AGGRESSIVENESS = 3
RT_SAMPLERATE = 16000
# VAD感度を動的に変更するための定数
RT_LONG_SPEECH_THRESHOLD_S = 7.0  # 長時間発話とみなす閾値（秒）
RT_PADDING_FRAMES_NORMAL = 15     # 通常時の無音判定フレーム数 (15 * 30ms = 450ms)
RT_PADDING_FRAMES_SENSITIVE = 7   # 長時間発話時の無音判定フレーム数 (7 * 30ms = 210ms)


def detect_best_device():
    """最適なデバイスを自動検出"""
    if torch.cuda.is_available():
        return "cuda", "float16"
    else:
        return "cpu", "int8"


def format_srt_timestamp(seconds: float) -> str:
    """SRT形式のタイムスタンプに変換（丸め誤差に強い）"""
    total_ms = int(round(seconds * 1000))
    hours, rem = divmod(total_ms, 3600 * 1000)
    minutes, rem = divmod(rem, 60 * 1000)
    secs, millis = divmod(rem, 1000)
    return f"{hours:02d}:{minutes:02d}:{secs:02d},{millis:03d}"


class TranscriptionGUI:
    def __init__(self, root):
        self.root = root
        self.root.title("音声・動画ファイル文字起こしツール")
        self.root.geometry("800x600")

        self.auto_device, self.auto_compute_type = detect_best_device()
        self.model = None
        self.files_to_process = []
        self.worker_thread = None
        self.cancel_requested = threading.Event()

        self.audio_queue = queue.Queue()
        self.is_recording = threading.Event()
        self.audio_thread = None

        self.setup_gui()
        self.root.protocol("WM_DELETE_WINDOW", self.on_closing)

    def setup_gui(self):
        main_frame = ttk.Frame(self.root, padding="10")
        main_frame.pack(fill=tk.BOTH, expand=True)

        model_frame = ttk.LabelFrame(main_frame, text="設定", padding="5")
        model_frame.pack(fill=tk.X, pady=(0, 10))

        ttk.Label(model_frame, text="モデル:").pack(side=tk.LEFT)
        self.model_var = tk.StringVar(value="medium")
        self.model_combobox = ttk.Combobox(
            model_frame, textvariable=self.model_var,
            values=["tiny", "base", "small", "medium", "large-v3", "large-v3-turbo"],
            state="readonly", width=15
        )
        self.model_combobox.pack(side=tk.LEFT, padx=(5, 20))

        ttk.Label(model_frame, text="デバイス:").pack(side=tk.LEFT)
        device_text = "GPU (CUDA)" if self.auto_device == "cuda" else "CPU"
        ttk.Label(model_frame, text=device_text, foreground="blue").pack(side=tk.LEFT, padx=(5, 20))

        file_frame = ttk.LabelFrame(main_frame, text="ファイルまたはフォルダ選択", padding="5")
        file_frame.pack(fill=tk.X, pady=(0, 10))
        self.file_path_var = tk.StringVar()
        ttk.Entry(file_frame, textvariable=self.file_path_var, width=60).pack(side=tk.LEFT, padx=(0, 5), fill=tk.X, expand=True)
        ttk.Button(file_frame, text="ファイル選択", command=self.select_files).pack(side=tk.LEFT, padx=(0, 5))
        ttk.Button(file_frame, text="フォルダ選択", command=self.select_folder).pack(side=tk.LEFT)

        action_frame = ttk.Frame(main_frame)
        action_frame.pack(fill=tk.X, pady=(5, 10))

        file_action_frame = ttk.Frame(action_frame)
        file_action_frame.pack(side=tk.LEFT)
        self.transcribe_button = ttk.Button(file_action_frame, text="処理を開始して保存...", command=self.start_file_transcription)
        self.transcribe_button.pack(side=tk.LEFT)

        self.save_format_var = tk.StringVar(value="SRT")
        self.save_format_combobox = ttk.Combobox(
            file_action_frame, textvariable=self.save_format_var,
            values=["SRT", "TXT", "CSV"],
            state="readonly", width=10
        )
        self.save_format_combobox.pack(side=tk.LEFT, padx=(5, 0))

        self.stop_file_button = ttk.Button(file_action_frame, text="中断", command=self.stop_transcription, state=tk.DISABLED)
        self.stop_file_button.pack(side=tk.LEFT, padx=(5, 0))

        mic_action_frame = ttk.Frame(action_frame)
        mic_action_frame.pack(side=tk.RIGHT)
        self.mic_start_button = ttk.Button(mic_action_frame, text="マイク入力開始", command=self.start_mic_transcription)
        self.mic_start_button.pack(side=tk.LEFT)
        self.mic_stop_button = ttk.Button(mic_action_frame, text="停止", command=self.stop_mic_transcription, state=tk.DISABLED)
        self.mic_stop_button.pack(side=tk.LEFT, padx=(5, 0))

        self.progress_var = tk.DoubleVar()
        self.progressbar = ttk.Progressbar(main_frame, variable=self.progress_var)
        self.progressbar.pack(fill=tk.X, pady=(0, 10))

        self.result_text = scrolledtext.ScrolledText(main_frame, height=20, wrap=tk.WORD, font=("Consolas", 10))
        self.result_text.pack(fill=tk.BOTH, expand=True, pady=(0, 5))

        self.context_menu = tk.Menu(self.result_text, tearoff=0)
        self.context_menu.add_command(label="コピー", command=lambda: self.result_text.event_generate("<<Copy>>"))
        self.context_menu.add_separator()
        self.context_menu.add_command(label="すべて選択", command=lambda: self.result_text.tag_add("sel", "1.0", "end"))
        self.result_text.bind("<Button-3>", self.show_context_menu)

        status_frame = ttk.Frame(main_frame)
        status_frame.pack(fill=tk.X)
        self.status_var = tk.StringVar()
        self.status_label = ttk.Label(status_frame, textvariable=self.status_var, anchor=tk.W)
        self.status_label.pack(fill=tk.X)

    def set_file_controls_state(self, state):
        """ファイル処理関連のコントロールの状態を一括で変更する"""
        self.transcribe_button.config(state=state)
        self.save_format_combobox.config(state="readonly" if state == tk.NORMAL else tk.DISABLED)
        self.stop_file_button.config(state=tk.DISABLED if state == tk.NORMAL else tk.NORMAL)
        self.mic_start_button.config(state=state)

    def set_mic_controls_state(self, state):
        """マイク処理関連のコントロールの状態を一括で変更する"""
        self.mic_start_button.config(state=state)
        self.mic_stop_button.config(state=tk.DISABLED if state == tk.NORMAL else tk.NORMAL)
        self.transcribe_button.config(state=state)
        self.model_combobox.config(state="readonly" if state == tk.NORMAL else tk.DISABLED)
        self.save_format_combobox.config(state=tk.DISABLED)

    def show_context_menu(self, event):
        """右クリックメニューを表示し、選択状態に応じて「コピー」を有効/無効化"""
        try:
            self.result_text.get("sel.first", "sel.last")
            self.context_menu.entryconfig(0, state=tk.NORMAL)
        except tk.TclError:
            self.context_menu.entryconfig(0, state=tk.DISABLED)
        self.context_menu.tk_popup(event.x_root, event.y_root)

    def select_files(self):
        filenames = filedialog.askopenfilenames(
            title="音声・動画ファイルを選択（複数可）",
            filetypes=[("対応ファイル", " ".join(f"*{ext}" for ext in SUPPORTED_EXTENSIONS)), ("すべて", "*.*")]
        )
        if filenames:
            self.files_to_process = list(filenames)
            self.file_path_var.set(f"{len(self.files_to_process)}個のファイルを選択しました")
            self.update_status(f"{len(self.files_to_process)}個のファイルを選択しました")

    def select_folder(self):
        folder = filedialog.askdirectory(title="処理対象のフォルダを選択")
        if folder:
            self.files_to_process = [
                os.path.join(folder, f) for f in os.listdir(folder)
                if Path(f).suffix.lower() in SUPPORTED_EXTENSIONS
            ]
            if self.files_to_process:
                self.file_path_var.set(f"{len(self.files_to_process)}個のファイルをフォルダから選択しました")
                self.update_status(f"{len(self.files_to_process)}個のファイルをフォルダから選択しました")
            else:
                self.file_path_var.set("フォルダ内に対応ファイルがありません")
                self.update_status("フォルダ内に対応ファイルがありません")

    def update_status(self, text: str):
        self.root.after(0, lambda: self.status_var.set(text))

    def update_result(self, text: str):
        def _update():
            self.result_text.insert(tk.END, text)
            self.result_text.see(tk.END)
        self.root.after(0, _update)

    def start_file_transcription(self):
        if not self.files_to_process:
            messagebox.showerror("エラー", "ファイルまたはフォルダを選択してください")
            return

        output_dir = filedialog.askdirectory(title="保存先のフォルダを選択してください")
        if not output_dir:
            messagebox.showwarning("キャンセル", "保存先が選択されなかったため、処理を中止しました。")
            return

        self.set_file_controls_state(tk.DISABLED)
        self.cancel_requested.clear()

        self.progressbar.config(maximum=len(self.files_to_process), value=0)

        save_format = self.save_format_var.get()

        self.worker_thread = threading.Thread(
            target=self.batch_transcribe_worker,
            args=(output_dir, save_format),
            daemon=True
        )
        self.worker_thread.start()

    def stop_transcription(self):
        self.update_status("処理の中断を要求しました...")
        self.cancel_requested.set()
        self.stop_file_button.config(state=tk.DISABLED)

    def batch_transcribe_worker(self, output_dir, save_format):
        self.update_status(f"一括処理を開始... 対象: {len(self.files_to_process)}ファイル")

        processed_count = 0
        for i, file_path in enumerate(self.files_to_process):
            if self.cancel_requested.is_set():
                self.update_status("処理がユーザーによって中断されました。")
                break

            self.root.after(0, lambda: self.result_text.delete(1.0, tk.END))
            self.update_status(f"処理中 ({i+1}/{len(self.files_to_process)}): {Path(file_path).name}")

            result_segments = self.transcribe_single_file(file_path)

            if result_segments:
                self.save_single_result(result_segments, file_path, output_dir, save_format)
                processed_count += 1

            self.root.after(0, lambda i=i: self.progressbar.config(value=i + 1))

        if not self.cancel_requested.is_set():
            self.update_status(f"すべての処理が完了しました。 {processed_count}個のファイルを保存しました。")

        self.root.after(0, self.file_transcription_finished)

    def transcribe_single_file(self, file_path):
        temp_wav_file = None
        try:
            file_ext = Path(file_path).suffix.lower()
            if file_ext in VIDEO_EXTENSIONS:
                with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmpfile:
                    temp_wav_file = tmpfile.name

                command = ["ffmpeg", "-y", "-i", file_path, "-vn", "-ar", "16000", "-ac", "1", "-f", "wav", temp_wav_file]
                proc = subprocess.run(command, capture_output=True, text=True, encoding='utf-8', errors='ignore')
                if proc.returncode != 0:
                    self.update_status(f"音声抽出に失敗: {proc.stderr or 'ffmpeg error'}")
                    return None
                audio_file_to_process = temp_wav_file
            else:
                audio_file_to_process = file_path

            if not self.model:
                model_size = self.model_var.get()
                self.model = WhisperModel(model_size, device=self.auto_device, compute_type=self.auto_compute_type)

            segments_gen, _ = self.model.transcribe(
                audio_file_to_process, language="ja", beam_size=5, temperature=0.0,
                condition_on_previous_text=True,
                initial_prompt="以下は日本語の音声である。文は意味のまとまりで区切り、適切に句読点「、」「。」を付ける。"
            )

            result_segments = []
            srt_index = 0
            for segment in segments_gen:
                if self.cancel_requested.is_set():
                    break
                srt_index += 1
                srt_entry = f"{srt_index}\n{format_srt_timestamp(segment.start)} --> {format_srt_timestamp(segment.end)}\n{segment.text.strip()}\n\n"
                self.update_result(srt_entry)
                result_segments.append({'start': segment.start, 'end': segment.end, 'text': segment.text.strip()})

            return result_segments
        except Exception as e:
            self.update_status(f"エラーが発生: {e}")
            return None
        finally:
            if temp_wav_file and os.path.exists(temp_wav_file):
                try:
                    os.remove(temp_wav_file)
                except OSError as e:
                    self.update_status(f"一時ファイル削除エラー: {e}")

    def save_single_result(self, segments, original_path, output_dir, save_format):
        base_name = Path(original_path).stem
        try:
            if save_format == "SRT":
                output_path = os.path.join(output_dir, f"{base_name}.srt")
                with open(output_path, 'w', encoding='utf-8') as f:
                    for i, seg in enumerate(segments):
                        f.write(f"{i+1}\n{format_srt_timestamp(seg['start'])} --> {format_srt_timestamp(seg['end'])}\n{seg['text']}\n\n")
            elif save_format == "TXT":
                output_path = os.path.join(output_dir, f"{base_name}.txt")
                with open(output_path, 'w', encoding='utf-8') as f:
                    for seg in segments:
                        f.write(f"{seg['text']}\n")
            elif save_format == "CSV":
                output_path = os.path.join(output_dir, f"{base_name}.csv")
                with open(output_path, 'w', encoding='utf-8', newline='') as f:
                    writer = csv.writer(f)
                    writer.writerow(["start_time", "end_time", "text"])
                    for seg in segments:
                        writer.writerow([f"{seg['start']:.3f}", f"{seg['end']:.3f}", seg['text']])
        except Exception as e:
            self.update_status(f"保存エラー: {e}")

    def file_transcription_finished(self):
        self.set_file_controls_state(tk.NORMAL)
        self.progressbar.config(value=0)

    def start_mic_transcription(self):
        try:
            # PyAudio初期化テスト
            p = pyaudio.PyAudio()
            p.terminate()
        except Exception as e:
            messagebox.showerror("マイクエラー", f"マイクの初期化に失敗しました: {e}")
            return

        self.set_mic_controls_state(tk.DISABLED)
        self.result_text.delete(1.0, tk.END)
        self.update_status("マイク入力を開始します...")

        self.is_recording.set()

        self.audio_thread = threading.Thread(target=self.record_audio, daemon=True)
        self.audio_thread.start()

        self.worker_thread = threading.Thread(target=self.mic_transcribe_worker, daemon=True)
        self.worker_thread.start()

    def stop_mic_transcription(self):
        self.update_status("マイク入力を停止しています...")
        self.is_recording.clear()
        self.mic_stop_button.config(state=tk.DISABLED)

    def record_audio(self):
        p = pyaudio.PyAudio()
        try:
            stream = p.open(format=pyaudio.paInt16, channels=1, rate=RT_SAMPLERATE,
                            input=True, frames_per_buffer=RT_CHUNK_SAMPLES)
            while self.is_recording.is_set():
                try:
                    data = stream.read(RT_CHUNK_SAMPLES, exception_on_overflow=False)
                    self.audio_queue.put(data)
                except Exception as e:
                    self.update_status(f"音声読み取りエラー: {e}")
                    break
            stream.stop_stream()
            stream.close()
        except Exception as e:
            self.update_status(f"マイクエラー: {e}")
        finally:
            p.terminate()

    def mic_transcribe_worker(self):
        if not self.model:
            model_size = self.model_var.get()
            self.model = WhisperModel(model_size, device=self.auto_device, compute_type=self.auto_compute_type)

        vad = webrtcvad.Vad(RT_VAD_AGGRESSIVENESS)

        frames = []
        ring_buffer_size = RT_PADDING_FRAMES_NORMAL
        ring_buffer = []
        triggered = False
        speech_start_time = None

        while self.is_recording.is_set() or not self.audio_queue.empty():
            try:
                chunk = self.audio_queue.get(timeout=0.1)

                # 空データをスキップ
                if len(chunk) != RT_CHUNK_SAMPLES * 2:  # 16bit = 2bytes
                    continue

                try:
                    is_speech = vad.is_speech(chunk, RT_SAMPLERATE)
                except Exception:
                    # VADエラー時は無音として扱う
                    is_speech = False

                if not triggered:
                    if len(ring_buffer) >= ring_buffer_size:
                        ring_buffer.pop(0)
                    ring_buffer.append(chunk)
                    if is_speech:
                        triggered = True
                        speech_start_time = time.time()
                        frames.extend(ring_buffer)
                        self.update_status("音声検出...")
                else:
                    frames.append(chunk)

                    elapsed_time = time.time() - speech_start_time if speech_start_time else 0
                    if elapsed_time > RT_LONG_SPEECH_THRESHOLD_S:
                        current_padding_frames = RT_PADDING_FRAMES_SENSITIVE
                    else:
                        current_padding_frames = RT_PADDING_FRAMES_NORMAL

                    if len(ring_buffer) >= current_padding_frames:
                        ring_buffer.pop(0)
                    ring_buffer.append(chunk)

                    if not is_speech and len(ring_buffer) >= current_padding_frames:
                        recent_silence = all(
                            not vad.is_speech(f, RT_SAMPLERATE)
                            if len(f) == RT_CHUNK_SAMPLES * 2 else True
                            for f in ring_buffer[-current_padding_frames:]
                        )

                        if recent_silence:
                            self.update_status("認識中...")
                            audio_data = b''.join(frames)
                            frames.clear()
                            triggered = False
                            speech_start_time = None
                            ring_buffer.clear()

                            if len(audio_data) > 0:
                                audio_np = np.frombuffer(audio_data, dtype=np.int16).astype(np.float32) / 32768.0
                                try:
                                    segments, _ = self.model.transcribe(audio_np, language="ja", beam_size=5)
                                    transcribed_text = "".join(seg.text for seg in segments).strip()
                                    if transcribed_text:
                                        self.result_text.insert(tk.END, transcribed_text + " ")
                                        self.result_text.see(tk.END)
                                        self.update_status("待機中...")
                                except Exception as e:
                                    self.update_status(f"認識エラー: {e}")

            except queue.Empty:
                continue
            except Exception as e:
                self.update_status(f"処理エラー: {e}")

        # ループを抜けた後にフレームが残っている場合、最後の発話を処理する
        if frames:
            self.update_status("最後の音声を認識中...")
            audio_data = b''.join(frames)

            if len(audio_data) > 0:
                audio_np = np.frombuffer(audio_data, dtype=np.int16).astype(np.float32) / 32768.0
                try:
                    segments, _ = self.model.transcribe(audio_np, language="ja", beam_size=5)
                    transcribed_text = "".join(seg.text for seg in segments).strip()
                    if transcribed_text:
                        self.update_result(transcribed_text + " ")
                except Exception as e:
                    self.update_status(f"最終認識エラー: {e}")

        self.root.after(0, self.mic_transcription_finished)

    def mic_transcription_finished(self):
        self.set_mic_controls_state(tk.NORMAL)
        self.update_status("マイク入力を停止しました。")

    def on_closing(self):
        if self.worker_thread and self.worker_thread.is_alive():
            if self.is_recording.is_set():
                self.stop_mic_transcription()
                self.wait_for_thread_and_close()
            else:
                if messagebox.askyesno("確認", "処理が実行中です。中断して終了しますか？"):
                    self.cancel_requested.set()
                    self.wait_for_thread_and_close()
        else:
            self.root.destroy()

    def wait_for_thread_and_close(self):
        max_wait_time = 5.0  # 最大5秒待機
        start_time = time.time()

        def check_and_close():
            elapsed_time = time.time() - start_time
            if (not self.worker_thread or not self.worker_thread.is_alive()) and \
               (not self.audio_thread or not self.audio_thread.is_alive()):
                self.root.destroy()
            elif elapsed_time >= max_wait_time:
                # 強制終了
                self.root.destroy()
            else:
                self.root.after(100, check_and_close)

        check_and_close()

def main():
    root = tk.Tk()
    app = TranscriptionGUI(root)
    root.mainloop()

if __name__ == "__main__":
    main()