AI 字幕生成 2026：多语言实时翻译

音频输入
    ↓
┌────────────────────────────────────┐
│         AI 字幕生成引擎             │
│                                    │
│  1. 语音识别（ASR）                 │
│     └── Whisper 3 / GPT-4o Audio   │
│                                    │
│  2. 说话人分离（Diarization）       │
│     └── Pyannote Audio             │
│                                    │
│  3. 时间轴对齐                      │
│     └── Forced Alignment           │
│                                    │
│  4. 文本翻译（可选）                │
│     └── GPT-4o / NLLB-200          │
│                                    │
│  5. 字幕生成与格式化                │
│     └── SRT / VTT / ASS            │
│                                    │
│  6. 后处理优化                      │
│     ├── 标点恢复                    │
│     ├── 断句优化                    │
│     └── 专业术语纠正                │
└────────────────────────────────────┘
    ↓
字幕文件（多格式、多语言）

import whisper
from whisper import load_model

# 加载模型（large-v3 为最新）
model = whisper.load_model("large-v3")

# 离线识别
result = model.transcribe(
    "video.mp4",
    language="zh",
    task="transcribe",
    verbose=True
)

# 提取字幕
for segment in result["segments"]:
    print(f"[{segment['start']:.2f} - {segment['end']:.2f}] "
          f"{segment['text']}")

import whisper
import queue
import threading
import pyaudio

class RealtimeSubtitle:
    """实时字幕生成"""
    
    def __init__(self, model_size="medium"):
        self.model = whisper.load_model(model_size)
        self.audio_queue = queue.Queue()
        self.is_running = False
    
    def start(self, language="zh"):
        """启动实时字幕"""
        self.is_running = True
        
        # 启动音频采集线程
        audio_thread = threading.Thread(
            target=self._capture_audio
        )
        audio_thread.start()
        
        # 启动转录线程
        transcribe_thread = threading.Thread(
            target=self._transcribe_loop,
            args=(language,)
        )
        transcribe_thread.start()
    
    def _capture_audio(self):
        """采集音频"""
        CHUNK = 1024
        FORMAT = pyaudio.paInt16
        CHANNELS = 1
        RATE = 16000
        
        p = pyaudio.PyAudio()
        stream = p.open(
            format=FORMAT,
            channels=CHANNELS,
            rate=RATE,
            input=True,
            frames_per_buffer=CHUNK
        )
        
        buffer = []
        buffer_duration = 0
        buffer_max = 10  # 10秒缓冲
        
        while self.is_running:
            data = stream.read(CHUNK)
            buffer.append(data)
            buffer_duration += CHUNK / RATE
            
            if buffer_duration >= buffer_max:
                audio_data = b"".join(buffer)
                self.audio_queue.put(audio_data)
                buffer = []
                buffer_duration = 0
        
        stream.stop_stream()
        stream.close()
        p.terminate()
    
    def _transcribe_loop(self, language):
        """转录循环"""
        while self.is_running:
            try:
                audio_data = self.audio_queue.get(timeout=1)
                
                # Whisper 转录
                result = self.model.transcribe(
                    audio_data,
                    language=language,
                    task="transcribe"
                )
                
                # 输出字幕
                for segment in result["segments"]:
                    print(f"字幕：{segment['text']}")
                    
            except queue.Empty:
                continue
    
    def stop(self):
        """停止"""
        self.is_running = False

from pyannote.audio import Pipeline
import whisper

class SpeakerDiarization:
    """说话人分离 + 字幕生成"""
    
    def __init__(self, whisper_model="large-v3"):
        self.whisper = whisper.load_model(whisper_model)
        self.diarization = Pipeline.from_pretrained(
            "pyannote/speaker-diarization-3.1"
        )
    
    def process(self, audio_path, num_speakers=None):
        """处理音频，生成带说话人标记的字幕"""
        
        # 1. Whisper 转录
        transcript = self.whisper.transcribe(audio_path)
        
        # 2. Pyannote 说话人分离
        diarization = self.diarization(
            audio_path,
            num_speakers=num_speakers
        )
        
        # 3. 对齐说话人和字幕
        subtitles = []
        for segment in transcript["segments"]:
            start = segment["start"]
            end = segment["end"]
            text = segment["text"]
            
            # 找到对应时间段的说话人
            speaker = self._find_speaker(diarization, start, end)
            
            subtitles.append({
                "start": start,
                "end": end,
                "text": text,
                "speaker": speaker
            })
        
        return subtitles
    
    def _find_speaker(self, diarization, start, end):
        """根据时间找说话人"""
        mid = (start + end) / 2
        for turn, _, speaker in diarization.itertracks(yield_label=True):
            if turn.start <= mid <= turn.end:
                return speaker
        return "Unknown"

from openai import OpenAI
import whisper

class MultilingualSubtitle:
    """多语言字幕生成"""
    
    def __init__(self):
        self.client = OpenAI()
        self.whisper = whisper.load_model("large-v3")
    
    async def generate(
        self, 
        audio_path, 
        source_lang="zh",
        target_langs=["en", "ja", "ko"]
    ):
        """生成多语言字幕"""
        
        # 1. 语音识别
        transcript = self.whisper.transcribe(
            audio_path,
            language=source_lang
        )
        
        # 2. 提取文本段落
        segments = transcript["segments"]
        
        # 3. 批量翻译
        translations = {}
        for target_lang in target_langs:
            translated_segments = await self._translate_segments(
                segments, source_lang, target_lang
            )
            translations[target_lang] = translated_segments
        
        # 4. 生成 SRT 文件
        subtitles = {}
        for lang, segs in translations.items():
            srt = self._generate_srt(segs)
            subtitles[lang] = srt
        
        return subtitles
    
    async def _translate_segments(
        self, 
        segments, 
        source_lang, 
        target_lang
    ):
        """翻译字幕段落"""
        translated = []
        
        # 批量翻译（提升效率）
        batch_size = 20
        for i in range(0, len(segments), batch_size):
            batch = segments[i:i+batch_size]
            texts = [s["text"] for s in batch]
            
            # GPT-4o 翻译
            response = await self.client.chat.completions.acreate(
                model="gpt-4o",
                messages=[{
                    "role": "user",
                    "content": f"""
                    将以下字幕翻译为{target_lang}。
                    保持简洁、口语化。
                    每行一条字幕，不要添加编号。
                    
                    原文字幕：
                    {chr(10).join(texts)}
                    """
                }]
            )
            
            translated_texts = response.choices[0].message.content.split("\n")
            
            for j, seg in enumerate(batch):
                if j < len(translated_texts):
                    translated.append({
                        "start": seg["start"],
                        "end": seg["end"],
                        "text": translated_texts[j]
                    })
        
        return translated
    
    def _generate_srt(self, segments):
        """生成 SRT 格式"""
        lines = []
        for i, seg in enumerate(segments, 1):
            start = self._format_time(seg["start"])
            end = self._format_time(seg["end"])
            lines.append(f"{i}")
            lines.append(f"{start} --> {end}")
            lines.append(seg["text"])
            lines.append("")
        return "\n".join(lines)
    
    def _format_time(self, seconds):
        """格式化时间"""
        hours = int(seconds // 3600)
        minutes = int((seconds % 3600) // 60)
        secs = int(seconds % 60)
        millis = int((seconds % 1) * 1000)
        return f"{hours:02d}:{minutes:02d}:{secs:02d},{millis:03d}"

class SubtitleOptimizer:
    """字幕优化器"""
    
    def optimize_breaks(self, segments, max_chars=42):
        """优化断句，确保每行字数合理"""
        optimized = []
        
        for seg in segments:
            text = seg["text"]
            
            # 如果字数超过限制，重新断句
            if len(text) > max_chars:
                # 使用 GPT 重新断句
                new_segments = self._smart_break(
                    seg["start"],
                    seg["end"],
                    text,
                    max_chars
                )
                optimized.extend(new_segments)
            else:
                optimized.append(seg)
        
        return optimized
    
    def _smart_break(self, start, end, text, max_chars):
        """智能断句"""
        # 计算每个字符的平均时长
        char_duration = (end - start) / len(text)
        
        # 找到合适的断句点
        breaks = []
        current_start = 0
        
        while current_start < len(text):
            # 寻找最近的标点或空格
            next_break = min(current_start + max_chars, len(text))
            
            # 向后查找标点
            for i in range(next_break, min(next_break + 10, len(text))):
                if text[i] in "，。！？,.!?":
                    next_break = i + 1
                    break
            
            segment_text = text[current_start:next_break].strip()
            if segment_text:
                segment_start = start + current_start * char_duration
                segment_end = start + next_break * char_duration
                
                breaks.append({
                    "start": segment_start,
                    "end": segment_end,
                    "text": segment_text
                })
            
            current_start = next_break
        
        return breaks

class TermCorrector:
    """专业术语纠正"""
    
    def __init__(self, domain="tech"):
        self.terminology = self._load_terminology(domain)
    
    def correct(self, text):
        """纠正专业术语"""
        # 使用 GPT-4o 进行上下文感知的纠正
        response = self.client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{
                "role": "user",
                "content": f"""
                纠正以下字幕中的专业术语错误。
                只输出纠正后的文本，不要解释。
                
                字幕：{text}
                """
            }]
        )
        
        return response.choices[0].message.content

class SubtitleConverter:
    """字幕格式转换"""
    
    def srt_to_vtt(self, srt_content):
        """SRT 转 VTT"""
        lines = ["WEBVTT", ""]
        
        for block in srt_content.strip().split("\n\n"):
            parts = block.split("\n")
            if len(parts) >= 3:
                # 时间轴转换（, → .）
                time_line = parts[1].replace(",", ".")
                text = "\n".join(parts[2:])
                
                lines.append(time_line)
                lines.append(text)
                lines.append("")
        
        return "\n".join(lines)
    
    def srt_to_ass(self, srt_content, style="Default"):
        """SRT 转 ASS"""
        header = """[Script Info]
Title: AI Generated Subtitle
ScriptType: v4.00+
PlayResX: 1920
PlayResY: 1080

[V4+ Styles]
Format: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour, Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle, BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding
Style: Default,Microsoft YaHei,60,&H00FFFFFF,&H000000FF,&H00000000,&H00000000,0,0,0,0,100,100,0,0,1,2,0,2,10,10,10,1

[Events]
Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text
"""
        
        events = []
        for block in srt_content.strip().split("\n\n"):
            parts = block.split("\n")
            if len(parts) >= 3:
                time_line = parts[1]
                start, end = time_line.split(" --> ")
                text = "\\N".join(parts[2:])
                
                # 转换时间格式
                start_ass = self._srt_to_ass_time(start)
                end_ass = self._srt_to_ass_time(end)
                
                events.append(
                    f"Dialogue: 0,{start_ass},{end_ass},{style},,0,0,0,,{text}"
                )
        
        return header + "\n".join(events)

# 降噪 + 音量归一化
def preprocess_audio(input_path, output_path):
    import librosa
    import soundfile as sf
    
    # 加载
    y, sr = librosa.load(input_path, sr=16000)
    
    # 降噪（使用 DeepFilterNet 或 noisereduce）
    y_denoised = librosa.effects.preemphasis(y)
    
    # 音量归一化
    y_normalized = librosa.util.normalize(y_denoised)
    
    # 保存
    sf.write(output_path, y_normalized, sr)

def validate_subtitle(segments):
    """验证字幕质量"""
    issues = []
    
    for i, seg in enumerate(segments):
        # 1. 时长检查
        duration = seg["end"] - seg["start"]
        if duration < 1:
            issues.append(f"段落 {i+1} 时长过短（{duration:.2f}s）")
        elif duration > 7:
            issues.append(f"段落 {i+1} 时长过长（{duration:.2f}s）")
        
        # 2. 字数检查
        char_count = len(seg["text"])
        if char_count > 50:
            issues.append(f"段落 {i+1} 字数过多（{char_count}字）")
        
        # 3. 空白检查
        if not seg["text"].strip():
            issues.append(f"段落 {i+1} 为空白")
    
    return issues