kan

transcribe_cli.py

@@ -11,6 +11,7 @@ import json
 from typing import List, Tuple, Optional, Set # Python 3.9+ では Optional, Set は typing から不要な場合あり
 import argparse
 import time # ★処理時間計測のために追加
+import sys # ★コマンドライン引数チェックのために追加
 from nemo.collections.asr.models import ASRModel # NeMo ASRモデル
 
 # --- グローバル設定 ---
@@ -252,32 +253,87 @@ def write_vtt(segments: List, words: List, path: Path):
         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 
-
-        if not words: 
-            print("    VTT書き出し: 単語情報がないため、セグメント情報を使用します。")
+        f.write("STYLE\n")
+        f.write("::cue(.current) { color: #ffff00; font-weight: bold; }\n")
+        f.write("::cue(.past) { color: #888888; }\n") 
+        f.write("::cue(.future) { color: #ffffff; }\n")
+        f.write("::cue(.line) { background: rgba(0,0,0,0.7); padding: 4px; }\n\n")
+        
+        if not words:
+            # フォールバック処理は同じ
             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")
+                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): 
-            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 : 
-                    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
+        # セグメント単位でグループ化してカラオケ風に
+        for seg_data in segments:
+            seg_start = float(seg_data[0])
+            seg_end = float(seg_data[1])
+            
+            # このセグメントに含まれる単語を特定
+            segment_words = []
+            for word_idx, word_data in enumerate(words):
+                word_start = float(word_data[0])
+                word_end = float(word_data[1])
+                if word_start >= seg_start - 0.1 and word_end <= seg_end + 0.1:
+                    segment_words.append((word_idx, word_data))
+            
+            if not segment_words:
+                continue
+                
+            # セグメント開始時刻から最初の単語開始まで（全て未来色）
+            first_word_start = float(segment_words[0][1][0])
+            if seg_start < first_word_start - 0.05:
+                line_parts = [f'<c.future>{w_data[2]}</c>' for _, w_data in segment_words]
+                f.write(f"{sec2vtt(seg_start)} --> {sec2vtt(first_word_start)}\n")
+                f.write(f'<c.line>{" ".join(line_parts)}</c>\n\n')
+            
+            # 各単語の処理
+            for local_idx, (global_word_idx, word_data) in enumerate(segment_words):
+                w_start = float(word_data[0])
+                w_end = float(word_data[1])
+                
+                # 単語再生中：現在の単語をハイライト
+                line_parts = []
+                for i, (_, w_data) in enumerate(segment_words):
+                    w_text = w_data[2]
+                    if i == local_idx:
+                        line_parts.append(f'<c.current>{w_text}</c>')
+                    elif i < local_idx:
+                        line_parts.append(f'<c.past>{w_text}</c>')
+                    else:
+                        line_parts.append(f'<c.future>{w_text}</c>')
+                
+                f.write(f"{sec2vtt(w_start)} --> {sec2vtt(w_end)}\n")
+                f.write(f'<c.line>{" ".join(line_parts)}</c>\n\n')
+                
+                # 単語終了から次の単語開始まで（無音期間）：過去・未来のみ
+                if local_idx < len(segment_words) - 1:  # 最後の単語でない場合
+                    next_word_start = float(segment_words[local_idx + 1][1][0])
+                    gap_duration = next_word_start - w_end
+                    
+                    if gap_duration > 0.05:  # 50ms以上の無音期間がある場合
+                        gap_line_parts = []
+                        for i, (_, w_data) in enumerate(segment_words):
+                            w_text = w_data[2]
+                            if i <= local_idx:  # 現在の単語まで（過去）
+                                gap_line_parts.append(f'<c.past>{w_text}</c>')
+                            else:  # 未来の単語
+                                gap_line_parts.append(f'<c.future>{w_text}</c>')
+                        
+                        f.write(f"{sec2vtt(w_end)} --> {sec2vtt(next_word_start)}\n")
+                        f.write(f'<c.line>{" ".join(gap_line_parts)}</c>\n\n')
+                else:
+                    # 最後の単語終了からセグメント終了まで（全て過去色）
+                    if w_end < seg_end - 0.05:
+                        line_parts = [f'<c.past>{w_data[2]}</c>' for _, w_data in segment_words]
+                        f.write(f"{sec2vtt(w_end)} --> {sec2vtt(seg_end)}\n")
+                        f.write(f'<c.line>{" ".join(line_parts)}</c>\n\n')
+                        
 
 def write_json(segments: List, words: List, path: Path):
     result = {"segments": []}; word_idx = 0
@@ -354,7 +410,6 @@ def process_single_file(
     input_file_stem = input_file_path_obj.stem
     output_and_temp_dir_str = input_file_path_obj.parent.as_posix()
     
-    # ★ファイル処理開始時刻を記録
     file_processing_start_time = time.time()
     actual_audio_duration_sec: Optional[float] = None 
     success_status = False 
@@ -449,7 +504,6 @@ def process_single_file(
         print(f"エラー: ファイル {input_file_path_obj.name} の処理中にエラーが発生しました: {e}")
         success_status = False
     finally:
-        # ★ファイル処理時間とサマリーをログに出力
         file_processing_end_time = time.time()
         time_taken_seconds = file_processing_end_time - file_processing_start_time
         proc_m = int(time_taken_seconds // 60)
@@ -473,7 +527,7 @@ def process_single_file(
             if Path(chunk_f_str).exists():
                 try: os.remove(chunk_f_str); print(f"  一時チャンクファイル {Path(chunk_f_str).name} を削除しました。")
                 except OSError as e_os_chunk: print(f"  警告: 一時チャンクファイル {Path(chunk_f_str).name} の削除に失敗: {e_os_chunk}")
-        print(f"======== ファイル処理終了: {input_file_path_obj.name} ========\n")
+        # process_single_file の最後では "ファイル処理終了" のログは batch_process_directory に任せる
         return success_status
 
 # --- ディレクトリ内ファイルの一括処理関数 ---
@@ -483,9 +537,7 @@ def batch_process_directory(
     device_to_use: str,
     output_formats: Optional[List[str]] = None 
 ):
-    # ★バッチ処理全体の開始時刻
     batch_start_time = time.time()
-
     if output_formats is None: 
         output_formats_to_use = DEFAULT_OUTPUT_FORMATS
     else:
@@ -529,18 +581,17 @@ def batch_process_directory(
     failed_count = 0
 
     for input_file_to_process_obj in files_to_actually_process:
+        print(f"\n======== ファイル処理開始: {input_file_to_process_obj.name} ========") # 各ファイルの開始ログ
         is_skipped_at_batch_level = False
-        if "csv" in output_formats_to_use: # CSVスキップ判定はバッチレベルで行う
+        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} ========")
                 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_at_batch_level = True
+                print(f"======== ファイル処理終了 (スキップ): {input_file_to_process_obj.name} ========\n") # スキップ時の終了ログ
         
         if not is_skipped_at_batch_level:
-            print(f"\n======== ファイル処理開始: {input_file_to_process_obj.name} ========") # process_single_file に移譲
             success_flag = process_single_file(
                 input_file_to_process_obj, 
                 asr_model_instance, 
@@ -551,6 +602,7 @@ def batch_process_directory(
                 processed_successfully_count += 1
             else:
                 failed_count += 1
+            # process_single_file内で "ファイル処理終了" ログが出力される
             
     print("\n======== 全ファイルのバッチ処理が完了しました ========")
     total_considered = len(files_to_actually_process)
@@ -559,7 +611,6 @@ def batch_process_directory(
     print(f"  CSV既存によりスキップされたファイル数: {skipped_due_to_existing_csv_count}")
     print(f"  処理失敗ファイル数: {failed_count}")
 
-    # ★バッチ処理全体の総所要時間を表示
     batch_end_time = time.time()
     total_batch_time_seconds = batch_end_time - batch_start_time
     batch_m = int(total_batch_time_seconds // 60)
@@ -568,28 +619,73 @@ def batch_process_directory(
 
 # --- スクリプト実行のエントリポイント ---
 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="処理対象のファイルが含まれるディレクトリのパス。"
-    )
-    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)。指定がなければ自動判別。"
-    )
-    args = parser.parse_args()
-
-    if args.device: selected_device = args.device
+    # ★ 引数処理とGUI分岐のための準備
+    target_directory_arg: Optional[str] = None
+    formats_arg_str: str = ",".join(DEFAULT_OUTPUT_FORMATS) # GUI時のデフォルト
+    device_arg_str: Optional[str] = None # GUI時のデフォルト (自動判別)
+
+    if len(sys.argv) == 1: # コマンドライン引数なしの場合
+        print("コマンドライン引数なしで起動されました。GUIでディレクトリを選択します。")
+        try:
+            import tkinter as tk
+            from tkinter import filedialog
+
+            def get_directory_from_gui_local() -> Optional[str]:
+                """GUIでディレクトリ選択ダイアログを表示し、選択されたパスを返す"""
+                root = tk.Tk()
+                root.withdraw() # メインウィンドウは表示しない
+                # ダイアログを最前面に表示する試み (環境による)
+                root.attributes('-topmost', True) 
+                selected_path = filedialog.askdirectory(title="処理対象のディレクトリを選択してください")
+                root.attributes('-topmost', False)
+                root.destroy() # Tkinterウィンドウを破棄
+                return selected_path if selected_path else None
+            
+            target_directory_arg = get_directory_from_gui_local()
+            if not target_directory_arg:
+                print("ディレクトリが選択されませんでした。処理を中止します。")
+                sys.exit(0) # 正常終了
+            # formats_arg_str と device_arg_str は初期化されたデフォルト値を使用
+            print(f"GUIで選択されたディレクトリ: {target_directory_arg}")
+            print(f"出力フォーマット (デフォルト): {formats_arg_str}")
+            # device_arg_strがNoneの場合、後続の処理で自動判別される
+
+        except ImportError:
+            print("エラー: GUIモードに必要なTkinterライブラリが見つかりません。")
+            print("Tkinterをインストールするか、コマンドライン引数を使用してスクリプトを実行してください。例:")
+            print(f"  python {Path(sys.argv[0]).name} /path/to/your/audio_directory")
+            sys.exit(1) # エラー終了
+        except Exception as e_gui:
+            print(f"GUIの表示中に予期せぬエラーが発生しました: {e_gui}")
+            sys.exit(1) # エラー終了
+    else: # コマンドライン引数がある場合
+        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="処理対象のファイルが含まれるディレクトリのパス。"
+        )
+        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)。指定がなければ自動判別。"
+        )
+        args = parser.parse_args() # sys.argv[1:] から解析
+        
+        target_directory_arg = args.target_directory
+        formats_arg_str = args.formats
+        device_arg_str = args.device
+
+    # --- 共通のセットアップ処理 ---
+    if device_arg_str: selected_device = device_arg_str
     else: selected_device = "cuda" if torch.cuda.is_available() else "cpu"
     print(f"使用デバイス: {selected_device.upper()}")
     if selected_device == "cuda":
@@ -606,17 +702,26 @@ if __name__ == "__main__":
         asr_model_main.eval() 
         print(f"モデル '{MODEL_NAME}' のロード完了。")
     except Exception as model_load_e:
-        print(f"致命的エラー: ASRモデル '{MODEL_NAME}' のロードに失敗: {model_load_e}"); exit(1)
+        print(f"致命的エラー: ASRモデル '{MODEL_NAME}' のロードに失敗: {model_load_e}"); sys.exit(1)
 
-    output_formats_requested = [fmt.strip().lower() for fmt in args.formats.split(',') if fmt.strip()]
+    output_formats_requested = [fmt.strip().lower() for fmt in formats_arg_str.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 output_formats_requested and formats_arg_str: 
+        print(f"警告: 指定された出力フォーマット '{formats_arg_str}' は無効です。")
     if not final_output_formats_to_use : 
          print(f"情報: 有効な出力フォーマットが指定されなかったため、デフォルトの全形式 ({','.join(DEFAULT_OUTPUT_FORMATS)}) で出力します。")
          final_output_formats_to_use = DEFAULT_OUTPUT_FORMATS
+    
+    # target_directory_arg が None でないことを確認 (GUIキャンセル時など)
+    if not target_directory_arg:
+        print("エラー: 処理対象のディレクトリが指定されていません。処理を中止します。")
+        sys.exit(1)
+    
+    if not asr_model_main: # 通常、モデルロード失敗で既にexitしているはずだが念のため
+        print("致命的エラー: ASRモデルがロードされていません。処理を中止します。")
+        sys.exit(1)
 
     batch_process_directory(
-        args.target_directory, asr_model_main, selected_device,
+        target_directory_arg, asr_model_main, selected_device,
         output_formats=final_output_formats_to_use
-    )
+    )