235 lines
8.4 KiB
Python
235 lines
8.4 KiB
Python
import os
|
|
import asyncio
|
|
from pathlib import Path
|
|
import io
|
|
import logging
|
|
from dotenv import load_dotenv
|
|
import shutil
|
|
import yaml
|
|
|
|
from pydub import AudioSegment
|
|
from dashscope.audio.tts import SpeechSynthesizer
|
|
import dashscope
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
|
|
)
|
|
logger = logging.getLogger("aliyun_tts_generator")
|
|
|
|
|
|
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__)), "aliyun.env")
|
|
if os.path.exists(env_path):
|
|
load_dotenv(env_path)
|
|
logger.info(f"已加载环境变量文件: {env_path}")
|
|
else:
|
|
logger.warning(f"环境变量文件不存在: {env_path}")
|
|
|
|
|
|
PHRASES_TEMPLATES = {
|
|
"fr": {
|
|
"welcome": "Bonjour! Je suis {name}. Comment puis-je vous aider aujourd'hui?",
|
|
"tts_error": "Désolé, je n'ai pas bien compris ce que vous avez dit.",
|
|
"low_battery": "Attention, ma batterie est faible. J'aurais besoin d'être rechargé bientôt.",
|
|
"sleep": "Je vais me mettre en veille pour économiser de l'énergie. À bientôt!"
|
|
},
|
|
"de": {
|
|
"welcome": "Hallo! Ich bin {name}. Wie kann ich Ihnen heute helfen?",
|
|
"tts_error": "Entschuldigung, ich habe nicht verstanden, was Sie gesagt haben.",
|
|
"low_battery": "Achtung, mein Akku ist schwach. Ich muss bald aufgeladen werden.",
|
|
"sleep": "Ich gehe in den Ruhemodus, um Energie zu sparen. Bis bald!"
|
|
},
|
|
"es": {
|
|
"welcome": "¡Hola! Soy {name}. ¿Cómo puedo ayudarte hoy?",
|
|
"tts_error": "Lo siento, no he entendido lo que has dicho.",
|
|
"low_battery": "Atención, mi batería está baja. Necesitaré recargarme pronto.",
|
|
"sleep": "Voy a entrar en modo de reposo para ahorrar energía. ¡Hasta pronto!"
|
|
}
|
|
}
|
|
|
|
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
|
|
|
|
|
|
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_name = lang_config.get('aliyun_voice_name', None)
|
|
|
|
if not voice_name:
|
|
logger.warning(f"角色的{lang_code}语言配置缺少aliyun_voice_name")
|
|
return False
|
|
|
|
output_dir = Path(base_dir) / url_path
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
api_key = os.getenv("ALIYUN_API_KEY", "")
|
|
if not api_key:
|
|
logger.error("缺少必要的配置: ALIYUN_API_KEY")
|
|
return False
|
|
|
|
dashscope.api_key = api_key
|
|
logger.info(f"开始生成角色[{role_name}]的{lang_code}语言音频，音色: {voice_name}")
|
|
|
|
phrases = generate_phrases_for_language(lang_code, role_name)
|
|
if not phrases:
|
|
logger.warning(f"没有为{lang_code}语言生成短语")
|
|
return False
|
|
|
|
rate = float(os.getenv("RATE", "1.0"))
|
|
|
|
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"
|
|
|
|
logger.info(f"准备发送请求，角色: {role_name}, 语言: {lang_code}, 文件: {file_prefix}")
|
|
|
|
try:
|
|
loop = asyncio.get_running_loop()
|
|
result = await loop.run_in_executor(
|
|
None,
|
|
lambda: SpeechSynthesizer.call(
|
|
model=voice_name,
|
|
text=phrase,
|
|
sample_rate=16000,
|
|
format="mp3",
|
|
volume=50,
|
|
rate=rate,
|
|
pitch=1.0,
|
|
)
|
|
)
|
|
|
|
if result.get_audio_data():
|
|
audio_data = result.get_audio_data()
|
|
|
|
with open(output_file, "wb") as f:
|
|
f.write(audio_data)
|
|
|
|
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")
|
|
|
|
duration = len(sound) / 1000.0 # 转换为秒
|
|
logger.info(f"TTS合成成功! 音频时长: {duration}秒")
|
|
logger.info(f"保存音频到: {output_file}")
|
|
else:
|
|
logger.error("TTS合成失败: 未返回音频数据")
|
|
if hasattr(result, 'code') and result.code != 0:
|
|
logger.error(f"错误码: {result.code}, 错误信息: {result.message}")
|
|
|
|
except Exception as e:
|
|
logger.error(f"TTS请求出错: {str(e)}", exc_info=True)
|
|
|
|
|
|
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 = ['fr', 'de', 'es']
|
|
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} 语言配置")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(process_all_roles())
|