如何用Python基于OpenAI Whisper转录超25MB大视频？解决分片问题

针对你遇到的三个问题，以下是具体原因分析和修复后的完整实现：

问题1：分片转录时间轴不对应 #

原因 #

每个分片是从0开始截取的，Whisper返回的时间戳是分片内部的相对时间，未关联原视频的绝对时间轴。

修复 #

在拆分视频时记录每个分片的实际起始时间，转录完成后给每个segment的start和end字段加上该分片的时间偏移量。

问题2：静音时段无法生成分片 #

原因 #

原代码仅提取静音起始点，拆分时直接将分片结束时间设为静音起始点，可能生成极短分片甚至空文件；同时分片数量基于预估时长计算，未适配静音调整后的实际分片情况。

修复 #

• 完善静音检测逻辑，提取完整的静音时间段（起始+结束）
• 优先在静音区间的结束位置截断分片，避免生成无效短分片
• 动态调整分片数量，确保所有内容都被覆盖

问题3：相邻分片首尾语句重叠 #

原因 #

为避免语句截断设置的重叠时间，在合并转录时未做去重处理，导致首尾重复。

修复 #

• 记录每个分片的时间范围，合并时过滤掉重叠时间段内的重复segment
• 对比相邻分片的首尾文本，自动去除重复内容

修复后的完整代码 #

import openai
import math
import os
import subprocess
from difflib import SequenceMatcher
import json

openai.api_key = "sk-oijfowiejfo"
filename = 'test.mp4'

# Constants
max_bytes = 26214400  # Whisper 最大文件限制（25MB）
overlap_seconds = 5
silence_threshold = -40  # 静音阈值（dB），可根据视频调整
min_chunk_duration = 10  # 最小分片时长，避免生成极短分片

output_folder = "chunks"
os.makedirs(output_folder, exist_ok=True)
file_extension = os.path.splitext(filename)[1]

# 获取视频的比特率和总时长
bit_rate = float(subprocess.check_output(
    ["ffprobe", "-v", "quiet", "-show_entries", "format=bit_rate", "-of",
     "default=noprint_wrappers=1:nokey=1", filename]).strip())
audio_duration_s = float(subprocess.check_output(
    ["ffprobe", "-v", "quiet", "-show_entries", "format=duration", "-of",
     "default=noprint_wrappers=1:nokey=1", filename]).strip())

# 预估每个分片的基础时长
chunk_duration_s = (max_bytes * 8.0) / bit_rate * 0.9

# 完善静音检测：返回完整的静音时间段（start, end）
def detect_silence(file):
    cmd = ["ffmpeg", "-i", file, "-af", f"silencedetect=noise={silence_threshold}dB:d=0.5", "-f", "null", "-"]
    result = subprocess.run(cmd, capture_output=True, text=True)
    output = result.stderr
    silence_segments = []
    current_start = None
    for line in output.split('\n'):
        if "silence_start" in line:
            current_start = float(line.split(':')[1].strip())
        elif "silence_end" in line and current_start is not None:
            current_end = float(line.split(':')[1].strip())
            silence_segments.append((current_start, current_end))
            current_start = None
    return silence_segments

# 拆分视频，记录每个分片的元数据（起始时间、结束时间、文件名）
def split_with_silence():
    silence_segments = detect_silence(filename)
    chunk_metadata = []
    start_time = 0
    chunk_index = 1

    while start_time < audio_duration_s:
        end_time = min(start_time + chunk_duration_s, audio_duration_s)
        
        # 寻找最近的静音结束点，避免截断语句
        for (sil_start, sil_end) in silence_segments:
            # 只考虑在当前分片范围内的静音区间，且结束点距离分片结束不超过10秒
            if start_time < sil_end < end_time and (end_time - sil_end) < 10:
                end_time = sil_end
                break
        
        # 确保分片时长不小于最小值
        if end_time - start_time < min_chunk_duration:
            end_time = min(start_time + min_chunk_duration, audio_duration_s)
        
        chunk_file = os.path.join(output_folder, f"chunk_{chunk_index}{file_extension}")
        subprocess.run([
            "ffmpeg", "-i", filename, "-ss", str(start_time), "-to", str(end_time),
            "-c:v", "copy", "-c:a", "copy", "-y", chunk_file
        ], check=True)
        
        chunk_metadata.append({
            "file": chunk_file,
            "start": start_time,
            "end": end_time
        })
        
        # 更新下一个分片的起始时间（减去重叠时长）
        start_time = end_time - overlap_seconds
        chunk_index += 1
    
    return chunk_metadata

# 转录分片并修正时间轴
def transcribe_chunks(chunk_metadata):
    transcriptions = []
    for meta in chunk_metadata:
        with open(meta["file"], "rb") as file:
            transcription = openai.Audio.transcribe("whisper-1", file, response_format="verbose_json")
            
            # 修正每个segment的时间戳（加上分片的起始偏移）
            time_offset = meta["start"]
            for segment in transcription["segments"]:
                segment["start"] += time_offset
                segment["end"] += time_offset
            
            transcriptions.append({
                "metadata": meta,
                "transcription": transcription
            })
    return transcriptions

# 合并转录结果，去除重叠部分
def merge_transcriptions(transcriptions):
    combined_segments = []
    last_end_time = 0

    for entry in transcriptions:
        segments = entry["transcription"]["segments"]
        # 过滤掉与上一个分片重叠的部分
        filtered_segments = [seg for seg in segments if seg["start"] >= last_end_time - overlap_seconds * 0.5]
        
        # 进一步去除文本重复（如果有的话）
        if combined_segments and filtered_segments:
            last_text = combined_segments[-1]["text"].strip()
            first_text = filtered_segments[0]["text"].strip()
            similarity = SequenceMatcher(None, last_text, first_text).ratio()
            if similarity > 0.7:
                # 保留较长的文本，或者截断重复部分
                if len(last_text) > len(first_text):
                    filtered_segments.pop(0)
                else:
                    combined_segments.pop()
        
        combined_segments.extend(filtered_segments)
        if combined_segments:
            last_end_time = combined_segments[-1]["end"]
    
    # 生成合并后的文本和结构化数据
    combined_text = " ".join([seg["text"].strip() for seg in combined_segments])
    return {
        "segments": combined_segments,
        "full_text": combined_text
    }

# 主流程
if __name__ == "__main__":
    # 拆分视频
    chunk_metadata = split_with_silence()
    
    # 转录分片
    transcribed_chunks = transcribe_chunks(chunk_metadata)
    
    # 合并结果
    final_result = merge_transcriptions(transcribed_chunks)
    
    # 保存结构化结果
    with open("final_transcription.json", "w", encoding="utf-8") as f:
        json.dump(final_result, f, indent=2, ensure_ascii=False)
    
    # 保存纯文本结果
    with open("final_transcription.txt", "w", encoding="utf-8") as f:
        f.write(final_result["full_text"])
    
    print("转录完成，结果已保存到 final_transcription.json 和 final_transcription.txt")

关键修改说明 #

1. 时间轴修正：通过chunk_metadata记录每个分片的绝对起始时间，转录后给每个segment的时间戳加上偏移量，确保与原视频时间轴对齐。
2. 静音分片优化：完善静音检测逻辑，提取完整静音区间，优先在静音结束点截断分片，同时设置最小分片时长，避免生成无效文件。
3. 重叠去重：合并时先通过时间戳过滤重叠片段，再通过文本相似度对比去除重复内容，确保最终转录无冗余。

内容的提问来源于stack exchange，提问作者CloudExplorer