kanarisusunda

app_wsl copy.py

@@ -0,0 +1,669 @@
+from nemo.collections.asr.models import ASRModel
+import torch
+import gradio as gr
+import spaces
+import gc
+import shutil
+from pathlib import Path
+from pydub import AudioSegment
+import numpy as np
+import os
+import gradio.themes as gr_themes
+import csv
+import json
+from typing import List, Tuple
+
+device = "cuda" if torch.cuda.is_available() else "cpu"
+MODEL_NAME="nvidia/parakeet-tdt-0.6b-v2"
+
+model = ASRModel.from_pretrained(model_name=MODEL_NAME)
+model.eval()
+
+def start_session(request: gr.Request):
+    session_hash = request.session_hash
+    # プロジェクトディレクトリ内のoutputsフォルダを使用
+    base_dir = Path(__file__).parent
+    session_dir = base_dir / "outputs" / session_hash
+    session_dir.mkdir(parents=True, exist_ok=True)
+    print(f"Session with hash {session_hash} started in {session_dir}")
+    return session_dir.as_posix()
+
+def end_session(request: gr.Request):
+    session_hash = request.session_hash
+    base_dir = Path(__file__).parent
+    session_dir = base_dir / "outputs" / session_hash
+    if session_dir.exists():
+        print(f"Session directory {session_dir} will be preserved.")
+        # 削除しないように変更
+        # shutil.rmtree(session_dir)
+    print(f"Session with hash {session_hash} ended.")
+
+def get_audio_segment(audio_path, start_second, end_second):
+    if not audio_path or not Path(audio_path).exists():
+        print(f"Warning: Audio path '{audio_path}' not found or invalid for clipping.")
+        return None
+    try:
+        start_ms = int(start_second * 1000)
+        end_ms = int(end_second * 1000)
+
+        start_ms = max(0, start_ms)
+        if end_ms <= start_ms:
+            print(f"Warning: End time ({end_second}s) is not after start time ({start_second}s). Adjusting end time.")
+            end_ms = start_ms + 100
+
+        audio = AudioSegment.from_file(audio_path)
+        clipped_audio = audio[start_ms:end_ms]
+
+        samples = np.array(clipped_audio.get_array_of_samples())
+        if clipped_audio.channels == 2:
+            samples = samples.reshape((-1, 2)).mean(axis=1).astype(samples.dtype)
+
+        frame_rate = clipped_audio.frame_rate
+        if frame_rate <= 0:
+            print(f"Warning: Invalid frame rate ({frame_rate}) detected for clipped audio.")
+            frame_rate = audio.frame_rate
+
+        if samples.size == 0:
+            print(f"Warning: Clipped audio resulted in empty samples array ({start_second}s to {end_second}s).")
+            return None
+
+        return (frame_rate, samples)
+    except FileNotFoundError:
+        print(f"Error: Audio file not found at path: {audio_path}")
+        return None
+    except Exception as e:
+        print(f"Error clipping audio {audio_path} from {start_second}s to {end_second}s: {e}")
+        return None
+
+def preprocess_audio(audio_path, session_dir):
+    """
+    オーディオファイルの前処理（リサンプリング、モノラル変換）を行う。
+
+    Args:
+        audio_path (str): 入力オーディオファイルのパス。
+        session_dir (str): セッションディレクトリのパス。
+
+    Returns:
+        tuple: (processed_path, info_path_name, duration_sec) のタプル、または None（処理に失敗した場合）。
+    """
+    try:
+        original_path_name = Path(audio_path).name
+        audio_name = Path(audio_path).stem
+
+        try:
+            gr.Info(f"Loading audio: {original_path_name}", duration=2)
+            audio = AudioSegment.from_file(audio_path)
+            duration_sec = audio.duration_seconds
+        except Exception as load_e:
+            gr.Error(f"Failed to load audio file {original_path_name}: {load_e}", duration=None)
+            return None, None, None
+
+        resampled = False
+        mono = False
+        target_sr = 16000
+
+        if audio.frame_rate != target_sr:
+            try:
+                audio = audio.set_frame_rate(target_sr)
+                resampled = True
+            except Exception as resample_e:
+                gr.Error(f"Failed to resample audio: {resample_e}", duration=None)
+                return None, None, None
+
+        if audio.channels == 2:
+            try:
+                audio = audio.set_channels(1)
+                mono = True
+            except Exception as mono_e:
+                gr.Error(f"Failed to convert audio to mono: {mono_e}", duration=None)
+                return None, None, None
+        elif audio.channels > 2:
+            gr.Error(f"Audio has {audio.channels} channels. Only mono (1) or stereo (2) supported.", duration=None)
+            return None, None, None
+
+        processed_audio_path = None
+        if resampled or mono:
+            try:
+                processed_audio_path = Path(session_dir, f"{audio_name}_resampled.wav")
+                audio.export(processed_audio_path, format="wav")
+                transcribe_path = processed_audio_path.as_posix()
+                info_path_name = f"{original_path_name} (processed)"
+            except Exception as export_e:
+                gr.Error(f"Failed to export processed audio: {export_e}", duration=None)
+                if processed_audio_path and os.path.exists(processed_audio_path):
+                    os.remove(processed_audio_path)
+                return None, None, None
+        else:
+            transcribe_path = audio_path
+            info_path_name = original_path_name
+
+        return transcribe_path, info_path_name, duration_sec
+    except Exception as e:
+        gr.Error(f"Audio preprocessing failed: {e}", duration=None)
+        return None, None, None
+
+def transcribe_audio(transcribe_path, model, duration_sec, device):
+    """
+    オーディオファイルを文字起こしし、タイムスタンプを取得する。
+
+    Args:
+        transcribe_path (str): 入力オーディオファイルのパス。
+        model (ASRModel): 使用するASRモデル。
+        duration_sec (float): オーディオファイルの長さ（秒）。
+        device (str): 使用するデバイス（'cuda' or 'cpu'）。
+
+    Returns:
+        tuple: (vis_data, raw_times_data, word_vis_data) のタプル、または None（処理に失敗した場合）。
+    """
+    long_audio_settings_applied = False
+    try:
+        # CUDA使用前にメモリをクリア
+        if device == 'cuda':
+            torch.cuda.empty_cache()
+            gc.collect()
+
+        model.to(device)
+        model.to(torch.float32)
+        gr.Info(f"Transcribing on {device}...", duration=2)
+
+        if duration_sec > 480:
+            try:
+                gr.Info("Audio longer than 8 minutes. Applying optimized settings for long transcription.", duration=3)
+                print("Applying long audio settings: Local Attention and Chunking.")
+                model.change_attention_model("rel_pos_local_attn", [256,256])
+                model.change_subsampling_conv_chunking_factor(1)
+                # メモリ効率を改善するための設定
+                torch.cuda.empty_cache()
+                gc.collect()
+                long_audio_settings_applied = True
+            except Exception as setting_e:
+                gr.Warning(f"Could not apply long audio settings: {setting_e}", duration=5)
+                print(f"Warning: Failed to apply long audio settings: {setting_e}")
+        
+        # より効率的なメモリ使用のためにbfloat16を使用
+        model.to(torch.bfloat16)
+        
+        # メモリ使用状況をログに出力
+        if device == 'cuda':
+            print(f"CUDA Memory before transcription: {torch.cuda.memory_allocated() / 1024**2:.2f} MB")
+        
+        output = model.transcribe([transcribe_path], timestamps=True)
+
+        if not output or not isinstance(output, list) or not output[0] or not hasattr(output[0], 'timestamp') or not output[0].timestamp or 'segment' not in output[0].timestamp:
+            gr.Error("Transcription failed or produced unexpected output format.", duration=None)
+            return None, None, None
+
+        # 結果を処理する前にメモリを解放
+        if device == 'cuda':
+            model.cpu()
+            torch.cuda.empty_cache()
+            gc.collect()
+
+        segment_timestamps = output[0].timestamp['segment']
+        vis_data = [[f"{ts['start']:.2f}", f"{ts['end']:.2f}", ts['segment']] for ts in segment_timestamps]
+        raw_times_data = [[ts['start'], ts['end']] for ts in segment_timestamps]
+
+        word_timestamps_raw = output[0].timestamp.get("word", [])
+        word_vis_data = [
+            [f"{w['start']:.2f}", f"{w['end']:.2f}", w["word"]]
+            for w in word_timestamps_raw if isinstance(w, dict) and 'start' in w and 'end' in w and 'word' in w
+        ]
+
+        gr.Info("Transcription complete.", duration=2)
+        return vis_data, raw_times_data, word_vis_data
+
+    except torch.cuda.OutOfMemoryError as e:
+        error_msg = 'CUDA out of memory. Please try a shorter audio or reduce GPU load.'
+        print(f"CUDA OutOfMemoryError: {e}")
+        gr.Error(error_msg, duration=None)
+        # メモリエラー時に強制的にクリーンアップ
+        if device == 'cuda':
+            torch.cuda.empty_cache()
+            gc.collect()
+        return None, None, None
+
+    except Exception as e:
+        error_msg = f"Transcription failed: {e}"
+        print(f"Error during transcription processing: {e}")
+        gr.Error(error_msg, duration=None)
+        return None, None, None
+
+    finally:
+        try:
+            if long_audio_settings_applied:
+                try:
+                    print("Reverting long audio settings.")
+                    model.change_attention_model("rel_pos")
+                    model.change_subsampling_conv_chunking_factor(-1)
+                except Exception as revert_e:
+                    print(f"Warning: Failed to revert long audio settings: {revert_e}")
+                    gr.Warning(f"Issue reverting model settings after long transcription: {revert_e}", duration=5)
+
+            if device == 'cuda':
+                model.cpu()
+                torch.cuda.empty_cache()
+            gc.collect()
+        except Exception as cleanup_e:
+            print(f"Error during model cleanup: {cleanup_e}")
+            gr.Warning(f"Issue during model cleanup: {cleanup_e}", duration=5)
+
+def save_transcripts(session_dir, audio_name, vis_data, word_vis_data):
+    """
+    文字起こし結果を各種ファイル形式（CSV、SRT、VTT、JSON、LRC）で保存する。
+
+    Args:
+        session_dir (str): セッションディレクトリのパス。
+        audio_name (str): オーディオファイルの名前。
+        vis_data (list): 表示用の文字起こし結果のリスト。
+        word_vis_data (list): 単語レベルのタイムスタンプのリスト。
+
+    Returns:
+        tuple: 各ファイルのダウンロードボタンの更新情報を含むタプル。
+    """
+    try:
+        csv_headers = ["Start (s)", "End (s)", "Segment"]
+        csv_file_path = Path(session_dir, f"transcription_{audio_name}.csv")
+        with open(csv_file_path, 'w', newline='', encoding='utf-8') as f:
+            writer = csv.writer(f)
+            writer.writerow(csv_headers)
+            writer.writerows(vis_data)
+        print(f"CSV transcript saved to temporary file: {csv_file_path}")
+
+        srt_file_path = Path(session_dir, f"transcription_{audio_name}.srt")
+        vtt_file_path = Path(session_dir, f"transcription_{audio_name}.vtt")
+        json_file_path = Path(session_dir, f"transcription_{audio_name}.json")
+        write_srt(vis_data, srt_file_path)
+        write_vtt(vis_data, word_vis_data, vtt_file_path)
+        write_json(vis_data, word_vis_data, json_file_path)
+        print(f"SRT, VTT, JSON transcript saved to temporary files: {srt_file_path}, {vtt_file_path}, {json_file_path}")
+
+        lrc_file_path = Path(session_dir, f"transcription_{audio_name}.lrc")
+        write_lrc(vis_data, lrc_file_path)
+        print(f"LRC transcript saved to temporary file: {lrc_file_path}")
+
+        return (
+            gr.DownloadButton(value=csv_file_path.as_posix(), visible=True),
+            gr.DownloadButton(value=srt_file_path.as_posix(), visible=True),
+            gr.DownloadButton(value=vtt_file_path.as_posix(), visible=True),
+            gr.DownloadButton(value=json_file_path.as_posix(), visible=True),
+            gr.DownloadButton(value=lrc_file_path.as_posix(), visible=True)
+        )
+    except Exception as e:
+        gr.Error(f"Failed to create transcript files: {e}", duration=None)
+        print(f"Error writing transcript files: {e}")
+        return tuple([gr.DownloadButton(visible=False)] * 5)
+
+def split_audio_with_overlap(audio_path: str, session_dir: str, chunk_length_sec: int = 3600, overlap_sec: int = 30) -> List[str]:
+    """
+    音声ファイルをchunk_length_secごとにoverlap_secのオーバーラップ付きで分割し、
+    分割ファイルのパスリストを返す。
+    """
+    audio = AudioSegment.from_file(audio_path)
+    duration = audio.duration_seconds
+    chunk_paths = []
+    start = 0
+    chunk_idx = 0
+    while start < duration:
+        end = min(start + chunk_length_sec, duration)
+        # オーバーラップを考慮
+        chunk_start = max(0, start - (overlap_sec if start > 0 else 0))
+        chunk_end = min(end + (overlap_sec if end < duration else 0), duration)
+        chunk = audio[chunk_start * 1000:chunk_end * 1000]
+        chunk_path = Path(session_dir, f"chunk_{chunk_idx:03d}.wav").as_posix()
+        chunk.export(chunk_path, format="wav")
+        chunk_paths.append(chunk_path)
+        start += chunk_length_sec
+        chunk_idx += 1
+    return chunk_paths
+
+@spaces.GPU
+def get_transcripts_and_raw_times(audio_path, session_dir, progress=gr.Progress(track_tqdm=True)):
+    """
+    オーディオファイルを処理し、文字起こし結果を生成する。
+    3時間を超える場合は60分ごとに分割し、オーバーラップ付きでASRを実行してマージする。
+    """
+    if not audio_path:
+        gr.Error("No audio file path provided for transcription.", duration=None)
+        return [], [], [], None, gr.DownloadButton(visible=False), gr.DownloadButton(visible=False), gr.DownloadButton(visible=False), gr.DownloadButton(visible=False), gr.DownloadButton(visible=False)
+
+    audio_name = Path(audio_path).stem
+    processed_audio_path = None
+    temp_chunk_paths = []
+
+    try:
+        # オーディオの前処理
+        transcribe_path, info_path_name, duration_sec = preprocess_audio(audio_path, session_dir)
+        if not transcribe_path or not duration_sec:
+            return [], [], [], audio_path, gr.DownloadButton(visible=False), gr.DownloadButton(visible=False), gr.DownloadButton(visible=False), gr.DownloadButton(visible=False), gr.DownloadButton(visible=False)
+
+        processed_audio_path = transcribe_path if transcribe_path != audio_path else None        # 3時間超の場合は分割して逐次ASR
+        if duration_sec > 10800:
+            gr.Info("Audio is longer than 3 hours. Splitting into 1-hour chunks with overlap for transcription.", duration=5)
+            chunk_paths = split_audio_with_overlap(transcribe_path, session_dir, chunk_length_sec=3600, overlap_sec=30)
+            temp_chunk_paths = chunk_paths.copy()
+            all_vis_data = []
+            all_raw_times_data = []
+            all_word_vis_data = []
+            offset = 0.0
+            prev_end = 0.0
+            for i, chunk_path in enumerate(progress.tqdm(chunk_paths, desc="Processing audio chunks")):
+                chunk_audio = AudioSegment.from_file(chunk_path)
+                chunk_duration = chunk_audio.duration_seconds
+                # ASR実行
+                result = transcribe_audio(chunk_path, model, chunk_duration, device)
+                if not result:
+                    continue
+                vis_data, raw_times_data, word_vis_data = result
+                # タイムスタンプを全体のオフセットに合わせて補正
+                vis_data_offset = []
+                raw_times_data_offset = []
+                word_vis_data_offset = []
+                for row in vis_data:
+                    s, e, seg = float(row[0]), float(row[1]), row[2]
+                    vis_data_offset.append([f"{s+offset:.2f}", f"{e+offset:.2f}", seg])
+                for row in raw_times_data:
+                    s, e = float(row[0]), float(row[1])
+                    raw_times_data_offset.append([s+offset, e+offset])
+                for row in word_vis_data:
+                    s, e, w = float(row[0]), float(row[1]), row[2]
+                    word_vis_data_offset.append([f"{s+offset:.2f}", f"{e+offset:.2f}", w])
+                # オーバーラップ部分の重複除去（単純に前回のend以降のみ追加）
+                vis_data_offset = [row for row in vis_data_offset if float(row[0]) >= prev_end]
+                raw_times_data_offset = [row for row in raw_times_data_offset if row[0] >= prev_end]
+                word_vis_data_offset = [row for row in word_vis_data_offset if float(row[0]) >= prev_end]
+                if vis_data_offset:
+                    prev_end = float(vis_data_offset[-1][1])
+                all_vis_data.extend(vis_data_offset)
+                all_raw_times_data.extend(raw_times_data_offset)
+                all_word_vis_data.extend(word_vis_data_offset)
+                offset += chunk_duration - (30 if i < len(chunk_paths)-1 else 0)
+            # ファイルの保存
+            button_updates = save_transcripts(session_dir, audio_name, all_vis_data, all_word_vis_data)
+            # 一時分割ファイル削除
+            for p in temp_chunk_paths:
+                try:
+                    os.remove(p)
+                except Exception:
+                    pass
+            return (
+                all_vis_data,
+                all_raw_times_data,
+                all_word_vis_data,
+                audio_path,
+                *button_updates
+            )
+        else:
+            # 3時間以内は従来通り
+            result = transcribe_audio(transcribe_path, model, duration_sec, device)
+            if not result:
+                return [], [], [], audio_path, gr.DownloadButton(visible=False), gr.DownloadButton(visible=False), gr.DownloadButton(visible=False), gr.DownloadButton(visible=False), gr.DownloadButton(visible=False)
+            vis_data, raw_times_data, word_vis_data = result
+            button_updates = save_transcripts(session_dir, audio_name, vis_data, word_vis_data)
+            return (
+                vis_data,
+                raw_times_data,
+                word_vis_data,
+                audio_path,
+                *button_updates
+            )
+    finally:
+        if processed_audio_path and os.path.exists(processed_audio_path):
+            try:
+                os.remove(processed_audio_path)
+                print(f"Temporary audio file {processed_audio_path} removed.")
+            except Exception as e:
+                print(f"Error removing temporary audio file {processed_audio_path}: {e}")
+        # 分割ファイルの掃除
+        for p in temp_chunk_paths:
+            if os.path.exists(p):
+                try:
+                    os.remove(p)
+                except Exception:
+                    pass
+
+def play_segment(evt: gr.SelectData, raw_ts_list, current_audio_path):
+    if not isinstance(raw_ts_list, list):
+        print(f"Warning: raw_ts_list is not a list ({type(raw_ts_list)}). Cannot play segment.")
+        return gr.Audio(value=None, label="Selected Segment")
+
+    if not current_audio_path:
+        print("No audio path available to play segment from.")
+        return gr.Audio(value=None, label="Selected Segment")
+
+    selected_index = evt.index[0]
+
+    if selected_index < 0 or selected_index >= len(raw_ts_list):
+        print(f"Invalid index {selected_index} selected for list of length {len(raw_ts_list)}.")
+        return gr.Audio(value=None, label="Selected Segment")
+
+    if not isinstance(raw_ts_list[selected_index], (list, tuple)) or len(raw_ts_list[selected_index]) != 2:
+        print(f"Warning: Data at index {selected_index} is not in the expected format [start, end].")
+        return gr.Audio(value=None, label="Selected Segment")
+
+    start_time_s, end_time_s = raw_ts_list[selected_index]
+    print(f"Attempting to play segment: {current_audio_path} from {start_time_s:.2f}s to {end_time_s:.2f}s")
+    segment_data = get_audio_segment(current_audio_path, start_time_s, end_time_s)
+
+    if segment_data:
+        print("Segment data retrieved successfully.")
+        return gr.Audio(value=segment_data, autoplay=True, label=f"Segment: {start_time_s:.2f}s - {end_time_s:.2f}s", interactive=False)
+    else:
+        print("Failed to get audio segment data.")
+        return gr.Audio(value=None, label="Selected Segment")
+
+def write_srt(segments, path):
+    def sec2srt(t):
+        h, rem = divmod(int(float(t)), 3600)
+        m, s = divmod(rem, 60)
+        ms = int((float(t) - int(float(t))) * 1000)
+        return f"{h:02}:{m:02}:{s:02},{ms:03}"
+    with open(path, "w", encoding="utf-8") as f:
+        for i, seg in enumerate(segments, 1):
+            f.write(f"{i}\n{sec2srt(seg[0])} --> {sec2srt(seg[1])}\n{seg[2]}\n\n")
+
+def write_vtt(segments, words, path):
+    def sec2vtt(t):
+        h, rem = divmod(int(float(t)), 3600)
+        m, s = divmod(rem, 60)
+        ms = int((float(t) - int(float(t))) * 1000)
+        return f"{h:02}:{m:02}:{s:02}.{ms:03}"
+    
+    with open(path, "w", encoding="utf-8") as f:
+        f.write("WEBVTT\n\n")
+        
+        word_idx = 0
+        for seg_idx, seg in enumerate(segments): # segmentにもインデックスが必要な場合に備えてenumerateする
+            s_start = float(seg[0])
+            s_end = float(seg[1])
+            # s_text = seg[2] # s_textはこの関数内では直接VTT出力に使われていない模様
+            
+            segment_words = []
+            temp_word_idx = word_idx # 現在のword_idxから探索を開始
+            while temp_word_idx < len(words):
+                w = words[temp_word_idx]
+                w_start_val = float(w[0])
+                w_end_val = float(w[1])
+                # 単語が現在のセグメントに完全に含まれるか、一部でも重なっていれば含める
+                # ここでは元のロジックを踏襲し、セグメント内に開始・終了がある単語を対象とする
+                if w_start_val >= s_start and w_end_val <= s_end:
+                    segment_words.append(w)
+                    if temp_word_idx == word_idx: # segment_words に追加された最初の単語なら word_idx を進める
+                        word_idx = temp_word_idx + 1
+                    temp_word_idx += 1
+                elif w_start_val < s_start and w_end_val > s_start: # 単語がセグメント開始をまたぐ場合
+                    # 必要であれば、このようなケースの単語も segment_words に含める処理を追加
+                    temp_word_idx += 1
+                elif w_start_val > s_end: # 単語の開始がセグメントの終了より後なら、このセグメントの単語は終わり
+                    break
+                else: # 上記以外 (単語がセグメントより完全に前など)
+                    if temp_word_idx == word_idx: # word_idx が進まない場合を避ける
+                         word_idx = temp_word_idx + 1
+                    temp_word_idx += 1
+            
+            # 各単語ごとにタイムスタンプを生成
+            for i, word_data in enumerate(segment_words):
+                w_start = float(word_data[0])
+                w_end = float(word_data[1])
+                
+                # 現在の単語を強調表示し、他の単語は通常表示
+                colored_text = ""
+                for j, other_word_data in enumerate(segment_words):
+                    if j == i: # 現在の単語 (i番目) を強調
+                        colored_text += f"<c.yellow><b>{other_word_data[2]}</b></c> "
+                    else:
+                        colored_text += f"{other_word_data[2]} "
+                
+                f.write(f"{sec2vtt(w_start)} --> {sec2vtt(w_end)}\n{colored_text.strip()}\n\n")
+
+def write_json(segments, words, path):
+    result = {"segments": []}
+    word_idx = 0
+    for s in segments:
+        s_start = float(s[0])
+        s_end = float(s[1])
+        s_text = s[2]
+        word_list = []
+        while word_idx < len(words):
+            w = words[word_idx]
+            w_start = float(w[0])
+            w_end = float(w[1])
+            if w_start >= s_start and w_end <= s_end:
+                word_list.append({"start": w_start, "end": w_end, "word": w[2]})
+                word_idx += 1
+            elif w_end < s_start:
+                word_idx += 1
+            else:
+                break
+        result["segments"].append({
+            "start": s_start,
+            "end": s_end,
+            "text": s_text,
+            "words": word_list
+        })
+    with open(path, "w", encoding="utf-8") as f:
+        json.dump(result, f, ensure_ascii=False, indent=2)
+
+def write_lrc(segments, path):
+    def sec2lrc(t):
+        m, s = divmod(float(t), 60)
+        return f"[{int(m):02}:{s:05.2f}]"
+    with open(path, "w", encoding="utf-8") as f:
+        for seg in segments:
+            f.write(f"{sec2lrc(seg[0])}{seg[2]}\n")
+
+article = (
+    "<p style='font-size: 1.1em;'>"
+    "このデモは <code><a href='https://huggingface.co/nvidia/parakeet-tdt-0.6b-v2' target='_blank'>parakeet-tdt-0.6b-v2</a></code> "
+    "（約6億パラメータ）を用いた高精度な英語音声文字起こしを実演します。"
+    "</p>"
+    "<p><strong style='color: red; font-size: 1.2em;'>主な特長:</strong></p>"
+    "<ul style='font-size: 1.1em;'>"    "    <li>自動句読点・大文字化</li>"
+    "    <li>単語レベルのタイムスタンプ（下表クリックで該当区間を再生）</li>"
+    "    <li>文字レベルのタイムスタンプ表示にも対応</li>"
+    "    <li>自動チャンク処理による <strong>長時間音声</strong> の効率的な文字起こし（数時間以上の音声にも対応）</li>"
+    "    <li>数字や歌詞など発話の多様なケースに高いロバスト性</li>"
+    "</ul>"
+    "<p style='font-size: 1.1em;'>"
+    "商用・非商用ともに <strong>ライセンス制限なく利用可能</strong> です。"
+    "</p>"
+    "<p style='text-align: center;'>"
+    "<a href='https://huggingface.co/nvidia/parakeet-tdt-0.6b-v2' target='_blank'>🎙️ モデル詳細</a> | "
+    "<a href='https://arxiv.org/abs/2305.05084' target='_blank'>📄 Fast&nbsp;Conformer 論文</a> | "
+    "<a href='https://arxiv.org/abs/2304.06795' target='_blank'>📚 TDT 論文</a> | "
+    "<a href='https://github.com/NVIDIA/NeMo' target='_blank'>🧑‍💻 NeMo リポジトリ</a>"
+    "</p>"
+)
+
+examples = [
+    ["data/example-yt_saTD1u8PorI.mp3"],
+]
+
+nvidia_theme = gr_themes.Default(
+    primary_hue=gr_themes.Color(
+        c50="#E6F1D9", c100="#CEE3B3", c200="#B5D58C", c300="#9CC766",
+        c400="#84B940", c500="#76B900", c600="#68A600", c700="#5A9200",
+        c800="#4C7E00", c900="#3E6A00", c950="#2F5600"
+    ),
+    neutral_hue="gray",
+    font=[gr_themes.GoogleFont("Inter"), "ui-sans-serif", "system-ui", "sans-serif"],
+).set()
+
+with gr.Blocks(theme=nvidia_theme) as demo:
+    model_display_name = MODEL_NAME.split('/')[-1] if '/' in MODEL_NAME else MODEL_NAME
+    gr.Markdown(f"<h1 style='text-align: center; margin: 0 auto;'>長時間対応 音声文字起こし ({model_display_name})</h1>")
+    gr.HTML(article)
+
+    current_audio_path_state = gr.State(None)
+    raw_timestamps_list_state = gr.State([])
+    session_dir_state = gr.State()
+    demo.load(start_session, outputs=[session_dir_state])
+
+    with gr.Tabs():
+        with gr.TabItem("Audio File"):
+            file_input = gr.Audio(sources=["upload"], type="filepath", label="Upload Audio File")
+            gr.Examples(examples=examples, inputs=[file_input], label="Example Audio Files (Click to Load)")
+            file_transcribe_btn = gr.Button("Transcribe Uploaded File", variant="primary")
+        
+        with gr.TabItem("Microphone"):
+            mic_input = gr.Audio(sources=["microphone"], type="filepath", label="Record Audio")
+            mic_transcribe_btn = gr.Button("Transcribe Microphone Input", variant="primary")
+
+    gr.Markdown("---")
+    gr.Markdown("<p><strong style='color: #FF0000; font-size: 1.2em;'>Transcription Results</strong></p>")
+
+    download_btn = gr.DownloadButton(label="Download Segment Transcript (CSV)", visible=False)
+    srt_btn = gr.DownloadButton(label="Download SRT", visible=False)
+    vtt_btn = gr.DownloadButton(label="Download VTT", visible=False)
+    json_btn = gr.DownloadButton(label="Download JSON", visible=False)
+    lrc_btn = gr.DownloadButton(label="Download LRC", visible=False)
+
+    with gr.Tabs():
+        with gr.TabItem("Segment View (Click row to play segment)"):
+            vis_timestamps_df = gr.DataFrame(
+                headers=["Start (s)", "End (s)", "Segment"],
+                datatype=["number", "number", "str"],
+                wrap=True,
+            )
+            selected_segment_player = gr.Audio(label="Selected Segment", interactive=False)
+        
+        with gr.TabItem("Word View"):
+            word_vis_df = gr.DataFrame(
+                headers=["Start (s)", "End (s)", "Word"],
+                datatype=["number", "number", "str"],
+                wrap=False,
+            )
+
+    mic_transcribe_btn.click(
+        fn=get_transcripts_and_raw_times,
+        inputs=[mic_input, session_dir_state],
+        outputs=[vis_timestamps_df, raw_timestamps_list_state, word_vis_df, current_audio_path_state, download_btn, srt_btn, vtt_btn, json_btn, lrc_btn],
+        api_name="transcribe_mic"
+    )
+
+    file_transcribe_btn.click(
+        fn=get_transcripts_and_raw_times,
+        inputs=[file_input, session_dir_state],
+        outputs=[vis_timestamps_df, raw_timestamps_list_state, word_vis_df, current_audio_path_state, download_btn, srt_btn, vtt_btn, json_btn, lrc_btn],
+        api_name="transcribe_file"
+    )
+
+    vis_timestamps_df.select(
+        fn=play_segment,
+        inputs=[raw_timestamps_list_state, current_audio_path_state],
+        outputs=[selected_segment_player],
+    )
+
+    demo.unload(end_session)
+
+if __name__ == "__main__":
+    print("Launching Gradio Demo...")
+    demo.queue(
+        max_size=5,
+        default_concurrency_limit=1  # イベントリスナーのデフォルト同時実行数を1に設定
+    )
+    demo.launch(
+        server_name="127.0.0.1",
+        server_port=7860,
+        share=False,
+        max_threads=1               # サーバー全体の同時処理スレッド数を1に設定
+    )

app_wsl.py

@@ -38,6 +38,33 @@ def end_session(request: gr.Request):
         # shutil.rmtree(session_dir)
     print(f"Session with hash {session_hash} ended.")
 
+def get_server_files(server_dir: str = None) -> List[str]:
+    """
+    サーバー側の指定ディレクトリ内の音声ファイルの一覧を取得する。
+
+    Args:
+        server_dir (str, optional): 検索するディレクトリ。Noneの場合はデフォルトの場所を使用。
+
+    Returns:
+        List[str]: 音声ファイルのパスのリスト
+    """
+    if server_dir is None:
+        server_dir = str(Path(__file__).parent / "data")
+    
+    audio_extensions = {".mp3", ".wav", ".m4a", ".ogg", ".flac"}
+    audio_files = []
+    
+    try:
+        for root, _, files in os.walk(server_dir):
+            for file in files:
+                if Path(file).suffix.lower() in audio_extensions:
+                    full_path = str(Path(root) / file)
+                    audio_files.append(full_path)
+        return sorted(audio_files)
+    except Exception as e:
+        print(f"Error scanning directory {server_dir}: {e}")
+        return []
+
 def get_audio_segment(audio_path, start_second, end_second):
     if not audio_path or not Path(audio_path).exists():
         print(f"Warning: Audio path '{audio_path}' not found or invalid for clipping.")
@@ -317,7 +344,7 @@ def split_audio_with_overlap(audio_path: str, session_dir: str, chunk_length_sec
     return chunk_paths
 
 @spaces.GPU
-def get_transcripts_and_raw_times(audio_path, session_dir):
+def get_transcripts_and_raw_times(audio_path, session_dir, progress=gr.Progress(track_tqdm=True)):
     """
     オーディオファイルを処理し、文字起こし結果を生成する。
     3時間を超える場合は60分ごとに分割し、オーバーラップ付きでASRを実行してマージする。
@@ -336,10 +363,9 @@ def get_transcripts_and_raw_times(audio_path, session_dir):
         if not transcribe_path or not duration_sec:
             return [], [], [], audio_path, gr.DownloadButton(visible=False), gr.DownloadButton(visible=False), gr.DownloadButton(visible=False), gr.DownloadButton(visible=False), gr.DownloadButton(visible=False)
 
-        processed_audio_path = transcribe_path if transcribe_path != audio_path else None
-
-        # 3時間超の場合は分割して逐次ASR
+        processed_audio_path = transcribe_path if transcribe_path != audio_path else None        # 3時間超の場合は分割して逐次ASR
         if duration_sec > 10800:
+            gr.Info("Audio is longer than 3 hours. Splitting into 1-hour chunks with overlap for transcription.", duration=5)
             chunk_paths = split_audio_with_overlap(transcribe_path, session_dir, chunk_length_sec=3600, overlap_sec=30)
             temp_chunk_paths = chunk_paths.copy()
             all_vis_data = []
@@ -347,7 +373,7 @@ def get_transcripts_and_raw_times(audio_path, session_dir):
             all_word_vis_data = []
             offset = 0.0
             prev_end = 0.0
-            for i, chunk_path in enumerate(chunk_paths):
+            for i, chunk_path in enumerate(progress.tqdm(chunk_paths, desc="Processing audio chunks")):
                 chunk_audio = AudioSegment.from_file(chunk_path)
                 chunk_duration = chunk_audio.duration_seconds
                 # ASR実行
@@ -597,14 +623,22 @@ with gr.Blocks(theme=nvidia_theme) as demo:
     current_audio_path_state = gr.State(None)
     raw_timestamps_list_state = gr.State([])
     session_dir_state = gr.State()
-    demo.load(start_session, outputs=[session_dir_state])
-
-    with gr.Tabs():
-        with gr.TabItem("Audio File"):
+    demo.load(start_session, outputs=[session_dir_state])    with gr.Tabs():
+        with gr.TabItem("Upload Audio"):
             file_input = gr.Audio(sources=["upload"], type="filepath", label="Upload Audio File")
             gr.Examples(examples=examples, inputs=[file_input], label="Example Audio Files (Click to Load)")
             file_transcribe_btn = gr.Button("Transcribe Uploaded File", variant="primary")
         
+        with gr.TabItem("Server Files"):
+            server_files = get_server_files()
+            server_file_dropdown = gr.Dropdown(
+                choices=server_files,
+                value=server_files[0] if server_files else None,
+                label="Select Audio File from Server",
+                type="value"
+            )
+            server_file_transcribe_btn = gr.Button("Transcribe Selected File", variant="primary")
+        
         with gr.TabItem("Microphone"):
             mic_input = gr.Audio(sources=["microphone"], type="filepath", label="Record Audio")
             mic_transcribe_btn = gr.Button("Transcribe Microphone Input", variant="primary")

transcribe_cli.py

@@ -0,0 +1,754 @@
+# coding: utf-8
+import torch
+import gc
+from pathlib import Path
+from pydub import AudioSegment
+# numpy は直接は使用されていませんが、pydubやNeMoの依存関係で間接的に必要になる可能性があります。
+# import numpy as np 
+import os
+import csv
+import json
+from typing import List, Tuple, Optional, Set # Python 3.9+ では Optional, Set は typing から不要な場合あり
+import argparse
+from nemo.collections.asr.models import ASRModel # NeMo ASRモデル
+
+# --- グローバル設定 ---
+MODEL_NAME = "nvidia/parakeet-tdt-0.6b-v2"
+TARGET_SAMPLE_RATE = 16000
+# 音声の長さに関する閾値 (秒)
+LONG_AUDIO_THRESHOLD_SECONDS = 480  # 8分 (この長さを超えると長尺設定を試みる)
+VERY_LONG_AUDIO_THRESHOLD_SECONDS = 10800 # 3時間 (この長さを超えるとチャンク分割処理)
+# チャンク分割時の設定
+CHUNK_LENGTH_SECONDS = 3600 # 1時間
+CHUNK_OVERLAP_SECONDS = 30  # 30秒
+# ★ 入力ファイルの優先順位付き拡張子リスト
+INPUT_PRIORITY_EXTENSIONS: List[str] = ['.wav', '.mp3', '.mp4']
+# ★ デフォルトで出力するフォーマットリスト
+DEFAULT_OUTPUT_FORMATS: List[str] = ["csv", "srt", "vtt", "json", "lrc"]
+
+
+# --- 音声前処理関数 ---
+def preprocess_audio_cli(audio_path_str: str, output_dir_for_temp_files: str) -> Tuple[Optional[str], Optional[str], Optional[float]]:
+    """
+    オーディオファイルの前処理（リサンプリング、モノラル変換）を行います。
+    成功した場合、(処理済みファイルパス, 表示用名, 音声長) を返します。
+    失敗した場合、(None, None, None) を返します。
+    """
+    try:
+        audio_file_path = Path(audio_path_str)
+        original_path_name = audio_file_path.name
+        audio_name_stem = audio_file_path.stem
+
+        print(f"  音声ファイルをロード中: {original_path_name}")
+        audio = AudioSegment.from_file(audio_path_str)
+        duration_sec = audio.duration_seconds
+        print(f"  音声長: {duration_sec:.2f} 秒")
+
+    except FileNotFoundError:
+        print(f"エラー: 音声ファイルが見つかりません: {audio_path_str}")
+        return None, None, None
+    except Exception as load_e: # pydub.exceptions.CouldntDecodeError などを含む
+        print(f"エラー: 音声ファイル '{original_path_name}' のロード/デコードに失敗しました: {load_e}")
+        return None, None, None
+
+    resampled = False
+    mono_converted = False
+
+    # リサンプリング処理
+    if audio.frame_rate != TARGET_SAMPLE_RATE:
+        try:
+            print(f"  リサンプリング中: {audio.frame_rate}Hz -> {TARGET_SAMPLE_RATE}Hz")
+            audio = audio.set_frame_rate(TARGET_SAMPLE_RATE)
+            resampled = True
+        except Exception as resample_e:
+            print(f"エラー: 音声のリサンプリングに失敗しました: {resample_e}")
+            return None, None, None
+
+    # モノラル変換処理
+    if audio.channels == 2:
+        try:
+            print("  モノラルに変換中 (2ch -> 1ch)")
+            audio = audio.set_channels(1)
+            mono_converted = True
+        except Exception as mono_e:
+            print(f"エラー: 音声のモノラル変換に失敗しました: {mono_e}")
+            return None, None, None
+    elif audio.channels > 2:
+        print(f"エラー: 音声チャンネルが {audio.channels} です。1ch(モノラル)または2ch(ステレオ)のみサポートしています。")
+        return None, None, None
+    elif audio.channels == 1:
+        print("  音声は既にモノラルです。")
+
+    processed_temp_file_path_obj = None # 一時ファイルのPathオブジェクト
+    # 前処理が行われた場合、一時ファイルに保存
+    if resampled or mono_converted:
+        try:
+            temp_suffix = "_preprocessed_temp.wav" # 一時ファイルとわかるような接尾辞
+            processed_temp_file_path_obj = Path(output_dir_for_temp_files, f"{audio_name_stem}{temp_suffix}")
+            
+            print(f"  前処理済み音声の一時保存先: {processed_temp_file_path_obj.name}")
+            audio.export(processed_temp_file_path_obj, format="wav")
+            
+            path_for_transcription = processed_temp_file_path_obj.as_posix() # 文字起こしに使用するパス
+            display_name_for_info = f"{original_path_name} (前処理済み)"
+        except Exception as export_e:
+            print(f"エラー: 前処理済み音声のエクスポートに失敗しました: {export_e}")
+            # エクスポート失敗時、もしファイルが作られていたら削除試行
+            if processed_temp_file_path_obj and processed_temp_file_path_obj.exists():
+                try: os.remove(processed_temp_file_path_obj)
+                except OSError: pass # 削除エラーはここでは致命的ではない
+            return None, None, None
+    else:
+        # 前処理が不要だった場合
+        print("  前処理は不要でした。元のファイルを使用します。")
+        path_for_transcription = audio_path_str
+        display_name_for_info = original_path_name
+
+    return path_for_transcription, display_name_for_info, duration_sec
+
+
+# --- 文字起こしコア関数 ---
+def transcribe_audio_cli(
+    transcribe_path_str: str,
+    model: ASRModel,
+    duration_sec: float, # この音声セグメントの長さ
+    device: str
+) -> Tuple[Optional[List], Optional[List], Optional[List]]:
+    """
+    指定されたオーディオファイルをNeMo ASRモデルで文字起こしします。
+    成功した場合、(セグメント情報リスト, RAWタイムスタンプリスト, 単語情報リスト) を返します。
+    失敗した場合、(None, None, None) を返します。
+    """
+    long_audio_settings_applied = False # 長尺設定が適用されたかどうかのフラグ
+    original_model_dtype = model.dtype # モデルの元のデータ型を保存 (通常は torch.float32)
+
+    try:
+        # CUDAデバイス使用時はメモリキャッシュをクリア
+        if device == 'cuda':
+            torch.cuda.empty_cache()
+            gc.collect()
+
+        model.to(device) # モデルを推論に使用するデバイスへ移動
+
+        # 音声長に応じてモデル設定を変更 (長尺音声対応)
+        if duration_sec > LONG_AUDIO_THRESHOLD_SECONDS:
+            try:
+                print(f"  情報: 音声長 ({duration_sec:.0f}s) が閾値 ({LONG_AUDIO_THRESHOLD_SECONDS}s) を超えるため、長尺音声向け設定を適用します。")
+                # NeMoのドキュメントやモデルカードで推奨される設定を確認すること
+                model.change_attention_model(self_attention_model="rel_pos_local_attn", att_context_size=[256, 256])
+                model.change_subsampling_conv_chunking_factor(1) # 例: FastConformer系
+                long_audio_settings_applied = True
+                if device == 'cuda': torch.cuda.empty_cache(); gc.collect() # 設定変更後のメモリ確保のため
+            except Exception as setting_e:
+                print(f"  警告: 長尺音声向け設定の適用に失敗しました: {setting_e}。デフォルト設定で続行します。")
+        
+        # bfloat16の使用 (GPUが対応し、かつfloat32より効率が良い場合)
+        if device == 'cuda' and torch.cuda.is_bf16_supported():
+            print("  情報: モデルを bfloat16 に変換して推論を実行します。")
+            model.to(torch.bfloat16)
+        elif model.dtype != original_model_dtype: # bf16非対応時やCPU時は元のdtypeに戻す
+             model.to(original_model_dtype)
+
+
+        if device == 'cuda':
+            # メモリ使用状況のログ（デバッグ用）
+            # allocated = torch.cuda.memory_allocated() / 1024**2
+            # reserved = torch.cuda.memory_reserved() / 1024**2
+            # print(f"  CUDAメモリ (文字起こし前): Allocated={allocated:.2f}MB, Reserved={reserved:.2f}MB")
+            pass
+        
+        print(f"  文字起こしを実行中 (デバイス: {device}, モデルdtype: {model.dtype})...")
+        # `batch_size` はモデルやVRAMに応じて調整。Parakeet系は大きめでも大丈夫な場合がある。
+        output = model.transcribe([transcribe_path_str], timestamps=True, batch_size=4) # batch_sizeは環境に応じて調整
+
+        # 結果の検証
+        if not output or not isinstance(output, list) or not output[0] or \
+           not hasattr(output[0], 'timestamp') or not output[0].timestamp or \
+           'segment' not in output[0].timestamp:
+            print("  エラー: 文字起こしに失敗したか、予期しない出力形式です。")
+            return None, None, None
+
+        # 結果の抽出
+        segment_timestamps = output[0].timestamp['segment']
+        # vis_data: 表示やCSV保存用のセグメントリスト [[start_str, end_str, segment_text], ...]
+        vis_data = [[f"{ts['start']:.2f}", f"{ts['end']:.2f}", ts['segment']] for ts in segment_timestamps]
+        # raw_times_data: 内部処理用のfloat型タイムスタンプ [[start_float, end_float], ...]
+        raw_times_data = [[ts['start'], ts['end']] for ts in segment_timestamps]
+
+        word_timestamps_raw = output[0].timestamp.get("word", []) # 単語タイムスタンプがない場合もある
+        # word_vis_data: 単語ごとの情報リスト [[start_str, end_str, word_text], ...]
+        word_vis_data = [
+            [f"{w['start']:.2f}", f"{w['end']:.2f}", w["word"]]
+            for w in word_timestamps_raw if isinstance(w, dict) and 'start' in w and 'end' in w and 'word' in w
+        ]
+        print("  文字起こし完了。")
+        return vis_data, raw_times_data, word_vis_data
+
+    except torch.cuda.OutOfMemoryError as oom_e:
+        print(f"  致命的エラー: CUDAメモリ不足です。 {oom_e}")
+        # allocated = torch.cuda.memory_allocated() / 1024**2
+        # reserved = torch.cuda.memory_reserved() / 1024**2
+        # print(f"    現在のCUDAメモリ: Allocated={allocated:.2f}MB, Reserved={reserved:.2f}MB")
+        print("    バッチサイズを小さくする、他のGPU利用アプリを終了するなどの対策を試みてください。")
+        return None, None, None
+    except Exception as e:
+        print(f"  エラー: 文字起こし処理中に予期せぬエラーが発生しました: {e}")
+        import traceback
+        traceback.print_exc() # 詳細なエラー情報を表示
+        return None, None, None
+    finally:
+        # モデルの状態を元に戻す (長尺設定、データ型、デバイス)
+        if long_audio_settings_applied:
+            try:
+                print("  長尺音声向け設定を元に戻しています。")
+                model.change_attention_model(self_attention_model="rel_pos") # デフォルトに戻す例
+                model.change_subsampling_conv_chunking_factor(-1)    # デフォルトに戻す例
+            except Exception as revert_e:
+                print(f"  警告: 長尺音声設定の復元に失敗: {revert_e}")
+        
+        model.to(original_model_dtype) # 元のデータ型に戻す
+        if model.device.type != 'cpu': # 常にCPUに戻してメモリ解放を試みる
+            model.cpu()
+        
+        if device == 'cuda':
+            torch.cuda.empty_cache()
+            gc.collect()
+            # print("  CUDAキャッシュをクリアしました。")
+
+
+# --- 結果保存関数 ---
+def save_transcripts_cli(output_dir_str: str, audio_file_stem: str,
+                         vis_data: List, word_vis_data: List, formats: Optional[List[str]] = None):
+    """
+    文字起こし結果を指定された形式で保存します。
+    """
+    if formats is None: # formatsが指定されなかった場合はデフォルトを使用
+        formats_to_save = DEFAULT_OUTPUT_FORMATS
+    else:
+        formats_to_save = formats
+
+    output_dir_path = Path(output_dir_str)
+    output_dir_path.mkdir(parents=True, exist_ok=True) # 保存先ディレクトリがなければ作成
+    saved_files_count = 0
+
+    print(f"  結果を保存中 (対象形式: {', '.join(formats_to_save)})...")
+    try:
+        if "csv" in formats_to_save:
+            csv_file_path = output_dir_path / f"{audio_file_stem}.csv"
+            csv_headers = ["Start (s)", "End (s)", "Segment"]
+            with open(csv_file_path, 'w', newline='', encoding='utf-8') as f:
+                writer = csv.writer(f); writer.writerow(csv_headers); writer.writerows(vis_data)
+            print(f"    CSVファイルを保存: {csv_file_path.name}"); saved_files_count +=1
+        if "srt" in formats_to_save:
+            srt_file_path = output_dir_path / f"{audio_file_stem}.srt"
+            write_srt(vis_data, srt_file_path) # ヘルパー関数呼び出し
+            print(f"    SRTファイルを保存: {srt_file_path.name}"); saved_files_count +=1
+        if "vtt" in formats_to_save:
+            vtt_file_path = output_dir_path / f"{audio_file_stem}.vtt"
+            write_vtt(vis_data, word_vis_data, vtt_file_path) # ヘルパー関数呼び出し
+            print(f"    VTTファイルを保存: {vtt_file_path.name}"); saved_files_count +=1
+        if "json" in formats_to_save:
+            json_file_path = output_dir_path / f"{audio_file_stem}.json"
+            write_json(vis_data, word_vis_data, json_file_path) # ヘルパー関数呼び出し
+            print(f"    JSONファイルを保存: {json_file_path.name}"); saved_files_count +=1
+        if "lrc" in formats_to_save:
+            lrc_file_path = output_dir_path / f"{audio_file_stem}.lrc"
+            write_lrc(vis_data, lrc_file_path) # ヘルパー関数呼び出し
+            print(f"    LRCファイルを保存: {lrc_file_path.name}"); saved_files_count +=1
+        
+        if saved_files_count == 0 and formats_to_save: # 何も保存されなかった場合
+            print(f"  警告: 指定されたフォーマット {formats_to_save} でのファイルの保存は行われませんでした（対象なし等）。")
+    except Exception as e:
+        print(f"  エラー: 文字起こしファイルの保存中にエラーが発生しました: {e}")
+
+# --- 書き出しヘルパー関数群 (SRT, VTT, JSON, LRC) ---
+def write_srt(segments: List, path: Path):
+    """SRTファイル形式で書き出します。"""
+    def sec2srt(t_float: float) -> str:
+        h, rem = divmod(int(t_float), 3600); m, s = divmod(rem, 60)
+        ms = int((t_float - int(t_float)) * 1000)
+        return f"{h:02}:{m:02}:{s:02},{ms:03}"
+    with open(path, "w", encoding="utf-8") as f:
+        for i, seg_list in enumerate(segments, 1): # segmentsは [[start_str, end_str, text], ...]
+            f.write(f"{i}\n{sec2srt(float(seg_list[0]))} --> {sec2srt(float(seg_list[1]))}\n{seg_list[2]}\n\n")
+
+def write_vtt(segments: List, words: List, path: Path):
+    """VTTファイル形式で書き出します (単語ごとのタイムスタンプベース)。"""
+    def sec2vtt(t_float: float) -> str:
+        h, rem = divmod(int(t_float), 3600); m, s = divmod(rem, 60)
+        ms = int((t_float - int(t_float)) * 1000)
+        return f"{h:02}:{m:02}:{s:02}.{ms:03}"
+    with open(path, "w", encoding="utf-8") as f:
+        f.write("WEBVTT\n\n")
+        current_line: List[str] = []; line_start_time: Optional[float] = None
+        MAX_WORDS_PER_LINE = 7 # 1行あたりの最大単語数（調整可能）
+
+        if not words: # 単語情報がない場合はセグメント情報で代替
+            print("    VTT書き出し: 単語情報がないため、セグメント情報を使用します。")
+            for i, seg_list in enumerate(segments, 1):
+                 f.write(f"NOTE Segment {i}\n")
+                 f.write(f"{sec2vtt(float(seg_list[0]))} --> {sec2vtt(float(seg_list[1]))}\n{seg_list[2]}\n\n")
+            return
+
+        for i, word_data in enumerate(words): # wordsは [[start_str, end_str, word_text], ...]
+            w_start = float(word_data[0]); w_text = word_data[2]
+            if line_start_time is None: line_start_time = w_start
+            current_line.append(w_text)
+            
+            # 次の単語があるか、または現在の行が最大単語数に達したか、最後の単語か
+            next_word_start = float(words[i+1][0]) if i + 1 < len(words) else w_start + 999 # 適当な大きな値
+            current_word_end = float(word_data[1])
+
+            # 行区切りの条件
+            if len(current_line) >= MAX_WORDS_PER_LINE or \
+               (next_word_start - current_word_end > 1.0) or \
+               i == len(words) - 1: # 最後の単語
+                
+                line_end_time = current_word_end
+                if line_start_time is not None : # line_start_timeが設定されていることを確認
+                    f.write(f"{sec2vtt(line_start_time)} --> {sec2vtt(line_end_time)}\n")
+                    f.write(" ".join(current_line) + "\n\n")
+                current_line = []; line_start_time = None
+
+def write_json(segments: List, words: List, path: Path):
+    """JSONファイル形式で書き出します (セグメントとそれに含まれる単語情報)。"""
+    result = {"segments": []}; word_idx = 0
+    for seg_data in segments: # segmentsは [[start_str, end_str, text], ...]
+        s_start_time = float(seg_data[0]); s_end_time = float(seg_data[1]); s_text = seg_data[2]
+        segment_words_list: List[dict] = []; temp_current_word_idx = word_idx
+        
+        if words: # 単語情報がある場合のみ処理
+            while temp_current_word_idx < len(words):
+                w_data = words[temp_current_word_idx]; w_start_time = float(w_data[0]); w_end_time = float(w_data[1])
+                # 単語が現在のセグメントに（ほぼ）含まれるか
+                if w_start_time >= s_start_time and w_end_time <= s_end_time + 0.1: # 終了時間に少し許容
+                    segment_words_list.append({"start": w_start_time, "end": w_end_time, "word": w_data[2]})
+                    temp_current_word_idx += 1
+                elif w_start_time < s_start_time : # 単語がセグメントより前に開始 (word_idxを進める)
+                    temp_current_word_idx += 1
+                elif w_start_time > s_end_time: # 単語がセグメントより後に開始 (このセグメントの単語は終わり)
+                    break
+                else: # その他のケース (例: 単語がセグメントをまたぐなど)
+                    temp_current_word_idx += 1
+            word_idx = temp_current_word_idx # word_idxを更新
+        
+        result["segments"].append({"start": s_start_time, "end": s_end_time, "text": s_text, "words": segment_words_list})
+    with open(path, "w", encoding="utf-8") as f:
+        json.dump(result, f, ensure_ascii=False, indent=2)
+
+def write_lrc(segments: List, path: Path):
+    """LRCファイル形式で書き出します。"""
+    def sec2lrc(t_float: float) -> str:
+        m, s = divmod(float(t_float), 60)
+        return f"[{int(m):02d}:{s:05.2f}]" # mm:ss.xx (xxは100分の1秒)
+    with open(path, "w", encoding="utf-8") as f:
+        for seg_list in segments: # segmentsは [[start_str, end_str, text], ...]
+            f.write(f"{sec2lrc(float(seg_list[0]))}{seg_list[2]}\n")
+
+
+# --- 音声分割関数 ---
+def split_audio_with_overlap_cli(
+    audio_path_str: str,
+    output_dir_for_chunks: str, # チャンクファイルの一時保存先
+    chunk_length_sec: int = CHUNK_LENGTH_SECONDS,
+    overlap_sec: int = CHUNK_OVERLAP_SECONDS
+) -> List[str]:
+    """
+    音声ファイルを指定された長さのチャンクにオーバーラップ付きで分割し、
+    分割された一時チャンクファイルのパスリストを返します。
+    """
+    print(f"  音声分割中: 基本チャンク長 {chunk_length_sec}s, オーバーラップ {overlap_sec}s")
+    try: audio = AudioSegment.from_file(audio_path_str)
+    except Exception as e:
+        print(f"  エラー: 音声ファイル '{Path(audio_path_str).name}' のロード中にエラー（分割処理）: {e}")
+        return [] # 空リストを返して処理中断
+    
+    duration_ms = len(audio); chunk_length_ms = chunk_length_sec * 1000; overlap_ms = overlap_sec * 1000
+    chunk_paths_list: List[str] = []; start_ms = 0; chunk_idx = 0
+    audio_file_stem = Path(audio_path_str).stem # 元のファイル名（拡張子なし）
+
+    while start_ms < duration_ms:
+        # チャンクの実際の開始位置 (オーバーラップを考慮)
+        actual_chunk_start_ms = max(0, start_ms - (overlap_ms if start_ms > 0 else 0) )
+        # チャンクの基本終了位置
+        base_chunk_end_ms = start_ms + chunk_length_ms
+        # チャンクの実際の終了位置 (オーバーラップを考慮し、音声長を超えない)
+        actual_chunk_end_ms = min(base_chunk_end_ms + (overlap_ms if base_chunk_end_ms < duration_ms else 0), duration_ms)
+
+        # 実際のチャンクの長さが0以下になる異常ケースを避ける
+        if actual_chunk_start_ms >= actual_chunk_end_ms :
+            if start_ms >= duration_ms: break # ループ終了条件
+            print(f"  警告: チャンク計算で予期せぬ状態。actual_start({actual_chunk_start_ms}) >= actual_end({actual_chunk_end_ms})。スキップします。")
+            start_ms += chunk_length_ms; continue # 次の基本開始位置へ
+
+        chunk_segment = audio[actual_chunk_start_ms:actual_chunk_end_ms]
+        
+        # 一時チャンクファイル名
+        chunk_file_name = f"{audio_file_stem}_chunk_{chunk_idx:03d}_temp.wav"
+        chunk_file_path_obj = Path(output_dir_for_chunks, chunk_file_name)
+        
+        try:
+            chunk_segment.export(chunk_file_path_obj, format="wav")
+            chunk_paths_list.append(chunk_file_path_obj.as_posix())
+            # print(f"    チャンク {chunk_idx:03d} ({actual_chunk_start_ms/1000:.2f}s - {actual_chunk_end_ms/1000:.2f}s) を保存: {chunk_file_name}")
+        except Exception as export_chunk_e:
+            print(f"  エラー: 一時チャンクファイル {chunk_file_name} のエクスポートに失敗: {export_chunk_e}")
+            # エラーが発生しても処理を続行し、可能な限り他のチャンクを処理する
+
+        start_ms += chunk_length_ms # 次の基本開始位置へ
+        chunk_idx += 1
+        
+    print(f"  音声を {len(chunk_paths_list)} 個のチャンクに分割しました。")
+    return chunk_paths_list
+
+
+# --- 単一ファイル処理のメインロジック ---
+def process_single_file(
+    input_file_path_obj: Path,
+    asr_model_instance: ASRModel,
+    device_to_use: str,
+    output_formats_list: List[str] # 保存するフォーマットのリスト
+) -> bool: # 成功ならTrue、失敗またはスキップ（CSV既存）ならTrue、致命的エラーならFalse
+    """指定された単一の音声/動画ファイルを前処理、文字起こし、結果保存まで行います。"""
+    input_file_str = input_file_path_obj.as_posix()
+    input_file_stem = input_file_path_obj.stem
+    # 出力ファイルと一時ファイルは入力ファイルと同じディレクトリに作成
+    output_and_temp_dir_str = input_file_path_obj.parent.as_posix()
+
+    print(f"\n======== ファイル処理開始: {input_file_path_obj.name} ========")
+
+    # スキップ処理: CSVファイルの存在確認 (指定された出力形式にcsvが含まれる場合のみ)
+    output_csv_path = input_file_path_obj.with_suffix('.csv')
+    if "csv" in output_formats_list and output_csv_path.exists():
+        print(f"スキップ: 出力CSVファイル '{output_csv_path.name}' は既に存在します。")
+        return True # スキップも処理対応済みとみなす
+
+    temp_preprocessed_audio_path_str: Optional[str] = None # 前処理された一時音声ファイルのパス
+    temp_chunk_file_paths_str_list: List[str] = []     # 分割された一時チャンクファイルのパスリスト
+
+    try:
+        # 1. 音声前処理
+        print(f"--- ステップ1/3: {input_file_stem} の音声前処理 ---")
+        processed_path_for_asr, _, duration_sec_val = preprocess_audio_cli(
+            input_file_str, output_and_temp_dir_str
+        )
+        if not processed_path_for_asr or duration_sec_val is None:
+            print(f"エラー: {input_file_path_obj.name} の前処理に失敗しました。このファイルの処理を中断します。")
+            return False # 前処理失敗は致命的
+        
+        # 前処理で一時ファイルが作成された場合、そのパスを記録 (後で削除するため)
+        if processed_path_for_asr != input_file_str:
+            temp_preprocessed_audio_path_str = processed_path_for_asr
+        
+        # 2. 文字起こし (長時間・超長時間対応含む)
+        print(f"--- ステッ���2/3: {input_file_stem} の文字起こし (音声長: {duration_sec_val:.2f}秒) ---")
+        
+        final_vis_data: Optional[List] = None
+        final_word_vis_data: Optional[List] = None
+
+        # 超長時間音声 (例: 3時間以上) の場合はチャンク分割して処理
+        if duration_sec_val > VERY_LONG_AUDIO_THRESHOLD_SECONDS:
+            print(f"  情報: 音声長が{VERY_LONG_AUDIO_THRESHOLD_SECONDS/3600:.1f}時間を超えるため、約{CHUNK_LENGTH_SECONDS/3600:.0f}時間ごとのチャンクに分割して処理します。")
+            
+            chunk_file_paths_str = split_audio_with_overlap_cli(
+                processed_path_for_asr, # 前処理済みのパスを使用
+                output_and_temp_dir_str,      # チャンクも同じディレクトリに一時保存
+                chunk_length_sec=CHUNK_LENGTH_SECONDS,
+                overlap_sec=CHUNK_OVERLAP_SECONDS
+            )
+            if not chunk_file_paths_str: # チャンク分割に失敗した場合
+                print(f"  エラー: {input_file_path_obj.name} のチャンク分割に失敗しました。このファイルの処理を中断します。")
+                return False 
+            
+            temp_chunk_file_paths_str_list = chunk_file_paths_str[:] # 後で削除するためコピー
+
+            all_vis_data_merged: List[List[str]] = []
+            all_word_vis_data_merged: List[List[str]] = []
+            current_global_time_offset_sec = 0.0 # 各チャンクの開始時刻のオフセット
+            last_global_segment_end_time_sec = 0.0 # オーバーラップ除去のための前のセグメントの終了時刻
+
+            for i, chunk_file_path_str in enumerate(temp_chunk_file_paths_str_list):
+                print(f"  チャンク {i+1}/{len(temp_chunk_file_paths_str_list)} ({Path(chunk_file_path_str).name}) を処理中...")
+                try:
+                    # チャンクの長さは長尺設定の閾値判定に使う。最大で CHUNK_LENGTH + CHUNK_OVERLAP 程度。
+                    estimated_chunk_duration_for_asr_settings = CHUNK_LENGTH_SECONDS + CHUNK_OVERLAP_SECONDS
+                    
+                    vis_data_chunk, _, word_vis_data_chunk = transcribe_audio_cli(
+                        chunk_file_path_str, 
+                        asr_model_instance, 
+                        estimated_chunk_duration_for_asr_settings,
+                        device_to_use
+                    )
+                    if not vis_data_chunk: # チャンクの文字起こし失敗
+                        print(f"  警告: チャンク {Path(chunk_file_path_str).name} の文字起こしに失敗しました。このチャンクをスキップします。")
+                        # このチャンクが処理できなかったとしても、次のチャンクの開始時間は進める
+                        current_global_time_offset_sec += CHUNK_LENGTH_SECONDS - (CHUNK_OVERLAP_SECONDS if i < len(temp_chunk_file_paths_str_list) - 1 else 0)
+                        continue 
+                    
+                    # タイムスタンプをグローバルに補正し、オーバーラップを除去
+                    # セグメントデータの処理
+                    for seg_row_list in vis_data_chunk:
+                        s_local_sec = float(seg_row_list[0]); e_local_sec = float(seg_row_list[1]); text_seg = seg_row_list[2]
+                        s_global_sec = s_local_sec + current_global_time_offset_sec
+                        e_global_sec = e_local_sec + current_global_time_offset_sec
+                        if s_global_sec >= last_global_segment_end_time_sec - 0.1 : # 0.1秒程度の許容
+                            all_vis_data_merged.append([f"{s_global_sec:.2f}", f"{e_global_sec:.2f}", text_seg])
+                            last_global_segment_end_time_sec = max(last_global_segment_end_time_sec, e_global_sec)
+                    
+                    # 単語データの処理 (同様にオフセットとオーバーラップ考慮)
+                    temp_last_word_global_end_time_sec = float(all_word_vis_data_merged[-1][1]) if all_word_vis_data_merged else 0.0
+                    if word_vis_data_chunk: # 単語情報がある場合のみ
+                        for word_row_list in word_vis_data_chunk:
+                            w_s_local_sec = float(word_row_list[0]); w_e_local_sec = float(word_row_list[1]); text_word = word_row_list[2]
+                            w_s_global_sec = w_s_local_sec + current_global_time_offset_sec
+                            w_e_global_sec = w_e_local_sec + current_global_time_offset_sec
+                            if w_s_global_sec >= temp_last_word_global_end_time_sec - 0.05: # わずかな重複は許容
+                                 all_word_vis_data_merged.append([f"{w_s_global_sec:.2f}", f"{w_e_global_sec:.2f}", text_word])
+                                 temp_last_word_global_end_time_sec = max(temp_last_word_global_end_time_sec, w_e_global_sec)
+
+                    # 次のチャンクのためのオフセット更新
+                    if i < len(temp_chunk_file_paths_str_list) - 1: # 最後のチャンク以外
+                        current_global_time_offset_sec += (CHUNK_LENGTH_SECONDS - CHUNK_OVERLAP_SECONDS)
+                    # else: # 最後のチャンク。次のオフセットはない。
+                except Exception as chunk_proc_e:
+                    print(f"  エラー: チャンク {Path(chunk_file_path_str).name} の処理中に予期せぬエラー: {chunk_proc_e}")
+                    # エラーでもオフセットは進める
+                    if i < len(temp_chunk_file_paths_str_list) - 1:
+                        current_global_time_offset_sec += (CHUNK_LENGTH_SECONDS - CHUNK_OVERLAP_SECONDS)
+            
+            final_vis_data = all_vis_data_merged
+            final_word_vis_data = all_word_vis_data_merged
+
+        else: # 3時間以内 (通常の長時間処理、または短時間処理)
+            vis_data_single, _, word_vis_data_single = transcribe_audio_cli(
+                processed_path_for_asr, 
+                asr_model_instance, 
+                duration_sec_val, 
+                device_to_use
+            )
+            if not vis_data_single: # 文字起こし失敗
+                print(f"エラー: {input_file_path_obj.name} の文字起こしに失敗しました。このファイルの処理を中断します。")
+                return False
+            
+            final_vis_data = vis_data_single
+            final_word_vis_data = word_vis_data_single
+        
+        # 3. 結果保存
+        if final_vis_data: # 文字起こし結果が空でない場合のみ保存
+            print(f"--- ステップ3/3: {input_file_stem} の文字起こし結果保存 ---")
+            save_transcripts_cli(output_and_temp_dir_str, input_file_stem, 
+                                 final_vis_data, final_word_vis_data if final_word_vis_data else [], # word_vis_dataがNoneの場合空リスト
+                                 formats=output_formats_list)
+        else:
+            print(f"情報: {input_file_path_obj.name} の文字起こし結果が空のため、ファイルは保存しませんでした。")
+        
+        return True # 処理成功
+
+    except Exception as e:
+        print(f"エラー: ファイル {input_file_path_obj.name} の処理中に予期せぬ大域的エラーが発生しました: {e}")
+        import traceback
+        traceback.print_exc() # 詳細なエラー情報を表示
+        return False # 致命的エラー
+    finally:
+        # 一時ファイルの削除
+        if temp_preprocessed_audio_path_str and Path(temp_preprocessed_audio_path_str).exists():
+            try:
+                os.remove(temp_preprocessed_audio_path_str)
+                print(f"  一時ファイル {Path(temp_preprocessed_audio_path_str).name} を削除しました。")
+            except OSError as e_os:
+                print(f"  警告: 一時ファイル {Path(temp_preprocessed_audio_path_str).name} の削除に失敗: {e_os}")
+        
+        for chunk_f_str in temp_chunk_file_paths_str_list:
+            chunk_f_path_obj = Path(chunk_f_str)
+            if chunk_f_path_obj.exists():
+                try:
+                    os.remove(chunk_f_path_obj)
+                    print(f"  一時チャンクファイル {chunk_f_path_obj.name} を削除しました。")
+                except OSError as e_os_chunk:
+                    print(f"  警告: 一時チャンクファイル {chunk_f_path_obj.name} の削除に失敗: {e_os_chunk}")
+        print(f"======== ファイル処理終了: {input_file_path_obj.name} ========\n")
+
+
+# --- ディレクトリ内ファイルの一括処理関数 ---
+def batch_process_directory(
+    target_dir_str: str,
+    asr_model_instance: ASRModel,
+    device_to_use: str,
+    output_formats: Optional[List[str]] = None # 保存するフォーマットのリスト
+):
+    """指定されたディレクトリ内の音声/動画ファイルを優先順位に従って処理し文字起こしを行います。"""
+    if output_formats is None: # argparseでデフォルトが設定されるので通常ここは通らない
+        output_formats_to_use = DEFAULT_OUTPUT_FORMATS
+    else:
+        output_formats_to_use = output_formats
+        
+    target_dir_path = Path(target_dir_str)
+    if not target_dir_path.is_dir():
+        print(f"エラー: 指定されたパス '{target_dir_str}' は有効なディレクトリではありません。処理を中止します。")
+        return
+
+    print(f"処理対象ディレクトリ: {target_dir_path.resolve()}")
+    print(f"入力ファイルの探索優先順位: {', '.join(INPUT_PRIORITY_EXTENSIONS)}")
+    print(f"出力ファイル形式: {', '.join(output_formats_to_use)}")
+
+    # 1. ディレクトリ内の全ファイルから、処理対象となりうるファイルの「ステム名」(拡張子なしの名前)を収集
+    all_files_in_dir = list(target_dir_path.iterdir()) # ディレクトリ内の全アイテムを取得
+    potential_stems: Set[str] = set() # 重複しないステム名を格納
+    for f_path_obj in all_files_in_dir:
+        # ファイルであり、かつ優先度リストの拡張子に合致する場合
+        if f_path_obj.is_file() and f_path_obj.suffix.lower() in INPUT_PRIORITY_EXTENSIONS:
+            potential_stems.add(f_path_obj.stem) # ステム名を追加
+
+    if not potential_stems:
+        print(f"情報: ディレクトリ '{target_dir_path.name}' に処理対象の拡張子 ({', '.join(INPUT_PRIORITY_EXTENSIONS)}) を持つファイルは見つかりませんでした。")
+        return
+
+    print(f"{len(potential_stems)} 個のユニークなファイル名（拡張子除く）が見つかりました。優先順位に従って処理対象ファイルを選択します...")
+
+    files_to_actually_process: List[Path] = [] # 実際に処理するファイルのリスト
+    # ステム名をソートして処理順序を一貫させる (任意)
+    for stem_name in sorted(list(potential_stems)):
+        selected_file_for_this_stem: Optional[Path] = None
+        # 優先順位リストに従ってファイルを探す
+        for ext_priority in INPUT_PRIORITY_EXTENSIONS: #例: ['.wav', '.mp3', '.mp4']
+            potential_file = target_dir_path / f"{stem_name}{ext_priority}"
+            if potential_file.exists() and potential_file.is_file():
+                selected_file_for_this_stem = potential_file
+                print(f"  ファイル名 '{stem_name}': '{potential_file.name}' を処理対象として選択 (優先度: {ext_priority})。")
+                break # このステム名に対する処理ファイルが見つかったのでループを抜ける
+        
+        if selected_file_for_this_stem:
+            files_to_actually_process.append(selected_file_for_this_stem)
+        # else: このブロックには通常入らないはず (potential_stems の作り方から)
+
+    if not files_to_actually_process:
+        print("情報: 優先順位を適用した結果、実際に処理するファイルはありませんでした。")
+        return
+        
+    print(f"実際に処理するファイル数: {len(files_to_actually_process)} 個")
+    
+    processed_successfully_count = 0
+    skipped_due_to_existing_csv_count = 0 # process_single_file内で判定されるスキップ
+    failed_count = 0
+
+    for input_file_to_process_obj in files_to_actually_process:
+        # process_single_file は、内部でCSV存在によるスキップ判定も行う
+        # 戻り値: Trueなら成功またはスキップ、Falseなら失敗
+        
+        # スキップ判定をここでも行うことで、カウントを正確にする
+        is_skipped = False
+        if "csv" in output_formats_to_use:
+            output_csv_path_check = input_file_to_process_obj.with_suffix('.csv')
+            if output_csv_path_check.exists():
+                print(f"\n======== ファイル処理開始: {input_file_to_process_obj.name} ========") # process_single_file内のログと合わせる
+                print(f"スキップ: 出力CSVファイル '{output_csv_path_check.name}' は既に存在します。")
+                print(f"======== ファイル処理終了: {input_file_to_process_obj.name} ========\n")
+                skipped_due_to_existing_csv_count += 1
+                is_skipped = True
+        
+        if not is_skipped:
+            success_flag = process_single_file(
+                input_file_to_process_obj, 
+                asr_model_instance, 
+                device_to_use, 
+                output_formats_to_use
+            )
+            if success_flag:
+                processed_successfully_count += 1
+            else:
+                failed_count += 1
+            
+    print("\n======== 全ファイルのバッチ処理が完了しました ========")
+    total_considered = len(files_to_actually_process)
+    print(f"総対象ファイル数（優先度選択後）: {total_considered}")
+    print(f"  処理成功ファイル数: {processed_successfully_count}")
+    print(f"  CSV既存によりスキップされたファイル数: {skipped_due_to_existing_csv_count}")
+    print(f"  処理失敗ファイル数: {failed_count}")
+
+
+# --- スクリプト実行のエントリポイント ---
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(
+        description="指定されたディレクトリ内の音声/動画ファイルをNVIDIA Parakeet ASRモデルで文字起こしします。\n"
+                    f"同じ名前のファイルが複数ある場合、{' > '.join(INPUT_PRIORITY_EXTENSIONS)} の優先順位で処理します。",
+        formatter_class=argparse.RawTextHelpFormatter # ヘルプメッセージの改行を保持
+    )
+    parser.add_argument(
+        "target_directory",
+        type=str,
+        help="処理対象のファイルが含まれるディレクトリのパス。\n例: /mnt/data/my_videos  または  C:\\Users\\YourName\\Videos"
+    )
+    # --extension 引数は削除 (ファイル選択ロジック変更のため)
+    parser.add_argument(
+        "--formats",
+        type=str,
+        default=",".join(DEFAULT_OUTPUT_FORMATS), # デフォルトは全形式
+        help=(f"出力する文字起こしファイルの形式をカンマ区切りで指定。\n"
+              f"例: csv,srt (デフォルト: {','.join(DEFAULT_OUTPUT_FORMATS)})\n"
+              f"利用可能な形式: {','.join(DEFAULT_OUTPUT_FORMATS)}")
+    )
+    parser.add_argument(
+        "--device",
+        type=str,
+        default=None, # 指定がなければ自動判別
+        choices=['cuda', 'cpu'],
+        help="使用するデバイスを指定 (cuda または cpu)。\n指定がない場合は自動的にCUDAが利用可能か判別します。"
+    )
+    
+    args = parser.parse_args()
+
+    # デバイス選択処理
+    if args.device: # ユーザーが明示的にデバイスを指定した場合
+        selected_device = args.device
+    else: # 自動判別
+        selected_device = "cuda" if torch.cuda.is_available() else "cpu"
+    
+    print(f"使用デバイス: {selected_device.upper()}")
+    if selected_device == "cuda":
+        if not torch.cuda.is_available(): # CUDA指定だが利用不可の場合
+            print("警告: CUDAが指定されましたが、システムで利用できません。CPUを使用します。")
+            selected_device = "cpu"
+        else:
+            try:
+                print(f"CUDAデバイス名: {torch.cuda.get_device_name(0)}")
+            except Exception as e_cuda_name:
+                print(f"CUDAデバイス名の取得に失敗しました: {e_cuda_name}")
+
+
+    # ASRモデルのロード (スクリプト開始時に一度だけ)
+    print(f"ASRモデル '{MODEL_NAME}' をロードしています...")
+    asr_model_main: Optional[ASRModel] = None
+    try:
+        # `verbose=False` は NeMo のバージョンによってサポートされないため削除済み
+        asr_model_main = ASRModel.from_pretrained(model_name=MODEL_NAME)
+        asr_model_main.eval() # 評価モードに設定
+        print(f"モデル '{MODEL_NAME}' のロードが完了しました。")
+    except Exception as model_load_e:
+        print(f"致命的エラー: ASRモデル '{MODEL_NAME}' のロードに失敗しました: {model_load_e}")
+        print("  考えられる原因:")
+        print("    - NVIDIA NeMo Toolkit (nemo_toolkit[asr]) が正しくインストールされていない。")
+        print("    - インターネット接続がないか不安定で、モデルファイルのダウンロードに失敗した。")
+        print("    - モデル名が間違っているか、Hugging Face Hub等で利用できなくなっている。")
+        exit(1) # モデルロード失敗は致命的なので終了
+
+    # 出力フォーマットの解析と検証
+    # カンマ区切り文字列をリストに変換し、前後の空白を除去、小文字化
+    output_formats_requested = [fmt.strip().lower() for fmt in args.formats.split(',') if fmt.strip()]
+    # 有効なフォーマットのみを抽出
+    final_output_formats_to_use = [fmt for fmt in output_formats_requested if fmt in DEFAULT_OUTPUT_FORMATS]
+    
+    if not output_formats_requested and args.formats: # 何か入力したが全て無効だった場合など
+        print(f"警告: 指定された出力フォーマット '{args.formats}' に有効なものが含まれていません。")
+    
+    if not final_output_formats_to_use : # 結果的に有効なフォーマットが一つもなかった場合
+         print(f"情報: 有効な出力フォーマットが指定されなかったため、デフォルトの全形式 ({','.join(DEFAULT_OUTPUT_FORMATS)}) で出力します。")
+         final_output_formats_to_use = DEFAULT_OUTPUT_FORMATS
+
+
+    # メインのバッチ処理関数を呼び出し
+    batch_process_directory(
+        args.target_directory,
+        asr_model_main,
+        selected_device,
+        output_formats=final_output_formats_to_use
+    )