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 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('volcano_voice_type', None) if not voice_type and 'volcano_voice_type' in role_config: voice_type = role_config['volcano_voice_type'] logger.info(f"语言{lang_code}配置中未找到音色，使用顶层默认音色: {voice_type}") if not voice_type: logger.warning(f"角色的{lang_code}语言配置缺少volcano_voice_type，顶层也未定义") return False output_dir = Path(base_dir) / url_path output_dir.mkdir(parents=True, exist_ok=True) api_access_token = os.getenv("VOLCANO_ACCESS_TOKEN", "") appid = os.getenv("VOLCANO_APP_ID", "") if not api_access_token or not appid: logger.error("缺少必要的配置: VOLCANO_ACCESS_TOKEN 或 VOLCANO_APP_ID") return False cluster = _determine_cluster_from_voice_type(voice_type) logger.info(f"音色 {voice_type} 自动选择集群: {cluster}") base_url = os.getenv( "VOLCANO_TTS_BASE_URL", "https://openspeech.bytedance.com/api/v1/tts" ) 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" unique_reqid = str(uuid.uuid4()) payload = { "app": { "appid": appid, "token": "access_token", "cluster": cluster, }, "user": {"uid": "test_user"}, "audio": { "voice_type": voice_type, "encoding": "mp3", "speed_ratio": float(speed_ratio), }, "request": {"reqid": unique_reqid, "text": phrase, "operation": "query"}, } headers = { "Authorization": f"Bearer;{api_access_token}", "Content-Type": "application/json", } logger.info(f"准备发送请求，文本内容: {phrase}, 请求ID: {unique_reqid}") try: async with aiohttp.ClientSession() as session: async with session.post( base_url, headers=headers, json=payload ) as response: logid = response.headers.get('X-Tt-Logid', 'unknown') logger.debug(f"服务端返回Logid: {logid}") if response.status == 200: resp_json = await response.json() if "code" in resp_json: code = resp_json["code"] if code == 3000: # 成功状态码 audio_base64 = resp_json.get("data") if audio_base64: audio_data = base64.b64decode(audio_base64) audio_stream = io.BytesIO(audio_data) sound = AudioSegment.from_file( audio_stream, format="mp3" ) sound = ( sound.set_frame_rate(16000) .set_sample_width(2) .set_channels(1) ) sound.export(output_file, format="mp3", bitrate="16k") logger.info(f"TTS合成成功!") logger.info(f"保存音频到: {output_file}") else: error_msg = "音频数据不存在" logger.error(f"TTS合成失败: {error_msg}") else: error_code = resp_json.get('code') error_msg = resp_json.get('message', '未知错误') error_info = get_error_description(error_code, error_msg) logger.error(f"TTS合成失败: {error_info} (LogID: {logid})") else: logger.error(f"返回数据格式异常，缺少code字段: {resp_json}") else: error_text = await response.text() logger.error( f"TTS合成失败，状态码: {response.status}, 错误: {error_text} (LogID: {logid})" ) 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(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())