""" 会议音频处理器 - 基于 FunASR 支持: 音频上传转录、实时录音转录、说话人分离 """ import os import uuid import hashlib import asyncio import tempfile from datetime import datetime from typing import List, Dict, Optional, Tuple, AsyncGenerator from dataclasses import dataclass, asdict from concurrent.futures import ThreadPoolExecutor import numpy as np import soundfile as sf from funasr import AutoModel from funasr.utils.postprocess_utils import rich_transcription_postprocess @dataclass class AudioSegment: """音频片段""" start: float # 秒 end: float # 秒 speaker: int # 说话人ID text: str # 转录文本 emotion: str = "NEUTRAL" # 情感标签 @dataclass class MeetingRecord: """会议记录""" meeting_id: str title: str date: str duration: float audio_path: str transcript_path: Optional[str] segments: List[AudioSegment] speaker_count: int raw_result: Dict def to_dict(self) -> Dict: return { "meeting_id": self.meeting_id, "title": self.title, "date": self.date, "duration": self.duration, "audio_path": self.audio_path, "transcript_path": self.transcript_path, "speaker_count": self.speaker_count, "segments": [ { "start": s.start, "end": s.end, "speaker": s.speaker, "text": s.text, "emotion": s.emotion } for s in self.segments ] } class RealtimeTranscriber: """实时转写器""" def __init__(self, processor): self.processor = processor self.model = processor.model self._reset() def _reset(self): """重置状态""" self.segments = [] self.speakers = set() self.total_duration = 0.0 def process_audio_data(self, audio_data: bytes, sample_rate: int = 16000) -> Optional[AudioSegment]: """处理一个音频块""" try: # 将 bytes 转换为 numpy 数组 import wave import io # 解析 WAV 数据 with wave.open(io.BytesIO(audio_data), 'rb') as wav_file: channels = wav_file.getnchannels() sample_width = wav_file.getsampwidth() framerate = wav_file.getframerate() n_frames = wav_file.getnframes() audio_bytes = wav_file.readframes(n_frames) # 转换为 numpy 数组 audio_array = np.frombuffer(audio_bytes, dtype=np.int16) # 如果是多声道，转换为单声道 if channels > 1: audio_array = audio_array.reshape(-1, channels).mean(axis=1).astype(np.int16) # 重采样为 16kHz（如果需要） if framerate != 16000: import scipy.signal as signal num_samples = int(len(audio_array) * 16000 / framerate) audio_array = signal.resample(audio_array, num_samples).astype(np.int16) # 更新总时长 self.total_duration += len(audio_array) / sample_rate # 保存为临时文件 temp_file = tempfile.NamedTemporaryFile(suffix='.wav', delete=False) sf.write(temp_file.name, audio_array, sample_rate) temp_path = temp_file.name # 使用模型转录 result = self.model.generate( input=temp_path, batch_size_s=60, return_raw_text=True, sentence_timestamp=True ) # 清理临时文件 os.unlink(temp_path) if result and len(result) > 0: segment = self._parse_single_result(result[0]) if segment: self.speakers.add(segment.speaker) self.segments.append(segment) return segment except Exception as e: print(f"处理音频块失败: {e}") return None def _parse_single_result(self, result) -> Optional[AudioSegment]: """解析单个转录结果""" try: text = result.get("text", "") if not text: return None text = rich_transcription_postprocess(text) # 获取时间戳 sentence_info = result.get("sentence_info", []) if sentence_info: seg = sentence_info[0] start = seg.get("start", 0) / 1000 end = seg.get("end", 0) / 1000 speaker = seg.get("spk", 0) else: start = self.total_duration - len(text) / 10 # 估算 end = self.total_duration speaker = 0 return AudioSegment( start=start, end=end, speaker=speaker, text=text, emotion="NEUTRAL" ) except Exception as e: print(f"解析结果失败: {e}") return None def get_result(self) -> Dict: """获取当前转写结果""" return { "segments": [ { "start": s.start, "end": s.end, "speaker": s.speaker, "text": s.text, "emotion": s.emotion } for s in self.segments ], "speaker_count": len(self.speakers), "duration": self.total_duration } class MeetingProcessor: """会议处理器""" def __init__(self, device: str = "cpu"): """ 初始化处理器 Args: device: "cpu" 或 "cuda" """ self.device = device self.model = None self.model_name = "iic/SenseVoiceSmall" self.realtime_transcriber: Optional[RealtimeTranscriber] = None self._init_model() def _init_model(self): """初始化 FunASR 模型""" print(f"🔄 初始化模型: {self.model_name} (device={self.device})") self.model = AutoModel( model=self.model_name, vad_model="fsmn-vad", spk_model="cam++", device=self.device, disable_update=True ) print("✅ 模型加载完成") def create_realtime_transcriber(self) -> RealtimeTranscriber: """创建实时转写器实例""" self.realtime_transcriber = RealtimeTranscriber(self) return self.realtime_transcriber def _download_audio(self, url: str) -> Tuple[str, float]: """下载远程音频""" import urllib.request suffix = ".wav" if ".mp3" in url: suffix = ".mp3" elif ".m4a" in url: suffix = ".m4a" temp_file = tempfile.NamedTemporaryFile(suffix=suffix, delete=False) urllib.request.urlretrieve(url, temp_file.name) info = sf.info(temp_file.name) duration = info.duration return temp_file.name, duration def process_audio( self, audio_path: str, title: Optional[str] = None, is_url: bool = False ) -> MeetingRecord: """ 处理音频文件 Args: audio_path: 音频路径或URL title: 会议标题 is_url: 是否为URL Returns: MeetingRecord: 会议记录 """ # 生成会议ID meeting_id = datetime.now().strftime("%Y%m%d_%H%M%S") + "_" + uuid.uuid4().hex[:6] # 处理音频路径 if is_url: actual_path, duration = self._download_audio(audio_path) else: actual_path = audio_path info = sf.info(actual_path) duration = info.duration # 使用标题或默认标题 if not title: title = f"会议记录_{datetime.now().strftime('%Y-%m-%d %H:%M')}" print(f"🎙️ 开始处理: {title}") print(f"📁 音频: {actual_path}") # 执行转录 result = self.model.generate( input=actual_path, batch_size_s=300, return_raw_text=True, is_final=True, sentence_timestamp=True ) # 解析结果 segments = self._parse_result(result) # 统计说话人数量 speakers = set(s.speaker for s in segments) speaker_count = len(speakers) # 创建记录 record = MeetingRecord( meeting_id=meeting_id, title=title, date=datetime.now().isoformat(), duration=duration, audio_path=actual_path, transcript_path=None, segments=segments, speaker_count=speaker_count, raw_result=result[0] if result else {} ) print(f"✅ 处理完成: {len(segments)} 个片段, {speaker_count} 位说话人") # 清理临时文件 if is_url and os.path.exists(actual_path): os.unlink(actual_path) return record def _parse_result(self, result: List) -> List[AudioSegment]: """解析 FunASR 返回结果""" segments = [] if not result or len(result) == 0: return segments raw = result[0] # 尝试获取 sentence_info sentence_info = raw.get("sentence_info", []) if not sentence_info and isinstance(raw, dict): # 兼容不同格式 text = raw.get("text", "") if text: segments.append(AudioSegment( start=0, end=0, speaker=0, text=rich_transcription_postprocess(text) )) return segments for seg in sentence_info: start = seg.get("start", 0) / 1000 # 毫秒转秒 end = seg.get("end", 0) / 1000 speaker = seg.get("spk", 0) text = seg.get("sentence", "") # 后处理：标点、格式 text = rich_transcription_postprocess(text) # 提取情感标签 emotion = self._extract_emotion(text) segments.append(AudioSegment( start=start, end=end, speaker=speaker, text=text, emotion=emotion )) return segments def _extract_emotion(self, text: str) -> str: """从文本中提取情感标签""" emotions = ["NEUTRAL", "happy", "sad", "angry", "surprise"] for emo in emotions: if emo in text: return emo return "NEUTRAL" def transcribe_segment(self, audio_path: str, start: float, end: float) -> str: """转录指定时间段的音频""" data, sr = sf.read(audio_path, start=int(start * sr), stop=int(end * sr)) # 临时保存片段 temp_file = tempfile.NamedTemporaryFile(suffix=".wav", delete=False) sf.write(temp_file.name, data, sr) # 转录 result = self.model.generate(input=temp_file.name) # 清理 os.unlink(temp_file.name) if result: return rich_transcription_postprocess(result[0].get("text", "")) return "" def get_device() -> str: """自动检测可用设备""" try: import torch if torch.cuda.is_available(): return "cuda" except ImportError: pass return "cpu" if __name__ == "__main__": # 测试 processor = MeetingProcessor(device=get_device()) # 测试示例音频 test_url = "https://isv-data.oss-cn-hangzhou.aliyuncs.com/ics/MaaS/ASR/test_audio/asr_example_zh.wav" result = processor.process_audio(test_url, title="测试会议", is_url=True) print("\n=== 转录结果 ===") for seg in result.segments: print(f"[{seg.start:.1f}s] 说话人{seg.speaker}: {seg.text}")