263 lines
9.3 KiB
Python
263 lines
9.3 KiB
Python
import os
|
||
import asyncio
|
||
import aiohttp
|
||
import aiofiles
|
||
import binascii
|
||
import json
|
||
import logging
|
||
from pathlib import Path
|
||
from dotenv import load_dotenv
|
||
|
||
logging.basicConfig(
|
||
level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
|
||
)
|
||
logger = logging.getLogger("minimax_tts_test")
|
||
|
||
# 加载环境变量
|
||
env_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "minimax.env")
|
||
if os.path.exists(env_path):
|
||
load_dotenv(env_path)
|
||
logger.info(f"已加载环境变量文件: {env_path}")
|
||
else:
|
||
logger.warning(f"环境变量文件不存在: {env_path}")
|
||
|
||
async def test_minimax_tts():
|
||
# 获取配置
|
||
api_key = os.getenv("MINIMAX_API_KEY", "")
|
||
group_id = os.getenv("MINIMAX_GROUP_ID", "")
|
||
base_url = os.getenv("MINIMAX_BASE_URL", "https://api.minimax.chat/v1/t2a_v2")
|
||
voice_id = os.getenv("MINIMAX_VOICE_ID", "cartoon-boy-01")
|
||
|
||
if not api_key or not group_id:
|
||
logger.error("缺少必要的配置: MINIMAX_API_KEY 或 MINIMAX_GROUP_ID")
|
||
return
|
||
|
||
# 定义中英文短语对应表
|
||
phrases_dict = {
|
||
# "upgrading": {
|
||
# "zh": "升级中，请勿断电",
|
||
# "en": "Upgrading, please do not power off"
|
||
# },
|
||
# "upgrade_success": {
|
||
# "zh": "升级成功，正在重启设备",
|
||
# "en": "Upgrade successful, restarting device"
|
||
# },
|
||
# "upgrade_failed": {
|
||
# "zh": "升级失败，正在重启设备",
|
||
# "en": "Upgrade failed, restarting device"
|
||
# },
|
||
# "low_energy": {
|
||
# "zh": "感觉没能量了，罢工",
|
||
# "en": "Feeling out of energy, going on strike"
|
||
# },
|
||
# "network_connected": {
|
||
# "zh": "连上网络开始上班了",
|
||
# "en": "Connected to network, starting work"
|
||
# },
|
||
# "network_lost": {
|
||
# "zh": "没网了，下班了",
|
||
# "en": "No network, time to get off work"
|
||
# },
|
||
# "volume_down": {
|
||
# "zh": "音量已调小一点",
|
||
# "en": "Volume turned down a bit"
|
||
# },
|
||
# "max_volume": {
|
||
# "zh": "音量已最大",
|
||
# "en": "Volume is at maximum"
|
||
# },
|
||
# "volume_up": {
|
||
# "zh": "音量已调大一点",
|
||
# "en": "Volume turned up a bit"
|
||
# },
|
||
# "wakeup": {
|
||
# "zh": "我起床了",
|
||
# "en": "I'm awake now"
|
||
# },
|
||
# "enter_network_config": {
|
||
# "zh": "进入配网",
|
||
# "en": "Entering network configuration"
|
||
# },
|
||
# "exit_network_config": {
|
||
# "zh": "退出配网",
|
||
# "en": "Exiting network configuration"
|
||
# }
|
||
# "error_card":{
|
||
# "zh": "你好，卡片有误，请使用本人卡片收听留言"
|
||
# },
|
||
# "no_message":{
|
||
# "zh": "你好，已没可读留言"
|
||
# },
|
||
# "new_message":{
|
||
# "zh": "您有新的留言请注意查收"
|
||
# }
|
||
# "test":{
|
||
# "zh": "橙子我们今天去动物园玩吧？"
|
||
# }
|
||
# "welcome":{
|
||
# "zh": "你好，小朋友，想要和你的好朋友说些什么呢？"
|
||
# }
|
||
# "tts_error":{
|
||
# "zh": "没听到你的声音喔，本轮对话结束"
|
||
# },
|
||
# "message_ok":{
|
||
# "zh": "留言已收到"
|
||
# }
|
||
# "audio_save_fail":{
|
||
# "zh": "留言已收到，但是无法发送。"
|
||
# }
|
||
# "network_ok":{
|
||
# "zh": "已连上网络，可以按键或刷卡跟我对话喔。"
|
||
# },
|
||
# "network_connect":{
|
||
# "zh": "系统启动中，正在等待网络连接。"
|
||
# },
|
||
# "have_rest":{
|
||
# "zh": "小憩一下，待会儿见。"
|
||
# }
|
||
# "bind_nfc_ready":{
|
||
# "zh": "现在开始刷卡绑定设备吧。"
|
||
# },
|
||
# "bind_nfc_finish":{
|
||
# "zh": "卡片绑定成功。"
|
||
# }
|
||
"parent_online": {
|
||
"zh": "你好，你的家长已在线请留言"
|
||
}
|
||
}
|
||
|
||
# 创建输出目录
|
||
assets_dir = os.getenv("ASSETS_DIR", "assets")
|
||
output_dir = Path(assets_dir) / (voice_id + "_tts")
|
||
output_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
logger.info(f"开始测试MiniMax TTS服务，音色: {voice_id}")
|
||
|
||
# 遍历所有短语进行合成
|
||
for phrase_key, phrases in phrases_dict.items():
|
||
for lang, text in phrases.items():
|
||
await synthesize_speech(
|
||
text=text,
|
||
language=lang,
|
||
file_prefix=f"{phrase_key}_{lang}",
|
||
output_dir=output_dir,
|
||
api_key=api_key,
|
||
group_id=group_id,
|
||
base_url=base_url,
|
||
voice_id=voice_id
|
||
)
|
||
|
||
async def synthesize_speech(text, language, file_prefix, output_dir, api_key, group_id, base_url, voice_id):
|
||
"""合成单个语音文件"""
|
||
output_file = output_dir / f"{file_prefix}.mp3"
|
||
|
||
headers = {
|
||
"Authorization": f"Bearer {api_key}",
|
||
"Content-Type": "application/json"
|
||
}
|
||
|
||
# 语言映射
|
||
lang_mapping = {
|
||
"zh": "Chinese",
|
||
# "en": "English",
|
||
# "fr": "French",
|
||
# "de": "German",
|
||
# "es": "Spanish",
|
||
# "yue": "Chinese,Yue"
|
||
}
|
||
|
||
payload = {
|
||
"model": "speech-02-turbo",
|
||
"text": text,
|
||
"stream": True,
|
||
"voice_setting": {
|
||
"voice_id": voice_id,
|
||
"speed": 1.0,
|
||
"vol": 1.0,
|
||
"pitch": 0
|
||
},
|
||
"audio_setting": {
|
||
"sample_rate": 16000,
|
||
"bitrate": 32000,
|
||
"format": "mp3",
|
||
"channel": 1
|
||
},
|
||
"language_boost": lang_mapping.get(language, "auto")
|
||
}
|
||
|
||
url = f"{base_url}?GroupId={group_id}"
|
||
|
||
try:
|
||
logger.info(f"开始合成语音: [{language}] {text}")
|
||
|
||
audio_buffer = bytearray()
|
||
|
||
async with aiohttp.ClientSession() as session:
|
||
async with session.post(url, json=payload, headers=headers) as response:
|
||
if response.status != 200:
|
||
error_text = await response.text()
|
||
logger.error(f"MiniMax TTS请求失败: {response.status}, {error_text}")
|
||
return None
|
||
|
||
logger.info("开始接收MiniMax TTS流式响应")
|
||
|
||
line_count = 0
|
||
async for line in response.content:
|
||
line_count += 1
|
||
logger.debug(f"收到第{line_count}行原始数据: {line}")
|
||
|
||
line = line.strip() # 去除换行符
|
||
if line.startswith(b'data:'):
|
||
try:
|
||
# 解析JSON数据
|
||
json_str = line[5:].decode('utf-8').strip()
|
||
if not json_str:
|
||
logger.debug("空的JSON字符串，跳过")
|
||
continue
|
||
|
||
data_json = json.loads(json_str)
|
||
|
||
if "data" in data_json and "audio" in data_json["data"]:
|
||
status = data_json["data"].get("status", 1)
|
||
audio_hex = data_json["data"]["audio"]
|
||
|
||
if status == 1: # 只处理status=1(合成中)的音频数据，忽略status=2(合成结束)的汇总数据
|
||
if audio_hex:
|
||
try:
|
||
audio_binary = binascii.unhexlify(audio_hex)
|
||
audio_buffer.extend(audio_binary)
|
||
except binascii.Error as e:
|
||
logger.error(f"音频数据解码失败: {e}")
|
||
elif status == 2: # 合成结束
|
||
logger.info("MiniMax TTS流式合成完成")
|
||
break
|
||
else:
|
||
logger.warning(f"响应中没有音频数据: {data_json}")
|
||
except json.JSONDecodeError as e:
|
||
logger.error(f"JSON解析失败: {e}, 原始数据: {line}")
|
||
except Exception as e:
|
||
logger.error(f"处理MiniMax TTS流式响应出错: {str(e)}")
|
||
else:
|
||
logger.debug(f"非data行，跳过: {line}")
|
||
|
||
|
||
# 检查音频缓冲区大小
|
||
if len(audio_buffer) == 0:
|
||
logger.error("音频缓冲区为空，合成可能失败")
|
||
return None
|
||
|
||
logger.info(f"音频合成完成，总大小: {len(audio_buffer)} 字节")
|
||
|
||
# 保存音频文件
|
||
async with aiofiles.open(output_file, 'wb') as f:
|
||
await f.write(audio_buffer)
|
||
|
||
logger.info(f"TTS合成成功! 保存音频到: {output_file}")
|
||
return str(output_file)
|
||
|
||
except Exception as e:
|
||
logger.error(f"MiniMax TTS合成失败: {str(e)}")
|
||
return None
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(test_minimax_tts()) |