Merge pull request #648 from Soulter/feat-edge-tts

feat: 添加对于 edge-tts 支持 #471

astrbot/core/config/default.py

@@ -601,6 +601,13 @@ CONFIG_METADATA_2 = {
                         "openai-tts-voice": "alloy",
                         "timeout": "20",
                     },
+                    "Edge_TTS": {
+                        "id": "edge_tts",
+                        "type": "edge_tts",
+                        "enable": False,
+                        "edge-tts-voice": "zh-CN-XiaoxiaoNeural",
+                        "timeout": 20,
+                    },
                     "FishAudio_TTS(API)": {
                         "id": "fishaudio_tts",
                         "type": "fishaudio_tts_api",

astrbot/core/provider/manager.py

@@ -152,6 +152,8 @@ class ProviderManager():
                     from .sources.whisper_selfhosted_source import ProviderOpenAIWhisperSelfHost as ProviderOpenAIWhisperSelfHost
                 case "openai_tts_api":
                     from .sources.openai_tts_api_source import ProviderOpenAITTSAPI as ProviderOpenAITTSAPI
+                case "edge_tts":
+                    from .sources.edge_tts_source import ProviderEdgeTTS as ProviderEdgeTTS
                 case "fishaudio_tts_api":
                     from .sources.fishaudio_tts_api_source import ProviderFishAudioTTSAPI as ProviderFishAudioTTSAPI
         except (ImportError, ModuleNotFoundError) as e:

astrbot/core/provider/sources/edge_tts_source.py

@@ -0,0 +1,90 @@
+import uuid
+import os
+import edge_tts
+import subprocess
+from ..provider import TTSProvider
+from ..entites import ProviderType
+from ..register import register_provider_adapter
+from astrbot.core import logger
+
+"""
+edge_tts 方式，能够免费、快速生成语音，使用需要先安装edge-tts库
+```
+pip install edge_tts
+```
+Windows 如果提示找不到指定文件，以管理员身份运行命令行窗口，然后再次运行 AstrBot
+"""
+
+@register_provider_adapter("edge_tts", "Microsoft Edge TTS", provider_type=ProviderType.TEXT_TO_SPEECH)
+class ProviderEdgeTTS(TTSProvider):
+    def __init__(
+        self, 
+        provider_config: dict, 
+        provider_settings: dict,
+    ) -> None:
+        super().__init__(provider_config, provider_settings)
+        
+        # 设置默认语音，如果没有指定则使用中文小萱
+        self.voice = provider_config.get("edge-tts-voice", "zh-CN-XiaoxiaoNeural")
+        self.rate = provider_config.get("rate", None)
+        self.volume = provider_config.get("volume", None)
+        self.pitch = provider_config.get("pitch", None)
+        self.timeout = provider_config.get("timeout", 30)
+        
+        self.set_model("edge_tts")
+        
+    async def get_audio(self, text: str) -> str:
+        os.makedirs("data/temp", exist_ok=True)
+        mp3_path = f'data/temp/edge_tts_temp_{uuid.uuid4()}.mp3'
+        wav_path = f'data/temp/edge_tts_{uuid.uuid4()}.wav'
+        
+        # 构建Edge TTS参数
+        kwargs = {"text": text, "voice": self.voice}
+        if self.rate:
+            kwargs["rate"] = self.rate
+        if self.volume:
+            kwargs["volume"] = self.volume
+        if self.pitch:
+            kwargs["pitch"] = self.pitch
+        
+        try:
+            communicate = edge_tts.Communicate(**kwargs)
+            await communicate.save(mp3_path)
+            
+            # 使用ffmpeg将MP3转换为标准WAV格式
+            _ = subprocess.run([
+                "ffmpeg",
+                "-y",                   # 覆盖输出文件
+                "-i", mp3_path,         # 输入文件
+                "-acodec", "pcm_s16le", # 16位PCM编码
+                "-ar", "24000",         # 采样率24kHz (适合微信语音)
+                "-ac", "1",             # 单声道
+                wav_path                # 输出文件
+            ], capture_output=True, check=True)
+            
+
+            os.remove(mp3_path)
+            if os.path.exists(wav_path) and os.path.getsize(wav_path) > 0:
+                return wav_path
+            else:
+                logger.error("生成的WAV文件不存在或为空")
+                raise RuntimeError("生成的WAV文件不存在或为空")
+                
+        except subprocess.CalledProcessError as e:
+            logger.error(f"FFmpeg转换失败: {e.stderr.decode() if e.stderr else str(e)}")
+            try:
+                if os.path.exists(mp3_path):
+                    os.remove(mp3_path)
+            except:
+                pass
+            raise RuntimeError(f"FFmpeg转换失败: {str(e)}")
+            
+        except Exception as e:
+            logger.error(f"音频生成失败: {str(e)}")
+            try:
+                if os.path.exists(mp3_path):
+                    os.remove(mp3_path)
+            except:
+                pass
+            raise RuntimeError(f"音频生成失败: {str(e)}")
+