# [メタ情報]
# 識別子: 全自動で動画からvttファイルを生成する_exe
# 補足: 
# [/メタ情報]
要約：
このテキストは、長尺動画から全自動で高品質なWebVTT（.vtt）字幕を生成するPythonパイプラインスクリプトと、それをFinderから手軽に実行するためのAutomatorワークフローについて記述しています。

Pythonスクリプトは、FFmpegを用いて動画を無音区間や指定された秒数で精密に分割し、分割後のPTS（プレゼンテーションタイムスタンプ）を完全にリセットします。各分割動画はGoogle Gemini 3.6 Flash APIにアップロードされ、AIが高精度な文字起こしを行います。この際、システムプロンプトにより、日本語の不自然な空白除去、句読点の厳格な適用、フィラー除去、字幕のブロック時間管理、さらに単独話者か複数話者かに応じた話者ラベルの自動付与など、詳細な品質ルールが適用されます。API呼び出しの失敗時には自動リトライ機能も備わっています。

GeminiからのレスポンスはVTT形式で解析され、動画全体のタイムラインに合わせてタイムシフト補正後、すべてのブロックが結合され、連番が再付与されます。最終的なVTTファイルは元の動画と同じディレクトリに出力されます。

付属のAutomatorワークフローは、Finderで選択された動画ファイルに対し、指定されたPython仮想環境でこのスクリプトをTerminal経由で実行することで、GUIから字幕生成プロセスを自動的に開始できるよう設計されています。



/Users/XXXXXX/python_scripts/multimodal_vtt_pipeline.py

```
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
長尺動画の全自動マルチモーダルVTT字幕生成パイプライン
(FFmpeg精密分割・PTS完全リセット + Gemini 3.6 Flash + 自動リトライ + タイムシフト自動結合 + 日本語空白自動整形)
※ 最新SDK google-genai 対応版
"""

import os
import sys
import subprocess
import shutil
import re
import time
import tempfile
import warnings
warnings.filterwarnings("ignore", category=FutureWarning)
from google import genai
from google.genai import types

# ==========================================
# 設定とシステムプロンプト
# ==========================================
API_KEY = "[MY_API_KEYに入るべき文字列]"

SYSTEM_PROMPT = """添付された動画ファイルの音声を解析し、極めて正確な文字起こしを行い、指定されたWebVTT（.vtt）形式の構造（コードデータのみ）を出力してください。挨拶やマークダウン装飾、解説は一切出力しないでください。

[字幕構造ルール]
1行目: 通し番号（1から始まる整数）
2行目: タイムスタンプ（HH:MM:SS.mmm --> HH:MM:SS.mmm）
3行目: 日本語の字幕テキスト（文脈に応じた適切な句読点あり）
各ブロックの間には空行を1行入れる。

[日本語テキスト品質ルール（最重要）]
- 単語と単語の間、助詞（は、が、の、を、に等）の前後に半角スペースや全角スペースを絶対に入れないでください。通常の自然な日本語として文字を詰めて記述してください。
- 句読点（、や。）の前後にスペースを入れないでください。
- 「えー」「あのー」などの意味のないフィラーは自然に除去してください。
- 会話の抜け落ちがないよう、音声は一言一句漏らさず正確に文字起こししてください。

[タイムスタンプの厳格ルール]
- タイムスタンプは必ず渡された動画ファイルの冒頭「00:00:00.000（0秒）」からの相対経過時間で記述してください。
- 形式は必ず「HH:MM:SS.mmm --> HH:MM:SS.mmm」（時:分:秒.ミリ秒）としてください。
- 1つの字幕ブロックの継続時間は通常2秒〜6秒程度とし、10秒を超える長大な字幕は絶対に作らないでください。
- 前の字幕の終了時刻より後の字幕の開始時刻が逆戻り（逆行）しないようにしてください。

[話者区分の判断]
映像と音声を解析し、もし「複数人による会話（対談、インタビュー等）」であると判断した場合は、字幕テキストの先頭に必ず「[男性]：」「[女性]：」または「[話者A]：」「[話者B]：」などの話者ラベルを付与してください。
もし「単独の喋り手」による動画であると判断した場合は、話者ラベルは付与せず、通常の字幕テキストのみを出力してください。"""

SPLIT_BASE_SECONDS = 270.0  # 基本分割秒数 (4分30秒)

# ==========================================
# ユーティリティ関数
# ==========================================
def check_dependencies():
    """必要なコマンドやAPIキーが設定されているか確認する"""
    if not shutil.which("ffmpeg") or not shutil.which("ffprobe"):
        print("エラー: FFmpeg または FFprobe がインストールされていません。Homebrew等でインストールしてください。")
        sys.exit(1)
    
    if API_KEY == "ここにAPIキーを貼り付け" or not API_KEY.strip():
        print("エラー: APIキーが設定されていません。プログラム内部の API_KEY 変数を書き換えてください。")
        sys.exit(1)

def clean_japanese_spaces(text):
    """日本語文字間や句読点周りに混入した不自然なスペースを除去する"""
    jp_char = r'[\u3000-\u303F\u3040-\u309F\u30A0-\u30FF\u4E00-\u9FFF\uFF01-\uFF60]'
    for _ in range(3):
        text = re.sub(rf'({jp_char})\s+({jp_char})', r'\1\2', text)
    text = re.sub(r'\s+([、。！？，．）】」』])', r'\1', text)
    text = re.sub(r'([（【「『])\s+', r'\1', text)
    return text.strip()

def time_to_sec(ts, max_duration=None):
    """時間フォーマットの揺れ（時間省略やカンマ混入）を吸収して秒に変換"""
    ts = ts.strip().replace(',', '.')
    parts = ts.split(':')
    
    if len(parts) == 3:
        h_val = float(parts[0])
        m_val = float(parts[1])
        s_val = float(parts[2])
        if max_duration is not None and h_val > 0 and (h_val * 3600 > max_duration * 1.5):
            sec = h_val * 60 + m_val + s_val / 1000.0
        else:
            sec = h_val * 3600 + m_val * 60 + s_val
    elif len(parts) == 2:
        m_val = float(parts[0])
        s_val = float(parts[1])
        sec = m_val * 60 + s_val
    elif len(parts) == 1:
        sec = float(parts[0])
    else:
        raise ValueError("Invalid time format")
    
    return sec

def sec_to_time(sec):
    """秒（float）を HH:MM:SS.mmm 形式に変換"""
    if sec < 0:
        sec = 0.0
    h = int(sec // 3600)
    m = int((sec % 3600) // 60)
    s = int(sec % 60)
    ms = int(round((sec - int(sec)) * 1000))
    if ms >= 1000:
        s += 1
        ms -= 1000
    if s >= 60:
        s -= 60
        m += 1
    if m >= 60:
        m -= 60
        h += 1
    return f"{h:02d}:{m:02d}:{s:02d}.{ms:03d}"

def get_video_duration(video_path):
    cmd = ['ffprobe', '-v', 'error', '-show_entries', 'format=duration', 
           '-of', 'default=noprint_wrappers=1:nokey=1', video_path]
    result = subprocess.run(cmd, stdout=subprocess.PIPE, text=True)
    try:
        return float(result.stdout.strip())
    except:
        return 0.0

def get_silence_points(video_path):
    """無音開始ポイント（秒）のリストを取得"""
    cmd = ['ffmpeg', '-i', video_path, '-af', 'silencedetect=noise=-30dB:d=1.0', '-f', 'null', '-']
    result = subprocess.run(cmd, stderr=subprocess.PIPE, text=True)
    silence_points = []
    for line in result.stderr.split('\n'):
        match = re.search(r'silence_start: (\d+(\.\d+)?)', line)
        if match:
            silence_points.append(float(match.group(1)))
    return silence_points

def parse_and_shift_vtt(vtt_text, offset_sec, part_duration):
    """VTT文字列を解析し、テキストの空白整形および安全補正を行ってブロックを返す"""
    blocks = []
    lines = vtt_text.strip().split('\n')
    i = 0
    last_rel_end = 0.0
    
    while i < len(lines):
        line = lines[i].strip()
        if line == "WEBVTT" or line == "" or re.match(r'^\d+$', line) or line.startswith("```"):
            i += 1
            continue
        
        if "-->" in line:
            parts = line.split("-->")
            if len(parts) == 2:
                try:
                    rel_start = time_to_sec(parts[0], max_duration=part_duration)
                    rel_end = time_to_sec(parts[1], max_duration=part_duration)
                except:
                    i += 1
                    continue
                
                # --- タイムスタンプ安全補正 ---
                if rel_start > part_duration:
                    rel_start = min(last_rel_end + 0.5, part_duration)
                if rel_end > part_duration + 2.0:
                    rel_end = min(rel_start + 4.0, part_duration)
                if rel_end <= rel_start:
                    rel_end = rel_start + 4.0
                if (rel_end - rel_start) > 12.0:
                    rel_end = rel_start + 4.0
                if rel_start < last_rel_end:
                    rel_start = last_rel_end + 0.1
                    rel_end = max(rel_end, rel_start + 1.0)
                
                last_rel_end = rel_end
                start_sec = rel_start + offset_sec
                end_sec = rel_end + offset_sec
                
                i += 1
                text_lines = []
                while i < len(lines) and lines[i].strip() != "":
                    cleaned = lines[i].strip().replace("```vtt", "").replace("```", "")
                    if cleaned and not re.match(r'^\d+$', cleaned):
                        cleaned = clean_japanese_spaces(cleaned)
                        text_lines.append(cleaned)
                    i += 1
                
                if text_lines:
                    blocks.append({
                        "start": start_sec,
                        "end": end_sec,
                        "text": "\n".join(text_lines)
                    })
        else:
            i += 1
    return blocks

# ==========================================
# メイン処理
# ==========================================
def main():
    if len(sys.argv) < 2:
        print("エラー: 動画ファイルのパスを引数として渡してください。")
        sys.exit(1)

    video_path = sys.argv[1]
    if not os.path.exists(video_path):
        print(f"エラー: ファイルが見つかりません -> {video_path}")
        sys.exit(1)

    check_dependencies()
    
    client = genai.Client(api_key=API_KEY)

    print(f"🎬 処理開始: {os.path.basename(video_path)}")
    total_duration = get_video_duration(video_path)
    if total_duration == 0:
        print("エラー: 動画の長さを取得できませんでした。")
        sys.exit(1)

    # 1. 無音検出と分割ポイント計算
    print("🔍 無音区間を解析中...")
    silence_points = get_silence_points(video_path)
    split_points = [0.0]
    current_target = SPLIT_BASE_SECONDS
    
    while current_target < total_duration:
        valid_points = [p for p in silence_points if p > split_points[-1] and p <= current_target + 15]
        if valid_points:
            best_point = min(valid_points, key=lambda x: abs(x - current_target))
            split_points.append(best_point)
            current_target = best_point + SPLIT_BASE_SECONDS
        else:
            split_points.append(current_target)
            current_target += SPLIT_BASE_SECONDS
    split_points.append(total_duration)

    all_vtt_blocks = []
    
    # 一時フォルダを作成して分割処理
    with tempfile.TemporaryDirectory() as tmp_dir:
        part_count = len(split_points) - 1
        print(f"✂️  動画を {part_count} 個に分割して処理します。")

        for i in range(part_count):
            start = split_points[i]
            end = split_points[i+1]
            part_dur = end - start
            ext = os.path.splitext(video_path)[1]
            part_filename = f"part_{i+1}{ext}"
            part_path = os.path.join(tmp_dir, part_filename)
            
            print(f"\n[{i+1}/{part_count}] 動画の精密切り出し中 ({start:.1f}s - {end:.1f}s / 長さ: {part_dur:.1f}s)...")
            
            # ★ 修正ポイント：再エンコード処理を行い、内部のタイムスタンプ（PTS）を完全に0からにリセットする
            # 会話の抜け落ち（キーフレームによる切り出しズレ）を根絶します。
            split_cmd = [
                'ffmpeg', '-y',
                '-ss', str(start),
                '-i', video_path,
                '-t', str(part_dur),
                '-c:v', 'libx264', '-preset', 'ultrafast', '-crf', '28', # 映像は超高速・低画質でOK（解析用）
                '-c:a', 'aac', '-b:a', '128k',                           # 音声は確実に切り出す
                part_path
            ]
            subprocess.run(split_cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)

            # Gemini API 処理
            print(f"[{i+1}/{part_count}] APIへアップロード中...")
            try:
                uploaded_file = client.files.upload(file=part_path)
                
                print(f"[{i+1}/{part_count}] サーバーでの処理完了を待機中...")
                while True:
                    file_info = client.files.get(name=uploaded_file.name)
                    if file_info.state == "PROCESSING":
                        time.sleep(3)
                    elif file_info.state == "FAILED":
                        raise Exception("動画の処理に失敗しました。")
                    else:
                        break

                # ★ API側の指示に従い、最新機能を含む公式モデルにアップデート
                target_model = "gemini-3.6-flash"
                gen_config = types.GenerateContentConfig(
                    temperature=0.1
                )

                max_retries = 5
                retry_delay = 10.0
                response = None

                for attempt in range(1, max_retries + 1):
                    try:
                        print(f"[{i+1}/{part_count}] 高精度文字起こし（推論）を実行中... (試行 {attempt}/{max_retries})")
                        response = client.models.generate_content(
                            model=target_model,
                            contents=[uploaded_file, SYSTEM_PROMPT],
                            config=gen_config
                        )
                        break
                    except Exception as api_err:
                        err_str = str(api_err)
                        if attempt < max_retries and ("503" in err_str or "429" in err_str or "UNAVAILABLE" in err_str or "RESOURCE_EXHAUSTED" in err_str):
                            print(f"⚠️ API混雑/制限検知。{retry_delay:.0f}秒待機して再試行します... (エラー: {err_str[:80]})")
                            time.sleep(retry_delay)
                            retry_delay += 15.0
                        else:
                            raise api_err
                
                print(f"[{i+1}/{part_count}] VTTのパース・空白除去・タイムシフトを実行中...")
                blocks = parse_and_shift_vtt(response.text, start, part_dur)
                all_vtt_blocks.extend(blocks)

            except Exception as e:
                print(f"エラー発生 ([{i+1}/{part_count}]): {e}")
            finally:
                if 'uploaded_file' in locals():
                    print(f"[{i+1}/{part_count}] API上の動画ファイルを削除中...")
                    try:
                        client.files.delete(name=uploaded_file.name)
                    except:
                        pass

    # 最終VTTの構築（通し番号の再付与と結合）
    print("\n📝 最終的なVTTを結合し、連番を再付与中...")
    dir_name = os.path.dirname(video_path)
    base_name = os.path.splitext(os.path.basename(video_path))[0]
    out_path = os.path.join(dir_name, f"{base_name}_completed.vtt")

    # 全体ソート（開始時刻順）
    all_vtt_blocks.sort(key=lambda x: x['start'])

    with open(out_path, 'w', encoding='utf-8') as f:
        f.write("WEBVTT\n\n")
        for idx, block in enumerate(all_vtt_blocks, 1):
            start_str = sec_to_time(block['start'])
            end_str = sec_to_time(block['end'])
            f.write(f"{idx}\n")
            f.write(f"{start_str} --> {end_str}\n")
            f.write(f"{block['text']}\n\n")

    print(f"✅ 完了！ 保存先: {out_path}")
    
    try:
        subprocess.run([
            "osascript", "-e", 
            'display notification "字幕作成が完了しました！" with title "XXXXXXシステム"'
        ])
    except:
        pass

if __name__ == "__main__":
    main()
```

Automator
全自動VTT字幕作成.workflow
/Users/XXXXXX/Library/Services/全自動VTT字幕作成.workflow/
ファイルまたはフォルダ
Finder.app

```
on run {input, parameters}
    -- ▼ Pythonスクリプトを保存したパス
    set pythonScript to "/Users/XXXXXX/python_scripts/multimodal_vtt_pipeline.py"
    
    -- ▼ 仮想環境(venv)のPython
    set pythonBin to "/Users/XXXXXX/venv_vtt/bin/python"

    repeat with f in input
        set videoPath to POSIX path of f
        set qVideoPath to quoted form of videoPath
        set qPythonScript to quoted form of pythonScript
        
        -- ターミナルで実行するコマンドを組み立てる
        set cmd to "export PATH=\"/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin:/usr/sbin:/sbin\"; " & pythonBin & " " & qPythonScript & " " & qVideoPath
        
        -- ターミナルを起動してコマンドを実行（見える形）
        tell application "Terminal"
            activate
            do script cmd
        end tell
    end repeat
    
    return input
end run
```