282 lines
10 KiB
Python
282 lines
10 KiB
Python
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())
|