FSMN-VAD支持批量导出？文件打包下载功能实现教程

1. 引言

1.1 FSMN-VAD 离线语音端点检测控制台

基于 ModelScope 达摩院 FSMN-VAD 模型的离线语音检测服务，能够精准识别音频中的有效语音片段，并自动剔除静音部分。该工具支持上传本地音频文件或通过麦克风实时录音测试，检测结果以结构化表格形式展示，包含每个语音片段的开始时间、结束时间及总时长。适用于语音识别预处理、长音频自动切分和语音唤醒等典型场景。

当前版本已具备基础的语音检测能力，但缺乏对多文件处理结果的批量导出与打包下载功能。在实际工程应用中，用户往往需要对多个音频进行连续分析并集中获取所有检测报告。本文将在此前部署的基础上，扩展实现“批量导出”功能——允许用户一次性上传多个音频文件，系统逐个完成 VAD 分析后，自动生成独立的结果文本文件，并最终打包为.zip压缩包供一键下载。

1.2 教程目标与价值

本教程属于实践应用类文章，旨在解决 FSMN-VAD 控制台在真实使用场景下的效率瓶颈问题。我们将：

扩展 Gradio 接口以支持多文件上传
实现自动化批处理逻辑
构建结构化输出目录
集成 ZIP 打包与文件下载链路

最终达成：用户上传 N 个音频 → 后端依次执行 VAD 检测 → 生成 N 份.txt报告 → 自动压缩 → 提供可点击下载的压缩包。

2. 功能扩展设计与技术选型

2.1 需求拆解与流程规划

原始 FSMN-VAD 工具仅支持单文件交互式检测，无法满足批量处理需求。新功能需覆盖以下核心环节：

环节	目标
输入方式	支持一次上传多个音频文件（拖拽/选择）
处理逻辑	对每份音频独立调用`vad_pipeline`进行检测
输出格式	每个音频生成一个`.txt`文本报告，内容为 Markdown 表格
下载机制	将所有报告打包为`results.zip`，提供下载按钮

2.2 技术方案对比与选型

方案	优点	缺点	决策依据
使用`gr.File()`返回单个路径	实现简单	不支持多文件打包返回	❌ 不适用
返回`gr.Files()`组件列表	可显示多个文件	用户需手动逐个下载	⚠️ 体验差
返回`gr.File()`+ ZIP 包路径	单次点击下载全部	需额外构建压缩逻辑	✅ 最佳选择

最终决定采用ZIP 打包 +gr.File()下载组件的组合方案，兼顾用户体验与工程可行性。

2.3 核心依赖补充

除原有依赖外，新增标准库支持：

# 无需安装，Python 内置模块 import zipfile import tempfile

利用tempfile.TemporaryDirectory()创建临时工作区，确保每次运行隔离；使用zipfile模块构建压缩包。

3. 批量导出功能代码实现

3.1 修改 Web 服务脚本 (`web_app_batch.py`)

创建新文件web_app_batch.py，完整代码如下：

import os import gradio as gr from modelscope.pipelines import pipeline from modelscope.utils.constant import Tasks import soundfile as sf import zipfile import tempfile # 设置模型缓存路径 os.environ['MODELSCOPE_CACHE'] = './models' # 初始化 VAD 模型（全局加载） print("正在加载 FSMN-VAD 模型...") vad_pipeline = pipeline( task=Tasks.voice_activity_detection, model='iic/speech_fsmn_vad_zh-cn-16k-common-pytorch' ) print("模型加载完成！") def format_time(ms): """毫秒转秒，保留三位小数""" return round(ms / 1000.0, 3) def generate_report(segments): """根据检测结果生成 Markdown 表格字符串""" if not segments: return "未检测到有效语音段。\n" report = "| 片段序号 | 开始时间(s) | 结束时间(s) | 时长(s) |\n" report += "| :--- | :--- | :--- | :--- |\n" for i, seg in enumerate(segments): start_ms, end_ms = seg[0], seg[1] start_s, end_s = format_time(start_ms), format_time(end_ms) duration_s = round(end_s - start_s, 3) report += f"| {i+1} | {start_s} | {end_s} | {duration_s} |\n" return report def process_batch_audios(audio_files): """ 批量处理音频文件，生成 ZIP 压缩包 参数: audio_files - Gradio 传入的文件对象列表 返回: zip_file_path (用于 gr.File 下载) """ if not audio_files: return None, "请至少上传一个音频文件" # 创建临时目录存放报告 with tempfile.TemporaryDirectory() as tmpdir: reports_dir = os.path.join(tmpdir, "vad_reports") os.makedirs(reports_dir, exist_ok=True) success_count = 0 for file_obj in audio_files: try: # 获取原始文件名（不含路径） original_name = os.path.basename(file_obj.name) stem, _ = os.path.splitext(original_name) report_path = os.path.join(reports_dir, f"{stem}_vad.txt") # 加载音频数据（Gradio Audio 返回的是元组 (sr, data) 或路径） if hasattr(file_obj, 'name'): audio_path = file_obj.name else: return None, "音频格式异常" # 执行 VAD 检测 result = vad_pipeline(audio_path) if isinstance(result, list) and len(result) > 0: segments = result[0].get('value', []) else: segments = [] # 生成报告内容 content = f"音频文件: {original_name}\n" content += f"采样率: 16kHz\n" content += f"检测模型: iic/speech_fsmn_vad_zh-cn-16k-common-pytorch\n\n" content += "## 🎤 语音片段检测结果\n\n" content += generate_report(segments) # 写入 TXT 文件 with open(report_path, 'w', encoding='utf-8') as f: f.write(content) success_count += 1 except Exception as e: error_report = os.path.join(reports_dir, f"error_{success_count+1}.txt") with open(error_report, 'w', encoding='utf-8') as f: f.write(f"文件: {original_name}\n") f.write(f"错误: {str(e)}\n") # 打包所有报告 zip_path = os.path.join(tmpdir, "vad_results.zip") with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as zipf: for root, _, files in os.walk(reports_dir): for file in files: file_path = os.path.join(root, file) arcname = f"vad_reports/{file}" # 在 ZIP 中的相对路径 zipf.write(file_path, arcname) # 将 ZIP 复制到持久化路径以便 Gradio 访问 final_zip_path = "/tmp/vad_results.zip" os.makedirs("/tmp", exist_ok=True) if os.path.exists(final_zip_path): os.remove(final_zip_path) os.rename(zip_path, final_zip_path) return final_zip_path, f"✅ 成功处理 {success_count}/{len(audio_files)} 个文件，结果已打包。" # 构建 Gradio 界面 with gr.Blocks(title="FSMN-VAD 批量语音检测") as demo: gr.Markdown("# 📦 FSMN-VAD 批量语音端点检测与结果导出") gr.Markdown("支持多音频文件上传，系统将自动生成检测报告并打包为 ZIP 文件供下载。") with gr.Row(): with gr.Column(): audio_input = gr.File( label="上传多个音频文件", file_count="multiple", file_types=[".wav", ".mp3", ".flac"] ) run_btn = gr.Button("开始批量检测", variant="primary") with gr.Column(): status_text = gr.Textbox(label="处理状态") download_output = gr.File(label="下载结果压缩包", visible=False) # 绑定事件 run_btn.click( fn=process_batch_audios, inputs=audio_input, outputs=[download_output, status_text] ) # 动态控制下载组件可见性 def show_download(zipped_file, msg): return gr.File(visible=zipped_file is not None) download_output.change( fn=lambda x, y: gr.File(visible=x is not None), inputs=[download_output, status_text], outputs=download_output ) if __name__ == "__main__": demo.launch(server_name="127.0.0.1", server_port=6006)

3.2 关键代码解析

（1）多文件输入配置

audio_input = gr.File(file_count="multiple", ...)

启用file_count="multiple"允许用户同时选择多个文件。

（2）临时目录管理

with tempfile.TemporaryDirectory() as tmpdir:

保证每次运行环境隔离，避免文件冲突，程序退出后自动清理。

（3）ZIP 打包逻辑

with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as zipf: zipf.write(file_path, arcname)

使用压缩模式写入，减小包体积，arcname控制压缩包内路径结构。

（4）Gradio 文件返回机制

return final_zip_path, "✅ 成功..."

gr.File组件接收本地文件路径即可生成可下载链接。

4. 服务启动与功能验证

4.1 启动命令

python web_app_batch.py

确保此前已安装依赖并设置好模型镜像源：

export MODELSCOPE_CACHE='./models' export MODELSCOPE_ENDPOINT='https://mirrors.aliyun.com/modelscope/'

4.2 测试步骤

访问界面：通过 SSH 隧道映射后，在浏览器打开http://127.0.0.1:6006
上传文件：点击“上传多个音频文件”，选择 2~3 个.wav或.mp3文件
触发检测：点击“开始批量检测”
查看反馈：
文本框显示处理进度（如“✅ 成功处理 3/3 个文件”）
下方出现“Download”按钮，提示“vad_results.zip”
下载验证：
点击下载 ZIP 包
解压后检查每个_vad.txt文件是否包含正确的时间戳表格

4.3 预期输出示例

文件名：meeting_clip_vad.txt

音频文件: meeting_clip.wav 采样率: 16kHz 检测模型: iic/speech_fsmn_vad_zh-cn-16k-common-pytorch ## 🎤 语音片段检测结果 | 片段序号 | 开始时间(s) | 结束时间(s) | 时长(s) | | :--- | :--- | :--- | :--- | | 1 | 1.234 | 4.567 | 3.333 | | 2 | 6.789 | 9.012 | 2.223 |