""" Edge TTS API 接口实现 提供文本转语音功能的API接口 """ import edge_tts import asyncio import logging from typing import Optional, Dict, Any import tempfile import os import zipfile import json from pydub import AudioSegment import io import aiohttp # 配置日志 logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class EdgeTTSAPI: def __init__(self): # 中文语音 self.chinese_voices = [ "zh-CN-XiaoxiaoNeural", "zh-CN-XiaoyiNeural", "zh-CN-YunjianNeural", "zh-CN-YunxiNeural", "zh-CN-YunxiaNeural", "zh-CN-YunyangNeural", "zh-CN-liaoning-XiaobeiNeural", "zh-CN-shaanxi-XiaoniNeural", "zh-HK-HiuGaaiNeural", "zh-HK-HiuMaanNeural", "zh-HK-WanLungNeural", "zh-TW-HsiaoChenNeural", "zh-TW-YunJheNeural", "zh-TW-HsiaoYuNeural" ] # 英文语音 self.english_voices = [ "en-US-AriaNeural", "en-US-GuyNeural", "en-US-JennyNeural", "en-US-RogerNeural", "en-GB-SoniaNeural", "en-GB-RyanNeural" ] # 日语语音 self.japanese_voices = [ "ja-JP-NanamiNeural", "ja-JP-KeitaNeural" ] # 韩语语音 self.korean_voices = [ "ko-KR-SunHiNeural", "ko-KR-InJoonNeural" ] # 其他语言语音 self.other_voices = [ "de-DE-KatjaNeural", "de-DE-ConradNeural", "fr-FR-DeniseNeural", "fr-FR-HenriNeural", "es-ES-ElviraNeural", "es-ES-AlvaroNeural" ] # 合并所有语音 self.all_voices = (self.chinese_voices + self.english_voices + self.japanese_voices + self.korean_voices + self.other_voices) def get_available_voices(self, language: Optional[str] = None) -> list: """ 获取可用的语音列表 :param language: 语言代码，如 'zh', 'en', 'ja' 等，如果为None则返回所有语音 :return: 语音列表 """ if language is None: return self.all_voices elif language == 'zh': return self.chinese_voices elif language == 'en': return self.english_voices elif language == 'ja': return self.japanese_voices elif language == 'ko': return self.korean_voices else: # 根据语言代码筛选 return [voice for voice in self.all_voices if voice.startswith(language)] async def text_to_speech(self, text: str, voice: str = "zh-CN-XiaoxiaoNeural", rate: int = 0, pitch: int = 0, output_file: Optional[str] = None, output_format: str = "mp3") -> Optional[str]: """ 将文本转换为语音 :param text: 输入文本 :param voice: 选择的语音 :param rate: 语速调整 (-50 到 50) :param pitch: 音调调整 (-50 到 50) :param output_file: 输出文件路径，如果为None则创建临时文件 :param output_format: 输出格式，"mp3" 或 "wav" :return: 输出文件路径，失败返回None """ if not text or not text.strip(): logger.error("输入文本为空") return None # 验证语音是否可用 if voice not in self.all_voices: logger.warning(f"语音 {voice} 不在可用列表中，使用默认语音") voice = "zh-CN-XiaoxiaoNeural" # 创建输出文件路径 if output_file is None: ext = f".{output_format}" _, output_file = tempfile.mkstemp(suffix=ext) else: # 确保输出文件有正确的扩展名 if not output_file.lower().endswith((f'.{output_format}',)): output_file += f'.{output_format}' try: # 设置语速和音调 rate_str = f"{rate:+d}%" if rate >= 0 else f"{rate:d}%" pitch_str = f"{pitch:+d}Hz" if pitch >= 0 else f"{pitch:d}Hz" #直使用使用纯文本，避免TTS读出XML标签 # Edge TTS会自动处理文本，不需要SSML包装 ssml_text = text # 创建Communicate对象并保存音频 communicate = edge_tts.Communicate(ssml_text, voice) temp_mp3 = output_file if output_format == "mp3" else output_file + ".mp3" await communicate.save(temp_mp3) # 如果需要转换格式 if output_format == "wav": # 将MP3转换为WAV audio = AudioSegment.from_mp3(temp_mp3) audio.export(output_file, format="wav") # 删除临时MP3文件 os.remove(temp_mp3) else: output_file = temp_mp3 logger.info(f"语音生成成功: {output_file}") return output_file except Exception as e: logger.error(f"语音转换失败: {str(e)}") # 如果生成失败，删除临时文件 if output_file and os.path.exists(output_file): try: os.remove(output_file) except: pass return None async def text_to_speech_hf_sync(self, text: str, voice: str = "zh-CN-XiaoxiaoNeural", rate: int = 0, pitch: int = 0, output_file: Optional[str] = None, output_format: str = "mp3") -> Optional[str]: """ 同步版本的Hugging Face API文本转语音函数 通过异步包装实现同步调用 :param text: 输入文本 :param voice: 选择的语音 :param rate: 语速调整 (-50到50) :param pitch:音调调整 (-50到 50) :param output_file: 输出文件路径，如果为None则创建临时文件 :param output_format: 输出格式，"mp3" 或 "wav" :return: 输出文件路径，失败返回None """ import asyncio return asyncio.run(self.text_to_speech_hf(text, voice, rate, pitch, output_file, output_format)) async def text_to_speech_hf(self, text: str, voice: str = "zh-CN-XiaoxiaoNeural", rate: int = 0, pitch: int = 0, output_file: Optional[str] = None, output_format: str = "mp3") -> Optional[str]: """ 使用Hugging Face Spaces API将文本转换为语音 :param text: 输入文本 :param voice: 选择的语音 :param rate: 语速调整 (-50 到 50) :param pitch: 音调调整 (-50 到 50) :param output_file: 输出文件路径，如果为None则创建临时文件 :param output_format: 输出格式，"mp3" 或 "wav" :return: 输出文件路径，失败返回None """ if not text or not text.strip(): logger.error("输入文本为空") return None # 创建输出文件路径 if output_file is None: ext = f".{output_format}" _, output_file = tempfile.mkstemp(suffix=ext) try: # 使用正确的Hugging Face Spaces API # 使用Gradio API端点 api_url = "https://chaore-ttsedge.hf.space/gradio_api" # 构建请求数据 payload = { "data": [text, voice, rate, pitch, output_format], "event_data": None, "fn_index": 0, "session_hash": "abc123test" } async with aiohttp.ClientSession() as session: headers = { "Content-Type": "application/json", "Origin": "https://chaore-ttsedge.hf.space", "Referer": "https://chaore-ttsedge.hf.space/" } # 使用命名端点调用 predict_url = f"{api_url}/call/text_to_speech" async with session.post(predict_url, json=payload, headers=headers) as response: if response.status == 200: result = await response.json() if 'event_id' in result: # 获取任务状态 event_id = result['event_id'] result_url = f"{api_url}/call/text_to_speech/{event_id}" # 等待任务完成 import time max_wait = 30 # 最大等待时间30秒 wait_time = 0 while wait_time < max_wait: await asyncio.sleep(2) wait_time += 2 async with session.get(result_url, headers=headers) as status_response: if status_response.status == 200: # 读取SSE流 async for line in status_response.content: try: line_str = line.decode('utf-8').strip() if line_str.startswith('data:'): data_content = line_str[5:].strip() if data_content and data_content.lower() != 'null': try: json_data = json.loads(data_content) #检查是否有音频数据 if isinstance(json_data, dict): if 'data' in json_data and json_data['data']: if len(json_data['data']) > 0: potential_audio = json_data['data'][0] if isinstance(potential_audio, str) and potential_audio.startswith('data:'): #处理base64编码的音频数据 import base64 header, encoded = potential_audio.split(',', 1) audio_data = base64.b64decode(encoded) # 保存到输出文件 with open(output_file, 'wb') as f: f.write(audio_data) logger.info(f"从Hugging Face API获取语音成功: {output_file}") return output_file #检查是否完成 if 'status' in json_data and json_data['status'] == 'COMPLETE': if 'data' in json_data and json_data['data']: if len(json_data['data']) > 0: result_item = json_data['data'][0] if isinstance(result_item, str) and result_item.startswith('data:'): import base64 header, encoded = result_item.split(',', 1) audio_data = base64.b64decode(encoded) with open(output_file, 'wb') as f: f.write(audio_data) logger.info(f"从Hugging Face API获取语音成功: {output_file}") return output_file break elif 'status' in json_data and json_data['status'] == 'FAILED': logger.error("Hugging Face API任务失败") return None except json.JSONDecodeError: #处理纯文本数据 if data_content.startswith('data:audio/') or data_content.startswith('data:application/'): import base64 header, encoded = data_content.split(',', 1) audio_data = base64.b64decode(encoded) with open(output_file, 'wb') as f: f.write(audio_data) logger.info(f"从Hugging Face API获取语音成功: {output_file}") return output_file except Exception as e: continue else: logger.error("Hugging Face API任务超时") return None else: logger.error(f"Hugging Face API返回格式错误: {result}") return None else: logger.error(f"Hugging Face API请求失败: {response.status}") logger.error(f"响应内容: {await response.text()}") return None except Exception as e: logger.error(f"从Hugging Face API获取语音失败: {str(e)}") # 如果生成失败，删除临时文件 if output_file and os.path.exists(output_file): try: os.remove(output_file) except: pass return None async def get_voice_info(self, voice: str) -> Optional[Dict[str, Any]]: """ 获取语音信息 :param voice: 语音名称 :return: 语音信息字典 """ if voice not in self.all_voices: return None lang_parts = voice.split('-') language_code = f"{lang_parts[0]}-{lang_parts[1]}" # 确定性别 gender = "Female" if any(neural in voice.lower() for neural in ['guy', 'roger', 'ryan', 'keita', 'alvaro', 'conrad', 'henri', 'jake', 'eric', 'tony']): gender = "Male" return { "name": voice, "language": language_code, "gender": gender, "locale": f"{lang_parts[0]}-{lang_parts[1]}-{lang_parts[2] if len(lang_parts) > 2 else 'General'}" } async def batch_text_to_speech(self, texts: list, voice: str = "zh-CN-XiaoxiaoNeural", rate: int = 0, pitch: int = 0, output_format: str = "mp3") -> list: """ 批量将文本转换为语音 :param texts: 文本列表 :param voice: 选择的语音 :param rate: 语速调整 :param pitch: 音调调整 :param output_format: 输出格式 :return: 生成的音频文件路径列表 """ results = [] for text in texts: if text.strip(): # 只处理非空文本 audio_file = await self.text_to_speech(text, voice, rate, pitch, output_format=output_format) results.append(audio_file) else: results.append(None) return results async def create_audio_project(self, project_name: str, segments: list, voice: str = "zh-CN-XiaoxiaoNeural", rate: int = 0, pitch: int = 0, output_format: str = "mp3") -> Optional[str]: """ 创建音频项目，将多个文本片段合并为一个音频文件 :param project_name: 项目名称 :param segments: 包含文本和时间信息的片段列表，格式: [{"text": "文本", "delay": 毫秒}] :param voice: 选择的语音 :param rate: 语速调整 :param pitch: 音调调整 :param output_format: 输出格式 :return: 生成的音频文件路径 """ try: # 创建临时目录存储各个片段 temp_dir = tempfile.mkdtemp() segment_files = [] # 生成每个片段的音频 for i, segment in enumerate(segments): text = segment.get("text", "") if not text.strip(): continue delay = segment.get("delay", 0) # 延迟时间（毫秒） # 生成音频片段 segment_file = os.path.join(temp_dir, f"segment_{i}.{output_format}") result = await self.text_to_speech(text, voice, rate, pitch, segment_file, output_format) if result: segment_files.append((result, delay)) if not segment_files: logger.error("没有生成任何音频片段") return None # 合并音频片段 combined_audio = AudioSegment.empty() for audio_file, delay in segment_files: if delay > 0: # 添加静音间隔 silence = AudioSegment.silent(duration=delay) combined_audio += silence # 添加音频片段 segment_audio = AudioSegment.from_file(audio_file, format=output_format) combined_audio += segment_audio # 生成最终输出文件 output_path = os.path.join(temp_dir, f"{project_name}.{output_format}") combined_audio.export(output_path, format=output_format) # 清理临时片段文件 for audio_file, _ in segment_files: try: os.remove(audio_file) except: pass return output_path except Exception as e: logger.error(f"创建音频项目失败: {str(e)}") return None async def export_voice_settings(self) -> str: """ 导出语音设置 :return: JSON格式的设置字符串 """ settings = { "chinese_voices": self.chinese_voices, "english_voices": self.english_voices, "japanese_voices": self.japanese_voices, "korean_voices": self.korean_voices, "other_voices": self.other_voices, "all_voices": self.all_voices } return json.dumps(settings, ensure_ascii=False, indent=2) # 创建全局API实例 tts_api = EdgeTTSAPI() # 同步包装函数，方便在非异步环境中调用 def sync_text_to_speech(text: str, voice: str = "zh-CN-XiaoxiaoNeural", rate: int = 0, pitch: int = 0, output_file: Optional[str] = None, output_format: str = "mp3") -> Optional[str]: """ 同步版本的文本转语音函数 """ return asyncio.run(tts_api.text_to_speech(text, voice, rate, pitch, output_file, output_format)) def sync_text_to_speech_hf(text: str, voice: str = "zh-CN-XiaoxiaoNeural", rate: int = 0, pitch: int = 0, output_file: Optional[str] = None, output_format: str = "mp3") -> Optional[str]: """ 同步版本的Hugging Face API文本转语音函数 """ return asyncio.run(tts_api.text_to_speech_hf(text, voice, rate, pitch, output_file, output_format))