如何批量生成语音文件？API调用自动化脚本分享-育师

如何批量生成语音文件？API调用自动化脚本分享

📌 业务场景描述：从单次合成到批量处理的工程需求

在智能客服、有声书生成、语音播报系统等实际应用中，我们常常面临一个核心问题：如何高效地将大量文本内容转化为高质量语音文件？虽然许多语音合成模型（如 ModelScope 的 Sambert-Hifigan）提供了直观的 WebUI 界面供用户手动输入文本并生成音频，但这种方式在面对成百上千条文本时显得效率极低。

本文聚焦于Sambert-HifiGan 中文多情感语音合成服务，该服务基于 ModelScope 平台的经典模型构建，已集成 Flask 接口和 WebUI，并修复了datasets、numpy、scipy等关键依赖的版本冲突问题，确保环境稳定可靠。在此基础上，我们将重点介绍如何通过其提供的 HTTP API 实现语音文件的自动化、批量化生成，提升生产效率。

🔧 技术方案选型：为什么选择 API 而非 WebUI？

尽管 WebUI 提供了友好的交互体验，但在批量任务场景下存在明显局限：

| 对比维度 | WebUI 操作 | API 自动化调用 | |----------------|----------------------------------|-----------------------------------| | 操作方式 | 手动点击 + 文本输入 | 脚本驱动 + 批量数据传入 | | 效率 | 单条处理，耗时长 | 支持并发请求，可处理千级文本 | | 可集成性 | 不可嵌入其他系统 | 易与后台系统、定时任务集成 | | 错误重试机制 | 无自动重试 | 可编程实现失败重试逻辑 | | 日志与监控 | 无法记录 | 可记录请求日志、响应状态 |

✅结论：对于需要批量生成语音文件的场景，应优先使用 API 接口进行程序化调用。

🛠️ 实现步骤详解：编写 Python 自动化脚本

步骤 1：确认 API 接口地址与参数格式

启动镜像后，Flask 服务通常运行在http://localhost:7860或平台分配的端口上。通过查看源码或测试请求，可以确定语音合成接口为：

POST http://<host>:<port>/tts

请求体（JSON）示例：

{ "text": "今天天气真好", "speaker": "default", "emotion": "happy" }

返回结果： - 成功时返回.wav音频文件的下载链接或 Base64 编码数据 - 失败时返回错误信息 JSON

步骤 2：准备待合成的文本列表

我们将从一个简单的文本文件texts.txt中读取每行内容作为待合成文本。每行代表一条语音内容。

大家好，欢迎使用语音合成服务。 今天的课程内容是深度学习基础。 请注意，会议将在五分钟后开始。 感谢您的耐心等待，我们将尽快为您处理。

步骤 3：编写批量调用脚本（含异常处理与延迟控制）

以下是完整的 Python 脚本，支持批量发送请求、保存音频文件、自动重试失败请求，并避免高频调用导致服务过载。

import requests import time import os import json from urllib.parse import urljoin # ⚙️ 配置参数 TTS_API_URL = "http://localhost:7860/tts" # 根据实际部署地址修改 TEXT_FILE = "texts.txt" OUTPUT_DIR = "generated_audio" RETRY_LIMIT = 3 DELAY_BETWEEN_REQUESTS = 1.5 # 防止请求过快 # 创建输出目录 os.makedirs(OUTPUT_DIR, exist_ok=True) def call_tts_api(text, speaker="default", emotion="neutral"): payload = { "text": text.strip(), "speaker": speaker, "emotion": emotion } for attempt in range(RETRY_LIMIT): try: response = requests.post(TTS_API_URL, json=payload, timeout=30) if response.status_code == 200: result = response.json() # 判断返回类型：可能是 url 或 base64 数据 if 'audio_url' in result: audio_url = urljoin(TTS_API_URL, result['audio_url']) audio_data = requests.get(audio_url).content elif 'audio' in result and isinstance(result['audio'], str): import base64 audio_data = base64.b64decode(result['audio']) else: print(f"❌ 未知响应格式: {result}") return None return audio_data else: print(f"❌ 请求失败 [{attempt+1}/{RETRY_LIMIT}]: {response.status_code} - {response.text}") except Exception as e: print(f"⚠️ 请求异常 [{attempt+1}/{RETRY_LIMIT}]: {str(e)}") # 重试前等待 if attempt < RETRY_LIMIT - 1: time.sleep(2 ** attempt) # 指数退避 return None def main(): if not os.path.exists(TEXT_FILE): print(f"❗ 文本文件 {TEXT_FILE} 不存在，请检查路径") return with open(TEXT_FILE, 'r', encoding='utf-8') as f: texts = f.readlines() total = len(texts) success_count = 0 print(f"🚀 开始批量生成语音，共 {total} 条文本...") for idx, text in enumerate(texts, 1): if not text.strip(): continue print(f"📝 [{idx}/{total}] 正在合成: {text.strip()[:30]}...") audio_data = call_tts_api(text, emotion="happy") # 可根据需求调整情感 if audio_data: filename = f"speech_{idx:03d}.wav" filepath = os.path.join(OUTPUT_DIR, filename) with open(filepath, 'wb') as af: af.write(audio_data) print(f"✅ 已保存: {filepath}") success_count += 1 else: print(f"❌ 合成失败: {text.strip()[:30]}...") # 控制请求频率 time.sleep(DELAY_BETWEEN_REQUESTS) print(f"\n🎉 批量生成完成！成功 {success_count}/{total} 条，音频文件已保存至 '{OUTPUT_DIR}' 目录。") if __name__ == "__main__": main()

步骤 4：脚本功能解析

✅ 核心特性说明

| 功能点 | 实现方式 | |----------------------|--------------------------------------------------------------------------| |JSON 请求封装| 使用requests.post()发送结构化数据 | |Base64 / URL 支持| 自动识别返回格式并提取音频数据 | |失败重试机制| 最大重试 3 次，采用指数退避策略（1s, 2s, 4s） | |请求节流控制| 每次请求间隔 1.5 秒，防止服务压力过大 | |日志输出清晰| 显示当前进度、成功/失败状态 | |目录自动创建| 若输出目录不存在则自动创建 |

🔄 可扩展建议

添加 CSV 支持：支持从text,speaker,emotion表格导入配置
并发优化：使用concurrent.futures.ThreadPoolExecutor提升吞吐量
日志记录：写入generation.log文件便于追踪
情感动态切换：根据不同文本内容自动匹配情感标签（如“恭喜”→ happy，“警告”→ serious）

⚠️ 实践问题与优化建议

❗ 常见问题及解决方案

| 问题现象 | 原因分析 | 解决方法 | |-----------------------------------|------------------------------------------|------------------------------------------------| | 返回 500 错误或空响应 | 模型加载失败或后端异常 | 检查容器日志docker logs <container_id>| | 音频文件损坏或无法播放 | 返回数据未正确解码 | 确认是否为 Base64 编码，需b64decode| | 请求超时（Timeout） | CPU 推理慢或网络延迟 | 增加timeout=30，降低并发数 | | 中文乱码 | 请求头未设置 UTF-8 |requests默认支持 UTF-8，注意文本编码保存 | | 连续请求导致服务崩溃 | 缺乏限流机制 | 加入time.sleep()控制频率 |

💡 性能优化建议

启用 GPU 加速（若条件允许）
修改模型加载逻辑，在支持 CUDA 的环境中启用 GPU 推理，显著缩短单次合成时间。
使用异步接口（Async Flask）
若服务端支持异步处理，可改用aiohttp客户端实现并发请求，提高整体吞吐量。
缓存去重机制
对重复文本做 MD5 哈希缓存，避免重复请求相同内容。
分批提交 + 断点续传
将大文件拆分为多个批次，记录已完成 ID，支持中断后继续执行。

🧪 示例运行效果

假设texts.txt包含 5 条中文语句，运行脚本后输出如下：

🚀 开始批量生成语音，共 5 条文本... 📝 [1/5] 正在合成: 大家好，欢迎使用语音合成服务。 ✅ 已保存: generated_audio/speech_001.wav 📝 [2/5] 正在合成: 今天的课程内容是深度学习基础。 ✅ 已保存: generated_audio/speech_002.wav ... 🎉 批量生成完成！成功 5/5 条，音频文件已保存至 'generated_audio' 目录。

生成的.wav文件可在本地播放器中直接打开，音质清晰，情感自然。