import os import asyncio import aiohttp import base64 from pathlib import Path from pydub import AudioSegment import io import logging from dotenv import load_dotenv import shutil import uuid import yaml from dotenv import load_dotenv from minimax_client import MinimaxAPIClient env_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "minimax.env") load_dotenv(env_path) logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" ) logger = logging.getLogger("volcano_tts_test") def check_dependencies(): dependencies = ["ffmpeg", "ffprobe"] missing = [] for dep in dependencies: if not shutil.which(dep): missing.append(dep) if missing: logger.error(f"缺少必要依赖: {', '.join(missing)}") logger.error( "请安装缺失的依赖项。在Ubuntu上可以使用: sudo apt-get install ffmpeg" ) return False return True env_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "volcano.env") if (os.path.exists(env_path)): load_dotenv(env_path) logger.info(f"已加载环境变量文件: {env_path}") else: logger.warning(f"环境变量文件不存在: {env_path}") PHRASES_TEMPLATES = { "zh": { "welcome": "你好！我是{name}，你想和我聊聊吗？", "tts_error": "抱歉，我没听清楚。", "low_battery": "我的电池快没电了，你能帮我充电吗？", "sleep": "没人和我说话，我要小睡一会儿。" }, "en": { "welcome": "Hello! I'm {name}. Would you like to chat with me?", "tts_error": "Sorry, I didn't catch that.", "low_battery": "My battery is running low. Could you help me recharge?", "sleep": "Nobody is talking to me. I'm going to take a short nap." } } file_prefixes = ["welcome", "tts_error", "low_battery", "sleep"] def find_role_definition_files(base_dir="assets/roles_definitions"): """查找所有角色定义YAML文件""" base_path = Path(base_dir) if not base_path.exists(): logger.error(f"角色定义目录不存在: {base_dir}") return [] yaml_files = list(base_path.glob("**/*.yml")) + list(base_path.glob("**/*.yaml")) return yaml_files def load_role_definition(yaml_file): """加载并解析角色定义YAML文件""" try: with open(yaml_file, 'r', encoding='utf-8') as f: return yaml.safe_load(f) except Exception as e: logger.error(f"解析YAML文件失败 {yaml_file}: {str(e)}") return None def generate_phrases_for_language(lang_code, role_name): """为指定语言生成适当的短语""" if lang_code not in PHRASES_TEMPLATES: logger.warning(f"不支持的语言代码: {lang_code}") return [] templates = PHRASES_TEMPLATES[lang_code] phrases = [] for key in file_prefixes: if key in templates: phrases.append(templates[key].format(name=role_name)) else: logger.warning(f"在{lang_code}语言中找不到{key}模板") phrases.append("") return phrases def _determine_cluster_from_voice_type(voice_type: str) -> str: """根据音色ID自动确定集群类型""" if voice_type and voice_type.startswith("S_"): return "volcano_icl" return "volcano_tts" async def generate_audio_for_role_language(role_config, lang_code, base_dir="assets"): """为角色的特定语言生成音频文件""" if 'multilingual' not in role_config or lang_code not in role_config['multilingual']: logger.warning(f"角色缺少{lang_code}语言配置") return False lang_config = role_config['multilingual'][lang_code] if 'name' not in lang_config or 'url' not in lang_config: logger.warning(f"角色的{lang_code}语言配置缺少必要字段") return False role_name = lang_config['name'] url_path = lang_config['url'] voice_type = lang_config.get('minimax_voice_type', None) if not voice_type and 'minimax_voice_type' in role_config: voice_type = role_config['minimax_voice_type'] logger.info(f"语言{lang_code}配置中未找到音色，使用顶层默认音色: {voice_type}") if not voice_type: logger.warning(f"角色的{lang_code}语言配置缺少 minimax_voice_type，顶层也未定义") return False output_dir = Path(base_dir) / url_path output_dir.mkdir(parents=True, exist_ok=True) # api_access_token = os.getenv("MINIMAX_ACCESS_TOKEN", "") # appid = os.getenv("MINIMAX_APP_ID", "") # if not api_access_token or not appid: # logger.error("缺少必要的配置: MINIMAX_ACCESS_TOKEN 或 MINIMAX_APP_ID") # return False cluster = _determine_cluster_from_voice_type(voice_type) logger.info(f"音色 {voice_type} 自动选择集群: {cluster}") api_key = os.getenv("MINIMAX_API_KEY", "") api_host = os.getenv("MINIMAX_API_HOST", "https://api.minimax.chat") api_client = MinimaxAPIClient(api_key, api_host) speed_ratio = float(os.getenv("SPEED_RATIO", "1.0")) logger.info(f"开始生成角色[{role_name}]的{lang_code}语言音频，音色: {voice_type}, 集群: {cluster}") phrases = generate_phrases_for_language(lang_code, role_name) if not phrases: logger.warning(f"没有为{lang_code}语言生成短语") return False for phrase, file_prefix in zip(phrases, file_prefixes): if not phrase: logger.warning(f"跳过空短语: {file_prefix}") continue output_file_prefix = str(output_dir / file_prefix) output_file = f"{output_file_prefix}.mp3" payload = { "model": "speech-02-hd", "text": phrase, "voice_setting": { "voice_id": voice_type, "speed": float(speed_ratio), "vol": 1.0, "pitch": 0, "emotion": 'happy', }, "audio_setting": { "sample_rate": 16000, "bitrate": 32000, "format": "mp3", "channel": 1 }, "language_boost": lang_code } logger.info(payload) logger.info(f"准备发送请求，文本内容: {phrase}") try: response_data = api_client.post("/v1/t2a_v2", json=payload) audio_data = response_data.get('data', {}).get('audio', '') # print(audio_data) if not audio_data: raise Exception(f"Failed to get audio data from response") # hex->bytes audio_bytes = bytes.fromhex(audio_data) with open(output_file, "wb") as f: f.write(audio_bytes) logger.info(f"TTS合成成功!") logger.info(f"保存音频到: {output_file}") except Exception as e: logger.error(f"TTS请求出错: {str(e)}") await asyncio.sleep(1) return True async def process_all_roles(): """处理所有角色定义文件并生成对应语言的音频""" if not check_dependencies(): return yaml_files = find_role_definition_files() logger.info(f"找到 {len(yaml_files)} 个角色定义文件") target_languages = ['zh', 'en'] assets_base_dir = os.getenv("ASSETS_DIR", "assets") for yaml_file in yaml_files: logger.info(yaml_file.name != "Mengmeng.yaml") if yaml_file.name == "Mengmeng.yaml" or yaml_file.name == "Kuailehu.yaml": logger.info(f"处理角色定义文件: {yaml_file}") role_config = load_role_definition(yaml_file) if not role_config: continue if 'multilingual' not in role_config: logger.warning(f"角色定义文件 {yaml_file} 不包含多语言配置") continue role_name = role_config.get('name', Path(yaml_file).stem) logger.info(f"开始处理角色: {role_name}") for lang_code in target_languages: if lang_code in role_config['multilingual']: logger.info(f"为角色[{role_name}]处理 {lang_code} 语言配置") success = await generate_audio_for_role_language( role_config, lang_code, assets_base_dir ) if success: logger.info(f"角色[{role_name}]的 {lang_code} 语言音频生成完成") else: logger.warning(f"角色[{role_name}]的 {lang_code} 语言音频生成失败") else: logger.info(f"角色[{role_name}]没有 {lang_code} 语言配置") def get_error_description(code, message): """根据错误码返回详细的错误描述""" error_descriptions = { 3001: "无效的请求，请检查参数", 3003: "并发超限，请降低请求频率或增购并发", 3005: "后端服务忙，请稍后重试", 3006: "服务中断，请求已完成/失败之后，相同reqid再次请求", 3010: "文本长度超限，请减少文本长度", 3011: "无效文本，请检查文本内容", 3030: "处理超时，请重试或检查文本", 3031: "处理错误，后端出现异常", 3032: "等待获取音频超时，请重试", 3040: "后端链路连接错误，请重试", 3050: "音色不存在，请检查voice_type参数" } if "quota exceeded for types: xxxxxxxxx_lifetime" in message: return "试用版用量用完，需开通正式版才能继续使用" elif "quota exceeded for types: concurrency" in message: return "并发超过限定值，需减少并发调用或增购并发" elif "Init Engine Instance failed" in message: return "voice_type或cluster参数错误" elif "illegal input text" in message: return "文本无效，无可合成的有效内容" elif "requested grant not found" in message: return "鉴权失败，请检查appid和token是否正确" elif "access denied" in message: return "未拥有当前音色授权，请在控制台购买该音色" return f"错误码: {code}, 错误信息: {message} - {error_descriptions.get(code, '未知错误')}" if __name__ == "__main__": asyncio.run(process_all_roles())