Rust 結合 Python 發揮 OpenAI 與 FFmpeg 串流影音同步字幕 P2

import librosa
import numpy as np
import matplotlib.pyplot as plt
import librosa.display

# 載入音訊檔案
音訊檔案 = "演講.wav"
y, sr = librosa.load(音訊檔案, sr=16000)

# 生成梅爾聲譜圖(Mel spectrogram)
聲譜圖 = librosa.feature.melspectrogram(y=y, sr=sr)
對數聲譜圖 = librosa.power_to_db(聲譜圖, ref=np.max)

# 繪製聲譜圖
plt.figure(figsize=(10, 4))
librosa.display.specshow(對數聲譜圖, sr=sr, x_axis="time", y_axis="mel")
plt.colorbar(format="%+2.0f dB")
plt.title("梅爾聲譜圖")
plt.show()

# 安裝基本依賴
pip install openai-whisper ffmpeg-python numpy pydub

# 如果需要GPU加速（強烈建議用於生產環境）
pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118

import whisper
import torch
import numpy as np
from pydub import AudioSegment
import tempfile
import os

class WhisperTranscriber:
    def __init__(self, model_size="medium", device=None):
        """初始化Whisper轉錄器
        
        Args:
            model_size: 模型大小 (tiny, base, small, medium, large)
            device: 運算裝置 (cuda, cpu)
        """
        # 自動檢測是否有可用的GPU
        if device is None:
            self.device = "cuda" if torch.cuda.is_available() else "cpu"
        else:
            self.device = device
            
        print(f"使用 {self.device} 裝置載入 {model_size} 模型...")
        self.model = whisper.load_model(model_size).to(self.device)
        print("模型載入完成!")
    
    def enhance_audio(self, audio_path, output_path=None):
        """增強音訊品質以提高轉錄準確度
        
        Args:
            audio_path: 輸入音訊檔案路徑
            output_path: 輸出音訊檔案路徑，若為None則使用臨時檔案
            
        Returns:
            增強後的音訊檔案路徑
        """
        if output_path is None:
            # 建立臨時檔案
            temp_file = tempfile.NamedTemporaryFile(delete=False, suffix='.wav')
            output_path = temp_file.name
            temp_file.close()
            
        # 使用pydub增強音訊
        audio = AudioSegment.from_file(audio_path)
        
        # 標準化音量
        normalized_audio = audio.normalize()
        
        # 提高音量 (可選)
        enhanced_audio = normalized_audio + 3  # 增加3dB
        
        # 匯出增強後的音訊
        enhanced_audio.export(output_path, format="wav")
        
        return output_path
    
    def transcribe(self, audio_path, enhance=True, language=None):
        """轉錄音訊檔案為文字
        
        Args:
            audio_path: 音訊檔案路徑
            enhance: 是否增強音訊
            language: 指定語言程式碼 (如 "zh" 代表中文)，None表示自動檢測
            
        Returns:
            轉錄結果字典，包含文字、分段等資訊
        """
        # 視需要增強音訊
        if enhance:
            processed_audio = self.enhance_audio(audio_path)
        else:
            processed_audio = audio_path
            
        # 執行轉錄
        transcribe_options = {}
        if language:
            transcribe_options["language"] = language
            
        result = self.model.transcribe(processed_audio, **transcribe_options)
        
        # 如果使用了臨時檔案，清理它
        if enhance and processed_audio != audio_path:
            os.unlink(processed_audio)
            
        return result

def generate_webvtt(transcription_result, output_file):
    """將Whisper轉錄結果轉換為WebVTT格式字幕
    
    Args:
        transcription_result: Whisper轉錄結果
        output_file: 輸出WebVTT檔案路徑
    """
    with open(output_file, 'w', encoding='utf-8') as f:
        # 寫入WebVTT標頭
        f.write("WEBVTT\n\n")
        
        # 處理每個分段
        for i, segment in enumerate(transcription_result["segments"]):
            # 轉換時間格式 (從秒到 HH:MM:SS.mmm)
            start_time = format_timestamp(segment["start"])
            end_time = format_timestamp(segment["end"])
            
            # 寫入時間碼和文字
            f.write(f"{i+1}\n")
            f.write(f"{start_time} --> {end_time}\n")
            f.write(f"{segment['text'].strip()}\n\n")
            
def format_timestamp(seconds):
    """將秒數轉換為WebVTT時間戳格式 (HH:MM:SS.mmm)"""
    hours = int(seconds / 3600)
    minutes = int((seconds % 3600) / 60)
    seconds = seconds % 60
    milliseconds = int((seconds - int(seconds)) * 1000)
    
    return f"{hours:02d}:{minutes:02d}:{int(seconds):02d}.{milliseconds:03d}"

// Rust程式碼片段 - 在轉碼後呼叫Python字幕生成
use std::process::Command;

fn process_video(input_path: &str, output_dir: &str) -> Result<(), Box<dyn std::error::Error>> {
    // 執行FFmpeg HLS轉碼 (省略詳細實作)
    // ...
    
    // 提取音訊用於字幕生成
    let audio_path = format!("{}/audio.wav", output_dir);
    Command::new("ffmpeg")
        .args(&[
            "-i", input_path,
            "-vn", "-acodec", "pcm_s16le", 
            "-ar", "16000", "-ac", "1",
            &audio_path
        ])
        .status()?;
    
    // 呼叫Python字幕生成指令碼
    Command::new("python")
        .args(&[
            "generate_subtitles.py",
            &audio_path,
            &format!("{}/subtitles.vtt", output_dir)
        ])
        .status()?;
    
    println!("影片處理與字幕生成完成");
    Ok(())
}

#!/usr/bin/env python3
import sys
from whisper_transcriber import WhisperTranscriber, generate_webvtt

def main():
    if len(sys.argv) != 3:
        print("用法: generate_subtitles.py <音訊檔案> <輸出字幕檔>")
        sys.exit(1)
        
    audio_file = sys.argv[1]
    output_file = sys.argv[2]
    
    # 初始化轉錄器 (使用medium模型平衡準確度與速度)
    transcriber = WhisperTranscriber(model_size="medium")
    
    # 轉錄音訊
    print(f"正在轉錄 {audio_file}...")
    result = transcriber.transcribe(audio_file, enhance=True)
    
    # 生成WebVTT字幕
    generate_webvtt(result, output_file)
    ...

pip install openai-whisper

sudo apt-get install ffmpeg

import whisper

# 載入 Whisper 模型
model = whisper.load_model("base")

# 轉錄音訊檔案
result = model.transcribe("speech.wav")

# 輸出生成的文字稿
print("轉錄結果:", result["text"])

ffmpeg -i input.wav -af "highpass=f=200, lowpass=f=3000" output_clean.wav

ffmpeg -i input.wav -filter:a "volume=2.0" output_louder.wav

import whisper

# 載入「large」模型以獲得高準確度
model = whisper.load_model("large")

# 轉錄語音
result = model.transcribe("cleaned_audio.wav")
print("轉錄結果:", result["text"])

def get_appropriate_subtitle_format(deployment_context):
    """根據佈署環境選擇最合適的字幕格式"""
    if deployment_context == "web_streaming":
        return "WebVTT"  # 網頁播放最佳選擇
    elif deployment_context == "desktop_player":
        return "SRT"  # 桌面播放器通用格式
    elif deployment_context == "editing_workflow":
        return "JSON"  # 保留最多資訊供後續編輯
    else:
        return "WebVTT"  # 預設使用WebVTT

def convert_whisper_json_to_srt(json_data):
    """將Whisper JSON輸出轉換為SRT格式"""
    srt_content = ""
    subtitle_index = 1
    
    for segment in json_data["segments"]:
        start_time = format_timestamp(segment["start"])
        end_time = format_timestamp(segment["end"])
        text = segment["text"].strip()
        
        srt_content += f"{subtitle_index}\n"
        srt_content += f"{start_time} --> {end_time}\n"
        srt_content += f"{text}\n\n"
        
        subtitle_index += 1
    
    return srt_content

def format_timestamp(seconds):
    """將秒數轉換為SRT時間格式 (HH:MM:SS,mmm)"""
    hours = int(seconds / 3600)
    minutes = int((seconds % 3600) / 60)
    seconds = seconds % 60
    milliseconds = int((seconds - int(seconds)) * 1000)
    
    return f"{hours:02d}:{minutes:02d}:{int(seconds):02d},{milliseconds:03d}"

def adjust_subtitle_timing(srt_content, offset_seconds=0, stretch_factor=1.0):
    """調整SRT字幕的時間戳記
    
    引數:
        srt_content: SRT格式的字幕內容
        offset_seconds: 正值向後偏移，負值向前偏移
        stretch_factor: 大於1延長持續時間，小於1縮短持續時間
    """
    import re
    
    # 正規表示式比對SRT時間戳記
    time_pattern = re.compile(r'(\d{2}):(\d{2}):(\d{2}),(\d{3}) --> (\d{2}):(\d{2}):(\d{2}),(\d{3})')
    
    def adjust_timestamp(match):
        # 解析起始時間
        h1, m1, s1, ms1 = map(int, match.groups()[:4])
        start_seconds = h1*3600 + m1*60 + s1 + ms1/1000
        
        # 解析結束時間
        h2, m2, s2, ms2 = map(int, match.groups()[4:])
        end_seconds = h2*3600 + m2*60 + s2 + ms2/1000
        
        # 應用偏移和拉伸係數
        new_start = start_seconds * stretch_factor + offset_seconds
        new_end = end_seconds * stretch_factor + offset_seconds
        
        # 確保時間不為負
        new_start = max(0, new_start)
        new_end = max(new_start + 0.1, new_end)
        
        # 轉換回SRT格式
        return f"{format_timestamp(new_start)} --> {format_timestamp(new_end)}"
    
    # 替換所有時間戳記
    return time_pattern.sub(adjust_timestamp, srt_content)

def convert_srt_to_webvtt(srt_content):
    """將SRT格式轉換為WebVTT格式"""
    import re
    
    # 新增WebVTT頭部
    webvtt = "WEBVTT\n\n"
    
    # 分割SRT為字幕區塊
    subtitle_blocks = re.split(r'\n\s*\n', srt_content.strip())
    
    for block in subtitle_blocks:
        lines = block.split('\n')
        if len(lines) >= 3:
            # 跳過字幕編號
            timing = lines[1]
            # 將逗號改為點作為毫秒分隔符
            timing = timing.replace(',', '.')
            
            # 提取字幕文字（可能有多行）
            text = '\n'.join(lines[2:])
            
            webvtt += f"{timing}\n{text}\n\n"
    
    return webvtt

def optimize_chinese_subtitle_layout(text, max_chars_per_line=20):
    """最佳化中文字幕排版，自動分行避免過長"""
    import re
    
    # 標點符號處理，優先在標點後換行
    punctuation_pattern = r'[，。！？；：」』）]'
    
    # 如果字幕長度超過限制
    if len(text) > max_chars_per_line:
        # 尋找適合的斷句點
        matches = list(re.finditer(punctuation_pattern, text[:max_chars_per_line]))
        
        if matches:
            # 在最後一個標點後斷行
            last_punct = matches[-1].end()
            return text[:last_punct] + '\n' + optimize_chinese_subtitle_layout(text[last_punct:], max_chars_per_line)
        else:
            # 無適合標點，強制斷行
            return text[:max_chars_per_line] + '\n' + optimize_chinese_subtitle_layout(text[max_chars_per_line:], max_chars_per_line)
    
    return text

def integrate_subtitles_with_hls(video_m3u8_path, subtitle_path, output_m3u8_path):
    """將字幕整合到HLS串流中"""
    from m3u8 import M3U8
    import os
    
    # 讀取主播放列表
    m3u8_obj = M3U8(video_m3u8_path)
    
    # 取得字幕格式
    subtitle_ext = os.path.splitext(subtitle_path)[1].lower()
    subtitle_type = "SUBTITLES"
    
    # 設定字幕群組
    subtitle_group_id = "subs"
    
    # 新增字幕媒體資訊
    relative_subtitle_path = os.path.relpath(
        subtitle_path, 
        os.path.dirname(video_m3u8_path)
    )
    
    subtitle_media = {
        "type": subtitle_type,
        "group_id": subtitle_group_id,
        "name": "中文字幕",
        "default": True,
        "autoselect": True,
        "language": "zh-TW",
        "uri": relative_subtitle_path
    }
    
    # 新增到播放列表
    m3u8_obj.media.append(subtitle_media)
    
    # 更新變體串流以包含字幕
    for playlist in m3u8_obj.playlists:
        if not any(x.get("group_id") == subtitle_group_id for x in playlist.stream_info.get("subtitles", [])):
            if "subtitles" not in playlist.stream_info:
                playlist.stream_info["subtitles"] = []
            playlist.stream_info["subtitles"].append({"group_id": subtitle_group_id})
    
    # 寫入更新後的播放列表
    with open(output_m3u8_path, "w") as f:
        f.write(m3u8_obj.dumps())
    
    return output_m3u8_path

def process_multilingual_subtitles(audio_path, languages=["zh", "en", "ja"]):
    """處理多語言字幕生成與格式化"""
    import whisper
    
    # 載入Whisper模型
    model = whisper.load_model("large")
    results = {}
    
    for lang in languages:
        # 針對特定語言進行轉錄
        result = model.transcribe(audio_path, language=lang)
        
        # 轉換為SRT格式
        srt_content = convert_whisper_json_to_srt(result)
        
        # 特定語言的後處理
        if lang == "zh":
            # 中文字幕排版最佳化
            srt_content = apply_chinese_formatting(srt_content)
        elif lang == "ja":
            # 日文字幕處理
            srt_content = apply_japanese_formatting(srt_content)
        
        # 轉換為WebVTT
        vtt_content = convert_srt_to_webvtt(srt_content)
        
        # 儲存結果
        results[lang] = {
            "json": result,
            "srt": srt_content,
            "vtt": vtt_content
        }
    
    return results

def batch_process_subtitles(video_folder, output_folder, workers=4):
    """平行處理多個影片的字幕生成與整合"""
    from concurrent.futures import ProcessPoolExecutor
    import os
    
    # 取得所有影片檔案
    video_files = [f for f in os.listdir(video_folder) 
                  if f.endswith(('.mp4', '.m3u8', '.ts'))]
    
    def process_single_video(video_file):
        video_path = os.path.join(video_folder, video_file)
        
        # 提取音訊
        audio_path = extract_audio(video_path)
        
        # 生成字幕
        result = transcribe_with_whisper(audio_path)
        
        # 轉換為需要的格式
        srt_path = os.path.join(output_folder, os.path.splitext(video_file)[0] + ".srt")
        vtt_path = os.path.join(output_folder, os.path.splitext(video_file)[0] + ".vtt")
        
        with open(srt_path, "w", encoding="utf-8") as f:
            f.write(convert_whisper_json_to_srt(result))
        
        with open(vtt_path, "w", encoding="utf-8") as f:
            f.write(convert_srt_to_webvtt(convert_whisper_json_to_srt(result)))
        
        return {
            "video": video_file,
            "srt": srt_path,
            "vtt": vtt_path
        }
    
    # 使用多處理器平行處理
    with ProcessPoolExecutor(max_workers=workers) as executor:
        results = list(executor.map(process_single_video, video_files))
    
    return results

{
  "segments": [
    {
      "start": 0.0,
      "end": 2.5,
      "text": "歡迎來到這個教學影片。"
    },
    {
      "start": 2.7,
      "end": 5.8,
      "text": "今天我們將學習如何處理字幕檔。"
    }
  ]
}

1
00:00:00,000 --> 00:00:02,500
歡迎來到這個教學影片。

2
00:00:02,700 --> 00:00:05,800
今天我們將學習如何處理字幕檔。

import json

def json_to_srt(json_file, srt_file):
    with open(json_file, "r") as file:
        data = json.load(file)

    with open(srt_file, "w") as srt:
        for i, segment in enumerate(data["segments"], start=1):
            start_time = format_time(segment["start"])
            end_time = format_time(segment["end"])
            text = segment["text"]

            srt.write(f"{i}\n{start_time} --> {end_time}\n{text}\n\n")

def format_time(seconds):
    hours, remainder = divmod(seconds, 3600)
    minutes, seconds = divmod(remainder, 60)
    milliseconds = int((seconds - int(seconds)) * 1000)
    return f"{int(hours):02}:{int(minutes):02}:{int(seconds):02},{milliseconds:03}"

# 使用範例
json_to_srt("transcript.json", "subtitles.srt")

import os
import json

def batch_convert(json_dir, srt_dir):
    if not os.path.exists(srt_dir):
        os.makedirs(srt_dir)
        
    for filename in os.listdir(json_dir):
        if filename.endswith('.json'):
            json_path = os.path.join(json_dir, filename)
            srt_path = os.path.join(srt_dir, filename.replace('.json', '.srt'))
            
            try:
                json_to_srt(json_path, srt_path)
                print(f"轉換成功: {filename}")
            except Exception as e:
                print(f"轉換失敗: {filename} - {str(e)}")

def json_to_srt(json_file, srt_file):
    try:
        with open(json_file, "r", encoding="utf-8") as file:
            data = json.load(file)
        
        # 驗證 JSON 結構
        if "segments" not in data:
            raise ValueError("JSON 檔案缺少 'segments' 欄位")
            
        with open(srt_file, "w", encoding="utf-8") as srt:
            for i, segment in enumerate(data["segments"], start=1):
                # 驗證必要欄位
                if not all(key in segment for key in ["start", "end", "text"]):
                    print(f"警告: 片段 {i} 缺少必要欄位，已跳過")
                    continue
                    
                start_time = format_time(segment["start"])
                end_time = format_time(segment["end"])
                text = segment["text"].strip()
                
                srt.write(f"{i}\n{start_time} --> {end_time}\n{text}\n\n")
                
        return True
    except Exception as e:
        print(f"轉換過程發生錯誤: {str(e)}")
        return False

from concurrent.futures import ThreadPoolExecutor

def parallel_batch_convert(json_dir, srt_dir, max_workers=4):
    if not os.path.exists(srt_dir):
        os.makedirs(srt_dir)
        
    json_files = [f for f in os.listdir(json_dir) if f.endswith('.json')]
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        for filename in json_files:
            json_path = os.path.join(json_dir, filename)
            srt_path = os.path.join(srt_dir, filename.replace('.json', '.srt'))
            executor.submit(json_to_srt, json_path, srt_path)

def adjust_timing(json_file, srt_file, offset_seconds=0, speed_factor=1.0):
    with open(json_file, "r") as file:
        data = json.load(file)

    with open(srt_file, "w") as srt:
        for i, segment in enumerate(data["segments"], start=1):
            # 套用時間偏移和速度因子
            start = segment["start"] * speed_factor + offset_seconds
            end = segment["end"] * speed_factor + offset_seconds
            
            # 確保時間不為負數
            start = max(0, start)
            end = max(start + 0.1, end)  # 確保結束時間晚於開始時間
            
            start_time = format_time(start)
            end_time = format_time(end)
            text = segment["text"]

            srt.write(f"{i}\n{start_time} --> {end_time}\n{text}\n\n")