freedakgmail
b43e3725ee
- 配置浅色主题配色方案和 CSS 变量 - 修复组件在浅色模式下的样式适配 - 统一文字颜色类名使用 CSS 变量 - 优化玻璃效果在浅色主题下的显示
406 lines
12 KiB
Python
406 lines
12 KiB
Python
"""
|
|
会议音频处理器 - 基于 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}") |