fix: Edge TTS音频路径修复，静态文件挂载到main.py顶层

main.py

@@ -4,14 +4,18 @@
 
 import os
 import uvicorn
+import aiohttp
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
-from fastapi.responses import FileResponse
+from fastapi.responses import FileResponse, Response
 from fastapi.staticfiles import StaticFiles
 
 # 导入后端服务
 from server import app as api_app
 
+# 确保音频缓存目录存在
+os.makedirs("audio_cache", exist_ok=True)
+
 # 主应用
 app = FastAPI(title="Voice Chat Web")
 
@@ -24,6 +28,35 @@ app.add_middleware(
     allow_headers=["*"],
 )
 
+
+# ChatTTS 音频代理（解决 HTTPS 页面访问 HTTP 资源问题）
+@app.get("/chattts/audio/{filename}")
+async def proxy_chattts_audio(filename: str):
+    """代理 ChatTTS 音频文件"""
+    chattts_url = os.getenv("CHATTTS_URL", "http://192.168.2.5:12002")
+    
+    try:
+        async with aiohttp.ClientSession() as session:
+            async with session.get(
+                f"{chattts_url}/audio/{filename}",
+                timeout=aiohttp.ClientTimeout(total=30)
+            ) as resp:
+                if resp.status != 200:
+                    return Response(content=b'{"detail":"Audio not found"}', status_code=404, media_type="application/json")
+                
+                audio_data = await resp.read()
+                return Response(
+                    content=audio_data,
+                    media_type="audio/wav",
+                    headers={"Cache-Control": "public, max-age=3600"}
+                )
+    except Exception as e:
+        return Response(content=f'{"detail":"{str(e)}"}'.encode(), status_code=500, media_type="application/json")
+
+
+# 挂载音频文件目录（Edge TTS 生成的 MP3）
+app.mount("/audio", StaticFiles(directory="audio_cache"), name="audio")
+
 # 挂载 API
 app.mount("/api", api_app)
 
@@ -33,10 +66,16 @@ app.mount("/static", StaticFiles(directory="static"), name="static")
 
 @app.get("/")
 async def index():
-    """主页"""
+    """主页（原版）"""
     return FileResponse("static/index.html")
 
 
+@app.get("/tts")
+async def tts_page():
+    """TTS版本页面"""
+    return FileResponse("static/tts.html")
+
+
 if __name__ == "__main__":
     PORT = int(os.getenv("PORT", "19019"))
     SSL_KEY = os.getenv("SSL_KEY", "key.pem")

requirements.txt

@@ -1,4 +1,6 @@
 fastapi==0.110.0
 uvicorn==0.27.1
 python-multipart==0.0.9
-aiohttp==3.9.3
+aiohttp==3.9.3
+edge-tts==6.1.9
+requests==2.31.0

server.py

@@ -11,8 +11,13 @@ from datetime import datetime
 import aiohttp
 from fastapi import FastAPI, UploadFile, File, HTTPException, Form
 from fastapi.middleware.cors import CORSMiddleware
+from fastapi.staticfiles import StaticFiles
+from fastapi.responses import FileResponse, Response
 from pydantic import BaseModel
 
+# 导入 TTS 服务
+from tts_service import tts_manager, AUDIO_DIR
+
 # 配置
 MODEL_SERVICE_URL = os.getenv("MODEL_SERVICE_URL", "http://localhost:19018")
 PORT = int(os.getenv("PORT", "19019"))
@@ -186,6 +191,98 @@ async def delete_conversation(conversation_id: str):
         raise HTTPException(status_code=500, detail=str(e))
 
 
+# ========== TTS 相关接口 ==========
+
+class TTSSettings(BaseModel):
+    """TTS 设置"""
+    provider: str = "none"
+    voice: Optional[str] = None
+
+
+class TTSResponse(BaseModel):
+    """TTS 响应"""
+    audio_url: Optional[str]
+    provider: str
+
+
+@app.get("/tts/providers")
+async def get_tts_providers():
+    """获取可用的 TTS 方案列表"""
+    providers = tts_manager.list_providers()
+    voices = tts_manager.get_edge_voices()
+    return {
+        "providers": providers,
+        "voices": voices,
+        "current": tts_manager.current_provider
+    }
+
+
+@app.post("/tts/settings")
+async def set_tts_settings(settings: TTSSettings):
+    """设置 TTS 方案"""
+    tts_manager.set_provider(settings.provider)
+    
+    # 设置音色（仅 Edge TTS）
+    if settings.provider == "edge" and settings.voice:
+        provider = tts_manager.get_provider("edge")
+        if hasattr(provider, 'set_voice'):
+            provider.set_voice(settings.voice)
+    
+    return {
+        "provider": settings.provider,
+        "voice": settings.voice
+    }
+
+
+@app.post("/tts/synthesize")
+async def synthesize_tts(text: str = Form(...), provider: Optional[str] = Form(None)):
+    """
+    合成语音
+    返回音频文件 URL
+    """
+    try:
+        audio_url = await tts_manager.synthesize(text, provider)
+        return TTSResponse(
+            audio_url=audio_url,
+            provider=provider or tts_manager.current_provider
+        )
+    except Exception as e:
+        logger.error(f"TTS synthesis error: {e}")
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+# 挂载音频文件目录
+app.mount("/audio", StaticFiles(directory=AUDIO_DIR), name="audio")
+
+
+# ChatTTS 音频代理（解决 HTTPS 页面访问 HTTP 资源问题）
+@app.get("/chattts/audio/{filename}")
+async def proxy_chattts_audio(filename: str):
+    """代理 ChatTTS 音频文件"""
+    import aiohttp
+    
+    chattts_url = os.getenv("CHATTTS_URL", "http://192.168.2.5:12002")
+    
+    try:
+        async with aiohttp.ClientSession() as session:
+            async with session.get(
+                f"{chattts_url}/audio/{filename}",
+                timeout=aiohttp.ClientTimeout(total=30)
+            ) as resp:
+                if resp.status != 200:
+                    raise HTTPException(status_code=404, detail="Audio not found")
+                
+                audio_data = await resp.read()
+                return Response(
+                    content=audio_data,
+                    media_type="audio/wav",
+                    headers={"Cache-Control": "public, max-age=3600"}
+                )
+    except Exception as e:
+        logger.error(f"Proxy audio error: {e}")
+        raise HTTPException(status_code=500, detail=str(e))
+
+
 if __name__ == "__main__":
     import uvicorn
     uvicorn.run(app, host="0.0.0.0", port=PORT)

tts_service.py

@@ -0,0 +1,254 @@
+"""
+TTS 语音合成模块
+支持多种 TTS 方案
+"""
+
+import os
+import uuid
+import logging
+import asyncio
+from abc import ABC, abstractmethod
+from typing import Optional, Tuple
+from datetime import datetime
+
+# 配置
+AUDIO_DIR = os.getenv("AUDIO_DIR", "audio_cache")
+os.makedirs(AUDIO_DIR, exist_ok=True)
+
+logger = logging.getLogger(__name__)
+
+
+class TTSProvider(ABC):
+    """TTS 提供者抽象类"""
+    
+    @abstractmethod
+    async def synthesize(self, text: str) -> Tuple[str, str]:
+        """
+        合成语音
+        返回: (音频文件路径, 音频URL路径)
+        """
+        pass
+    
+    @abstractmethod
+    def get_name(self) -> str:
+        """获取提供者名称"""
+        pass
+    
+    @abstractmethod
+    def is_available(self) -> bool:
+        """检查是否可用"""
+        pass
+
+
+class EdgeTTSProvider(TTSProvider):
+    """Edge TTS 提供者（微软免费TTS）"""
+    
+    # 可用音色
+    VOICES = {
+        "zh-CN-XiaoxiaoNeural": "晓晓（女）",
+        "zh-CN-YunxiNeural": "云希（男）",
+        "zh-CN-YunyangNeural": "云扬（男）",
+        "zh-CN-XiaochenNeural": "晓晨（女）",
+        "zh-CN-XiaohanNeural": "晓涵（女）",
+        "zh-CN-XiaomengNeural": "晓梦（女）",
+        "zh-CN-XiaomoNeural": "晓墨（女）",
+        "zh-CN-XiaoruiNeural": "晓睿（女）",
+        "zh-CN-XiaoshuangNeural": "晓双（女）",
+        "zh-CN-XiaoxuanNeural": "晓萱（女）",
+        "zh-CN-XiaoyanNeural": "晓颜（女）",
+        "zh-CN-XiaoyouNeural": "晓悠（女）",
+    }
+    
+    DEFAULT_VOICE = "zh-CN-XiaoxiaoNeural"
+    
+    def __init__(self, voice: Optional[str] = None):
+        self.voice = voice or self.DEFAULT_VOICE
+        self._available = None
+    
+    async def synthesize(self, text: str) -> Tuple[str, str]:
+        """使用 Edge TTS 合成语音"""
+        import edge_tts
+        
+        # 生成唯一文件名
+        filename = f"{uuid.uuid4().hex}.mp3"
+        filepath = os.path.join(AUDIO_DIR, filename)
+        
+        # 合成语音
+        communicate = edge_tts.Communicate(text, self.voice)
+        await communicate.save(filepath)
+        
+        # 返回路径
+        audio_url = f"/audio/{filename}"
+        return filepath, audio_url
+    
+    def get_name(self) -> str:
+        return "Edge TTS"
+    
+    def get_voice_name(self) -> str:
+        """获取当前音色名称"""
+        return self.VOICES.get(self.voice, self.voice)
+    
+    def is_available(self) -> bool:
+        """检查 Edge TTS 是否可用"""
+        if self._available is None:
+            try:
+                import edge_tts
+                self._available = True
+            except ImportError:
+                logger.warning("edge-tts not installed")
+                self._available = False
+        return self._available
+    
+    def set_voice(self, voice: str):
+        """设置音色"""
+        if voice in self.VOICES:
+            self.voice = voice
+        else:
+            logger.warning(f"Unknown voice: {voice}, using default")
+
+
+class ChatTTSProvider(TTSProvider):
+    """ChatTTS 提供者（本地部署）"""
+    
+    # ChatTTS 服务地址
+    CHATTTS_URL = os.getenv("CHATTTS_URL", "http://192.168.2.5:12002")
+    
+    def __init__(self):
+        self._available = None
+    
+    async def synthesize(self, text: str) -> Tuple[str, str]:
+        """使用 ChatTTS 合成语音"""
+        import aiohttp
+        
+        async with aiohttp.ClientSession() as session:
+            form = aiohttp.FormData()
+            form.add_field('text', text)
+            
+            async with session.post(
+                f"{self.CHATTTS_URL}/synthesize",
+                data=form,
+                timeout=aiohttp.ClientTimeout(total=60)
+            ) as resp:
+                if resp.status != 200:
+                    error = await resp.text()
+                    raise Exception(f"ChatTTS error: {error}")
+                
+                data = await resp.json()
+                # ChatTTS 返回的 URL 是 /audio/xxx.wav
+                # 改用本地代理路径（解决 HTTPS 页面访问 HTTP 问题）
+                original_url = data['audio_url']
+                # /audio/xxx.wav -> /chattts/audio/xxx.wav (通过本地代理)
+                filename = original_url.split('/')[-1]
+                audio_url = f"/chattts/audio/{filename}"
+                return None, audio_url
+    
+    def get_name(self) -> str:
+        return "ChatTTS"
+    
+    def is_available(self) -> bool:
+        """检查 ChatTTS 是否可用"""
+        if self._available is None:
+            try:
+                import requests
+                resp = requests.get(f"{self.CHATTTS_URL}/health", timeout=5)
+                if resp.status_code == 200:
+                    data = resp.json()
+                    self._available = data.get("status") == "ok"
+                else:
+                    self._available = False
+            except Exception as e:
+                logger.warning(f"ChatTTS check failed: {e}")
+                self._available = False
+        return self._available
+    
+    def set_url(self, url: str):
+        """设置服务地址"""
+        self.CHATTTS_URL = url
+        self._available = None  # 重新检测
+
+
+class NoTTSProvider(TTSProvider):
+    """不使用 TTS"""
+    
+    async def synthesize(self, text: str) -> Tuple[str, str]:
+        return None, None
+    
+    def get_name(self) -> str:
+        return "无 TTS"
+    
+    def is_available(self) -> bool:
+        return True
+
+
+# TTS 管理器
+class TTSManager:
+    """TTS 方案管理"""
+    
+    PROVIDERS = {
+        "edge": EdgeTTSProvider,
+        "chattts": ChatTTSProvider,
+        "none": NoTTSProvider,
+    }
+    
+    def __init__(self, default_provider: str = "none"):
+        self.current_provider = default_provider
+        self._providers = {}
+        
+        # 初始化 Edge TTS（如果可用）
+        edge_provider = EdgeTTSProvider()
+        if edge_provider.is_available():
+            self._providers["edge"] = edge_provider
+        
+        # 初始化 ChatTTS（预留）
+        self._providers["chattts"] = ChatTTSProvider()
+        
+        # 无 TTS
+        self._providers["none"] = NoTTSProvider()
+    
+    def get_provider(self, provider_name: Optional[str] = None) -> TTSProvider:
+        """获取 TTS 提供者"""
+        name = provider_name or self.current_provider
+        return self._providers.get(name, self._providers["none"])
+    
+    def set_provider(self, provider_name: str):
+        """设置当前 TTS 方案"""
+        if provider_name in self._providers:
+            self.current_provider = provider_name
+        else:
+            logger.warning(f"Unknown provider: {provider_name}")
+    
+    def list_providers(self) -> list:
+        """列出所有可用方案"""
+        return [
+            {
+                "name": name,
+                "display_name": provider.get_name(),
+                "available": provider.is_available()
+            }
+            for name, provider in self._providers.items()
+        ]
+    
+    def get_edge_voices(self) -> dict:
+        """获取 Edge TTS 可用音色"""
+        return EdgeTTSProvider.VOICES
+    
+    async def synthesize(self, text: str, provider_name: Optional[str] = None) -> Optional[str]:
+        """
+        合成语音
+        返回音频URL
+        """
+        provider = self.get_provider(provider_name)
+        if not provider.is_available():
+            logger.warning(f"Provider {provider.get_name()} not available")
+            return None
+        
+        try:
+            _, audio_url = await provider.synthesize(text)
+            return audio_url
+        except Exception as e:
+            logger.error(f"TTS synthesis failed: {e}")
+            return None
+
+
+# 全局 TTS 管理器
+tts_manager = TTSManager()