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())