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