大模型应用：语音转文本（ASR）实践：OpenAI Whisper精准转录解析.21

# 第一步：导入Whisper库（基础操作，所有功能都需要）
import whisper

# 第二步：定义基础转录函数
def transcribe_audio_basic(audio_path):
    # 2.1 加载模型：选择base尺寸
    # load_model是Whisper的核心函数，参数传模型尺寸
    model = whisper.load_model("base")
    
    # 2.2 执行转录：传入音频路径，得到结果
    # transcribe是实现“音频→文本”的核心方法
    result = model.transcribe(audio_path)
    
    # 2.3 返回关键结果（新手先关注text和language）
    return {
        "text": result["text"],       # 完整转录文本
        "language": result["language"], # 检测到的语言
        "segments": result["segments"]  # 段落时间戳
    }

# 第三步：调用函数并打印结果
if __name__ == "__main__":
    # 替换为你的音频文件路径
    audio_path = "qqqq.wav"
    # 调用转录函数
    result = transcribe_audio_basic(audio_path)
    # 打印结果（基础输出，看是否转写成功）
    print(f"转录文本: {result['text']}")
    print(f"检测语言: {result['language']}")

# 优化中文转写准确性的基础配置
result = model.transcribe(
    audio_path,
    language="zh",       # 强制指定中文，避免误判
    temperature=0.0,    # 固定结果，无随机错误
    beam_size=8,        # 增加候选数量，提升准确性
    best_of=6           # 配合beam_size，选最优结果
)

# 在上方代码的函数中添加以下代码，打印段落时间戳
for segment in result["segments"]:
    print(f"[{segment['start']:.1f}秒 - {segment['end']:.1f}秒]：{segment['text']}")

#输出结果
#[0.0秒 - 3.5秒]： 欢迎使用DDS文本朗读器。
#[3.5秒 - 6.8秒]： 请在这里输入想要朗读的文本。

def transcribe_basic_scene(audio_path):
    model = whisper.load_model("base")  # CPU首选
    result = model.transcribe(
        audio_path,
        language="zh",
        temperature=0.0,
        beam_size=8,
        best_of=6
    )
    return result

def transcribe_noise_scene(audio_path):
    model = whisper.load_model("small")  # 比base更准
    result = model.transcribe(
        audio_path,
        language="zh",
        temperature=0.0,
        beam_size=10,
        best_of=8
    )
    return result

def transcribe_long_audio(audio_path):
    model = whisper.load_model("medium")  # 有GPU优先选
    result = model.transcribe(
        audio_path,
        language="zh",
        temperature=0.0,
        beam_size=8,
        best_of=6,
        condition_on_previous_text=True  # 保持上下文连贯（新手无需改，默认True）
    )
    return result

def transcribe_with_word_timestamps(audio_path):
    model = whisper.load_model("base")
    result = model.transcribe(
        audio_path,
        language="zh",
        temperature=0.0,
        word_timestamps=True  # 启用词级时间戳
    )
    # 打印词级时间戳（新手可理解为“每个字/词的起止时间”）
    for segment in result["segments"]:
        print(f"[{segment['start']:.1f}秒 - {segment['end']:.1f}秒]：{segment['text']}")
        for word in segment["words"]:
            print(f"  「{word['word']}」：{word['start']:.1f}秒 - {word['end']:.1f}秒")
    return result

[0.0秒 - 3.5秒]： 欢迎使用DDS文本朗读器。
  「欢迎」：0.0秒 - 0.5秒
  「使用」：0.5秒 - 1.0秒
  「DDS」：1.0秒 - 1.5秒
  「文本」：1.5秒 - 2.0秒
  「朗读器」：2.0秒 - 3.5秒

def batch_transcribe(audio_files):
    # audio_files是音频文件路径列表，如["audio1.wav", "audio2.wav"]
    model = whisper.load_model("base")
    results = {}
    for file in audio_files:
        try:
            print(f"正在处理：{file}")
            result = model.transcribe(file, language="zh", temperature=0.0)
            results[file] = result["text"]  # 仅保存转写文本
        except Exception as e:
            print(f"处理{file}失败：{str(e)}")
            results[file] = "转写失败"
    return results

audio_list = ["meeting1.wav", "meeting2.wav"]
batch_results = batch_transcribe(audio_list)
# 打印批量结果
for file, text in batch_results.items():
    print(f"\n{file}转写结果：")
    print(text)

def transcribe_with_speaker(audio_path, hf_token):
    # 第一步：基础转写（获取文本+时间戳）
    model = whisper.load_model("base")
    transcribe_result = model.transcribe(
        audio_path,
        language="zh",
        temperature=0.0,
        word_timestamps=True
    )
    
    # 第二步：加载说话人分离模型
    from pyannote.audio import Pipeline
    diarization_pipeline = Pipeline.from_pretrained(
        "pyannote/speaker-diarization-3.1",
        use_auth_token=hf_token  # 替换为你的HuggingFace令牌
    )
    diarization_result = diarization_pipeline(audio_path)
    
    # 第三步：匹配“说话人-文本-时间戳”
    final_result = []
    for segment in transcribe_result["segments"]:
        # 找到该时间段内的主要说话人
        speaker = "unknown"
        for turn, _, spk in diarization_result.itertracks(yield_label=True):
            if turn.start <= segment["end"] and turn.end >= segment["start"]:
                speaker = spk
                break
        # 保存结果
        final_result.append({
            "start": segment["start"],
            "end": segment["end"],
            "speaker": speaker,
            "text": segment["text"]
        })
    
    # 打印结果
    for item in final_result:
        print(f"[{item['start']:.1f}秒 - {item['end']:.1f}秒] {item['speaker']}：{item['text']}")
    return final_result
hf_token = "你的HuggingFace令牌"
transcribe_with_speaker("group_discussion.wav", hf_token)

[0.0秒 - 5.0秒] SPEAKER_00： 今天我们讨论一下项目进度。
[5.0秒 - 10.0秒] SPEAKER_01： 我这边的模块已经完成了80%。

import noisereduce as nr
import soundfile as sf
# 读取音频
data, rate = sf.read(audio_path)
# 提取噪声样本（取音频前1秒作为噪声）
noise_sample = data[:rate]
# 降噪
reduced_noise = nr.reduce_noise(y=data, y_noise=noise_sample, sr=rate)
# 保存降噪后的音频
sf.write("audio_denoised.wav", reduced_noise, rate)
# 对降噪后的音频转写
model.transcribe("audio_denoised.wav", language="zh")

result_text = result["text"]
# 替换常见错字
corrected_text = result_text.replace("北惊", "北京").replace("百份之", "百分之")
print("校正后文本：", corrected_text)

import whisper
import numpy as np
import torch
from typing import Dict, List
import json

class AdvancedTranscriber:
    """
    高级语音转录器，包含多种优化功能
    """
    
    def __init__(self, model_size="base", device="cuda"):
        """
        初始化转录器
        
        Args:
            model_size: 模型尺寸 (tiny, base, small, medium, large)
            device: 计算设备 (cuda/cpu)
        """
        self.device = device if torch.cuda.is_available() else "cpu"
        self.model = whisper.load_model(model_size).to(self.device)
        
        # 配置转录参数
        self.transcribe_options = {
            "language": "zh",  # 指定语言（可选自动检测）
            "task": "transcribe",  # transcribe或translate
            "temperature": 0.0,  # 采样温度（0为确定性）
            "best_of": 5,  # 集束搜索候选数
            "beam_size": 5,  # 集束搜索宽度
            "fp16": True if self.device == "cuda" else False,
        }
    
    def transcribe_with_timestamps(self, audio_path: str) -> Dict:
        """
        带时间戳的详细转录
        """
        result = self.model.transcribe(
            audio_path,
            **self.transcribe_options,
            word_timestamps=True  # 启用词级时间戳
        )
        
        # 结构化输出
        structured_output = {
            "full_text": result["text"],
            "language": result["language"],
            "duration": self._get_audio_duration(audio_path),
            "segments": []
        }
        
        for segment in result["segments"]:
            segment_info = {
                "start": segment["start"],
                "end": segment["end"],
                "text": segment["text"],
                "confidence": segment.get("confidence", 0),
                "words": segment.get("words", [])
            }
            structured_output["segments"].append(segment_info)
        
        return structured_output
    
    def batch_transcribe(self, audio_files: List[str]) -> Dict[str, Dict]:
        """
        批量转录多个音频文件
        """
        results = {}
        
        for audio_file in audio_files:
            try:
                print(f"正在处理: {audio_file}")
                result = self.transcribe_with_timestamps(audio_file)
                results[audio_file] = result
            except Exception as e:
                print(f"处理失败 {audio_file}: {str(e)}")
                results[audio_file] = {"error": str(e)}
        
        return results
    
    def transcribe_with_speaker_diarization(self, audio_path: str):
        """
        结合说话人分离的转录
        需要额外安装：pip install pyannote.audio
        """
        from pyannote.audio import Pipeline
        
        # 第一步：使用Whisper转录
        transcription_result = self.transcribe_with_timestamps(audio_path)
        
        # 第二步：说话人分离
        diarization_pipeline = Pipeline.from_pretrained(
            "pyannote/speaker-diarization-3.1",
            use_auth_token="YOUR_HUGGINGFACE_TOKEN"
        )
        
        diarization_result = diarization_pipeline(audio_path)
        
        # 第三步：对齐说话人标签和转录
        aligned_result = self._align_speakers_with_transcription(
            transcription_result, 
            diarization_result
        )
        
        return aligned_result
    
    def _align_speakers_with_transcription(self, transcription, diarization):
        """
        对齐说话人标签和转录文本
        """
        aligned_segments = []
        
        for transcript_segment in transcription["segments"]:
            # 找到该时间段内的说话人
            segment_start = transcript_segment["start"]
            segment_end = transcript_segment["end"]
            
            # 收集该时间段的所有说话人
            speakers_in_segment = []
            for turn, _, speaker in diarization.itertracks(yield_label=True):
                if turn.start <= segment_end and turn.end >= segment_start:
                    overlap_start = max(turn.start, segment_start)
                    overlap_end = min(turn.end, segment_end)
                    overlap_duration = overlap_end - overlap_start
                    
                    speakers_in_segment.append({
                        "speaker": speaker,
                        "overlap_duration": overlap_duration
                    })
            
            # 选择重叠时间最长的说话人
            if speakers_in_segment:
                main_speaker = max(
                    speakers_in_segment, 
                    key=lambda x: x["overlap_duration"]
                )["speaker"]
                transcript_segment["speaker"] = main_speaker
            else:
                transcript_segment["speaker"] = "unknown"
            
            aligned_segments.append(transcript_segment)
        
        transcription["segments"] = aligned_segments
        return transcription
    
    def _get_audio_duration(self, audio_path: str) -> float:
        """获取音频时长"""
        import soundfile as sf
        data, sample_rate = sf.read(audio_path)
        return len(data) / sample_rate

# 使用示例
transcriber = AdvancedTranscriber(model_size="small")

# 单文件转录
result = transcriber.transcribe_with_timestamps("qqqq.wav")
print(json.dumps(result, ensure_ascii=False, indent=2))

# 批量转录
batch_results = transcriber.batch_transcribe([
    "meeting1.wav",
    "meeting2.wav",
    "presentation.wav"
])

# 带说话人分离的转录
diarization_result = transcriber.transcribe_with_speaker_diarization(
    "group_discussion.wav"
)

import re
from typing import List, Dict

class TranscriptionPostProcessor:
    """
    转录后处理：标点恢复、数字规范化等
    """
    
    def __init__(self, language="zh"):
        self.language = language
        
        # 定义正则规则
        self.patterns = {
            "numbers": {
                "zh": r"(\d+)",
                "en": r"(\d+)"
            },
            "urls": r"https?://\S+|www\.\S+",
            "emails": r"\S+@\S+\.\S+"
        }
        
        # 数字转换规则（中文）
        self.number_map_zh = {
            '0': '零', '1': '一', '2': '二', '3': '三', '4': '四',
            '5': '五', '6': '六', '7': '七', '8': '八', '9': '九'
        }
    
    def process_transcription(self, text: str) -> str:
        """
        完整的后处理流程
        """
        # 1. 清理多余空格
        text = self._clean_spaces(text)
        
        # 2. 标点标准化
        text = self._normalize_punctuation(text)
        
        # 3. 数字规范化
        if self.language == "zh":
            text = self._normalize_numbers_chinese(text)
        elif self.language == "en":
            text = self._normalize_numbers_english(text)
        
        # 4. 大写规范化（英文）
        if self.language == "en":
            text = self._normalize_capitalization(text)
        
        # 5. 分割过长的句子
        text = self._split_long_sentences(text)
        
        return text
    
    def _clean_spaces(self, text: str) -> str:
        """清理多余空格"""
        # 去除首尾空格
        text = text.strip()
        # 将多个空格合并为一个
        text = re.sub(r'\s+', ' ', text)
        # 修复标点前的空格
        text = re.sub(r'\s+([,.!?;:])', r'\1', text)
        
        return text
    
    def _normalize_punctuation(self, text: str) -> str:
        """标点符号标准化"""
        # 英文标点转中文标点（如果目标语言是中文）
        if self.language == "zh":
            punctuation_map = {
                ',': '，', '.': '。', '!': '！', '?': '？',
                ';': '；', ':': '：', '(': '（', ')': '）',
                '"': '「', "'": '『'
            }
            
            for eng_punc, zh_punc in punctuation_map.items():
                text = text.replace(eng_punc, zh_punc)
        
        # 确保标点后有空格（英文）
        if self.language == "en":
            text = re.sub(r'([,.!?;:])([A-Za-z])', r'\1 \2', text)
        
        return text
    
    def _normalize_numbers_chinese(self, text: str) -> str:
        """中文数字规范化"""
        def number_to_chinese(match):
            num_str = match.group(1)
            
            # 简单转换：将每个数字转为中文
            if len(num_str) == 1:
                return self.number_map_zh.get(num_str, num_str)
            elif len(num_str) == 4 and 1000 <= int(num_str) <= 9999:
                # 年份处理：2023 -> 二零二三
                return ''.join(self.number_map_zh.get(d, d) for d in num_str)
            else:
                # 保持数字形式
                return num_str
        
        # 替换数字
        text = re.sub(self.patterns["numbers"][self.language], number_to_chinese, text)
        
        return text
    
    def _normalize_numbers_english(self, text: str) -> str:
        """英文数字规范化"""
        def number_to_words(match):
            num = int(match.group(1))
            
            # 简单的数字转单词（0-99）
            if 0 <= num <= 99:
                words_0_19 = [
                    'zero', 'one', 'two', 'three', 'four', 'five',
                    'six', 'seven', 'eight', 'nine', 'ten',
                    'eleven', 'twelve', 'thirteen', 'fourteen', 'fifteen',
                    'sixteen', 'seventeen', 'eighteen', 'nineteen'
                ]
                tens = [
                    '', '', 'twenty', 'thirty', 'forty',
                    'fifty', 'sixty', 'seventy', 'eighty', 'ninety'
                ]
                
                if num < 20:
                    return words_0_19[num]
                else:
                    ten = tens[num // 10]
                    one = words_0_19[num % 10] if num % 10 != 0 else ''
                    return f"{ten}-{one}" if one else ten
            else:
                # 保持数字形式
                return str(num)
        
        # 替换数字
        text = re.sub(self.patterns["numbers"][self.language], number_to_words, text)
        
        return text
    
    def _split_long_sentences(self, text: str, max_length=50) -> str:
        """分割过长的句子"""
        sentences = re.split(r'([。！？.!?])', text)
        processed_sentences = []
        
        for i in range(0, len(sentences), 2):
            if i + 1 < len(sentences):
                sentence = sentences[i] + sentences[i + 1]
            else:
                sentence = sentences[i]
            
            if len(sentence) > max_length:
                # 根据逗号分割
                parts = re.split(r'([，,])', sentence)
                reconstructed = []
                
                for j in range(0, len(parts), 2):
                    if j + 1 < len(parts):
                        part = parts[j] + parts[j + 1]
                    else:
                        part = parts[j]
                    
                    reconstructed.append(part)
                
                sentence = ''.join(reconstructed)
            
            processed_sentences.append(sentence)
        
        return ''.join(processed_sentences)

# 使用后处理
post_processor = TranscriptionPostProcessor(language="zh")
transcribed_text = "今天天气很好 我们去公园玩吧123 "
processed_text = post_processor.process_transcription(transcribed_text)
print(f"处理后: {processed_text}")
# 输出: "今天天气很好，我们去公园玩吧一二三"