coder/voice-chat-web
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 添加TTS语音合成功能 - Edge TTS方案完整实现，ChatTTS预留接口

Browse Source
This commit is contained in:
hubian
2026-04-22 10:41:44 +08:00
parent bcb0fbb384
commit e00b0218a0
8 changed files with 1174 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

BIN
__pycache__/server.cpython-310.pyc
View File

Binary file not shown.

BIN
__pycache__/tts_service.cpython-310.pyc Normal file
View File

Binary file not shown.

BIN
logs/server.log
View File

Binary file not shown.

8
main.py
View File

@@ -33,10 +33,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")

3
requirements.txt
View File

@@ -1,4 +1,5 @@
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

69
server.py
View File

@@ -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
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,70 @@ 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")
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=PORT)

873
static/tts.html Normal file
View File

@@ -0,0 +1,873 @@
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>语音对话（TTS版）</title>
<style>
* {
margin: 0;
padding: 0;
box-sizing: border-box;
}
body {
font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif;
background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
min-height: 100vh;
display: flex;
justify-content: center;
align-items: center;
padding: 20px;
}
.container {
max-width: 700px;
width: 100%;
background: white;
border-radius: 20px;
padding: 30px;
box-shadow: 0 20px 60px rgba(0,0,0,0.3);
}
.header {
text-align: center;
margin-bottom: 20px;
}
.header h1 {
color: #333;
font-size: 28px;
margin-bottom: 10px;
}
.status-indicator {
display: inline-flex;
align-items: center;
gap: 8px;
font-size: 14px;
color: #666;
}
.status-dot {
width: 10px;
height: 10px;
border-radius: 50%;
background: #ccc;
}
.status-dot.online {
background: #4CAF50;
animation: pulse 2s infinite;
}
.status-dot.offline {
background: #f44336;
}
@keyframes pulse {
0%, 100% { opacity: 1; }
50% { opacity: 0.5; }
}
/* TTS 设置区域 */
.tts-section {
background: #f5f7fa;
border-radius: 10px;
padding: 15px;
margin-bottom: 20px;
}
.tts-section h3 {
color: #333;
font-size: 16px;
margin-bottom: 12px;
}
.tts-options {
display: flex;
gap: 10px;
margin-bottom: 10px;
}
.tts-option {
flex: 1;
padding: 10px;
border: 2px solid #eee;
border-radius: 8px;
cursor: pointer;
text-align: center;
transition: all 0.2s;
}
.tts-option:hover {
border-color: #667eea;
}
.tts-option.selected {
border-color: #667eea;
background: rgba(102,126,234,0.1);
}
.tts-option.disabled {
opacity: 0.5;
cursor: not-allowed;
}
.tts-option .name {
font-weight: bold;
color: #333;
}
.tts-option .status {
font-size: 12px;
color: #999;
}
.voice-select {
margin-top: 10px;
}
.voice-select select {
width: 100%;
padding: 8px;
border: 1px solid #ddd;
border-radius: 5px;
}
/* 录音区域 */
.record-section {
text-align: center;
margin-bottom: 20px;
}
.record-btn {
width: 100px;
height: 100px;
border-radius: 50%;
border: none;
background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
color: white;
font-size: 24px;
cursor: pointer;
transition: all 0.3s ease;
display: flex;
flex-direction: column;
align-items: center;
justify-content: center;
gap: 5px;
margin: 0 auto;
}
.record-btn:hover {
transform: scale(1.05);
box-shadow: 0 10px 30px rgba(102,126,234,0.4);
}
.record-btn.recording {
background: linear-gradient(135deg, #f44336 0%, #e53935 100%);
animation: recording-pulse 1s infinite;
}
@keyframes recording-pulse {
0%, 100% { transform: scale(1); }
50% { transform: scale(1.1); }
}
.record-status {
margin-top: 10px;
font-size: 14px;
color: #666;
}
.waveform {
display: flex;
justify-content: center;
gap: 3px;
margin-top: 10px;
height: 30px;
}
.wave-bar {
width: 4px;
height: 10px;
background: #667eea;
border-radius: 2px;
animation: wave 0.5s infinite ease-in-out;
}
@keyframes wave {
0%, 100% { height: 10px; }
50% { height: 25px; }
}
/* 文字输入 */
.text-section {
margin: 15px 0;
}
.text-input-wrapper {
display: flex;
gap: 10px;
}
.text-input {
flex: 1;
padding: 10px 12px;
border: 2px solid #eee;
border-radius: 8px;
font-size: 14px;
outline: none;
}
.text-input:focus {
border-color: #667eea;
}
.send-text-btn {
padding: 10px 15px;
border: none;
border-radius: 8px;
background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
color: white;
cursor: pointer;
}
/* 消息区域 */
.chat-section {
max-height: 350px;
overflow-y: auto;
border: 1px solid #eee;
border-radius: 10px;
padding: 15px;
background: #f9f9f9;
}
.message {
margin-bottom: 12px;
padding: 10px 12px;
border-radius: 10px;
max-width: 85%;
}
.message.user {
background: #667eea;
color: white;
margin-left: auto;
}
.message.assistant {
background: white;
color: #333;
border: 1px solid #eee;
}
.message .role {
font-size: 12px;
font-weight: bold;
margin-bottom: 5px;
opacity: 0.8;
}
.message .content {
font-size: 14px;
line-height: 1.5;
}
.audio-content {
display: flex;
align-items: center;
}
.play-btn {
display: inline-flex;
align-items: center;
gap: 6px;
padding: 6px 12px;
border-radius: 15px;
border: none;
background: rgba(255,255,255,0.2);
cursor: pointer;
font-size: 13px;
}
.play-btn:hover {
background: rgba(255,255,255,0.3);
}
.tts-play-btn {
background: linear-gradient(135deg, #4CAF50 0%, #45a049 100%);
color: white;
margin-left: 10px;
}
.loading {
display: flex;
justify-content: center;
gap: 8px;
padding: 15px;
}
.loading-dot {
width: 8px;
height: 8px;
border-radius: 50%;
background: #667eea;
animation: loading-bounce 0.6s infinite;
}
@keyframes loading-bounce {
0%, 80%, 100% { transform: scale(0); }
40% { transform: scale(1); }
}
.actions {
margin-top: 15px;
text-align: center;
}
.clear-btn {
padding: 8px 15px;
border: 1px solid #ddd;
border-radius: 6px;
background: white;
cursor: pointer;
}
.error-message {
background: #ffebee;
color: #c62828;
padding: 10px;
border-radius: 8px;
margin-bottom: 10px;
text-align: center;
}
.hint {
font-size: 13px;
color: #999;
text-align: center;
}
</style>
</head>
<body>
<div class="container">
<div class="header">
<h1>🎤 语音对话（TTS版）</h1>
<div class="status-indicator">
<span class="status-dot" id="statusDot"></span>
<span id="statusText">检测连接...</span>
</div>
</div>
<!-- TTS 设置 -->
<div class="tts-section">
<h3>🔊 TTS语音合成设置</h3>
<div class="tts-options" id="ttsOptions">
<div class="tts-option" data-provider="none">
<div class="name">❌ 无 TTS</div>
<div class="status">只显示文字</div>
</div>
<div class="tts-option" data-provider="edge">
<div class="name">🌐 Edge TTS</div>
<div class="status" id="edgeStatus">检测中...</div>
</div>
<div class="tts-option disabled" data-provider="chattts">
<div class="name">🤖 ChatTTS</div>
<div class="status">暂未部署</div>
</div>
</div>
<div class="voice-select" id="voiceSelect" style="display: none;">
<label>音色选择：</label>
<select id="voiceDropdown">
<option value="zh-CN-XiaoxiaoNeural">晓晓（女）</option>
<option value="zh-CN-YunxiNeural">云希（男）</option>
<option value="zh-CN-YunyangNeural">云扬（男）</option>
<option value="zh-CN-XiaochenNeural">晓晨（女）</option>
<option value="zh-CN-XiaohanNeural">晓涵（女）</option>
<option value="zh-CN-XiaoyouNeural">晓悠（女）</option>
</select>
</div>
</div>
<!-- 录音 -->
<div class="record-section">
<button class="record-btn" id="recordBtn">
<span class="icon">🎤</span>
<span class="text">录音</span>
</button>
<div class="record-status" id="recordStatus">点击按钮开始录音</div>
<div class="waveform" id="waveform" style="display: none;">
<div class="wave-bar"></div>
<div class="wave-bar"></div>
<div class="wave-bar"></div>
<div class="wave-bar"></div>
<div class="wave-bar"></div>
<div class="wave-bar"></div>
<div class="wave-bar"></div>
<div class="wave-bar"></div>
</div>
</div>
<!-- 文字输入 -->
<div class="text-section">
<div class="text-input-wrapper">
<input type="text" id="textInput" placeholder="输入文字消息..." class="text-input">
<button id="sendTextBtn" class="send-text-btn">发送</button>
</div>
</div>
<!-- 消息区域 -->
<div class="chat-section" id="chatSection">
<div class="hint">开始对话吧！</div>
</div>
<div class="actions">
<button class="clear-btn" id="clearBtn">清除对话</button>
</div>
</div>
<script>
const API_URL = '/api';
// 状态
let isRecording = false;
let audioContext = null;
let audioStream = null;
let scriptProcessor = null;
let recordedBuffers = [];
let conversationId = null;
let currentTTSProvider = 'none';
let currentVoice = 'zh-CN-XiaoxiaoNeural';
// 元素
const statusDot = document.getElementById('statusDot');
const statusText = document.getElementById('statusText');
const recordBtn = document.getElementById('recordBtn');
const recordStatus = document.getElementById('recordStatus');
const waveform = document.getElementById('waveform');
const chatSection = document.getElementById('chatSection');
const clearBtn = document.getElementById('clearBtn');
const textInput = document.getElementById('textInput');
const sendTextBtn = document.getElementById('sendTextBtn');
const ttsOptions = document.getElementById('ttsOptions');
const voiceSelect = document.getElementById('voiceSelect');
const voiceDropdown = document.getElementById('voiceDropdown');
// 初始化
async function init() {
await checkStatus();
await loadTTSProviders();
// 定时检查状态
setInterval(checkStatus, 10000);
}
// 检查服务状态
async function checkStatus() {
try {
const resp = await fetch(`${API_URL}/status`);
const data = await resp.json();
if (data.model_online) {
statusDot.className = 'status-dot online';
statusText.textContent = '模型服务已连接';
} else {
statusDot.className = 'status-dot offline';
statusText.textContent = '模型服务离线';
}
} catch (e) {
statusDot.className = 'status-dot offline';
statusText.textContent = '服务连接失败';
}
}
// 加载 TTS 方案
async function loadTTSProviders() {
try {
const resp = await fetch(`${API_URL}/tts/providers`);
const data = await resp.json();
// 更新状态
data.providers.forEach(p => {
if (p.name === 'edge') {
const statusEl = document.getElementById('edgeStatus');
statusEl.textContent = p.available ? '可用 ✓' : '不可用';
const optionEl = ttsOptions.querySelector('[data-provider="edge"]');
if (p.available) {
optionEl.classList.remove('disabled');
}
}
});
// 设置当前方案
currentTTSProvider = data.current || 'none';
selectProvider(currentTTSProvider);
} catch (e) {
console.error('加载 TTS 方案失败:', e);
}
}
// 选择 TTS 方案
function selectProvider(provider) {
ttsOptions.querySelectorAll('.tts-option').forEach(el => {
el.classList.remove('selected');
});
const option = ttsOptions.querySelector(`[data-provider="${provider}"]`);
if (option && !option.classList.contains('disabled')) {
option.classList.add('selected');
currentTTSProvider = provider;
// 显示/隐藏音色选择
voiceSelect.style.display = provider === 'edge' ? 'block' : 'none';
// 保存设置
saveTTSSettings();
}
}
// 保存 TTS 设置
async function saveTTSSettings() {
try {
await fetch(`${API_URL}/tts/settings`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({
provider: currentTTSProvider,
voice: currentVoice
})
});
} catch (e) {
console.error('保存设置失败:', e);
}
}
// WAV 创建
function createWavFile(audioBuffer, sampleRate = 16000) {
const numChannels = 1;
const bitsPerSample = 16;
const bytesPerSample = bitsPerSample / 8;
const blockAlign = numChannels * bytesPerSample;
const byteRate = sampleRate * blockAlign;
const dataSize = audioBuffer.length * bytesPerSample;
const headerSize = 44;
const totalSize = headerSize + dataSize;
const buffer = new ArrayBuffer(totalSize);
const view = new DataView(buffer);
writeString(view, 0, 'RIFF');
view.setUint32(4, totalSize - 8, true);
writeString(view, 8, 'WAVE');
writeString(view, 12, 'fmt ');
view.setUint32(16, 16, true);
view.setUint16(20, 1, true);
view.setUint16(22, numChannels, true);
view.setUint32(24, sampleRate, true);
view.setUint32(28, byteRate, true);
view.setUint16(32, blockAlign, true);
view.setUint16(34, bitsPerSample, true);
writeString(view, 36, 'data');
view.setUint32(40, dataSize, true);
floatTo16BitPCM(view, 44, audioBuffer);
return new Blob([buffer], { type: 'audio/wav' });
}
function writeString(view, offset, string) {
for (let i = 0; i < string.length; i++) {
view.setUint8(offset + i, string.charCodeAt(i));
}
}
function floatTo16BitPCM(view, offset, input) {
for (let i = 0; i < input.length; i++, offset += 2) {
const s = Math.max(-1, Math.min(1, input[i]));
view.setInt16(offset, s < 0 ? s * 0x8000 : s * 0x7FFF, true);
}
}
// 初始化录音
async function initAudio() {
try {
audioStream = await navigator.mediaDevices.getUserMedia({
audio: { echoCancellation: true, noiseSuppression: true, sampleRate: 16000 }
});
audioContext = new (window.AudioContext || window.webkitAudioContext)({ sampleRate: 16000 });
const source = audioContext.createMediaStreamSource(audioStream);
scriptProcessor = audioContext.createScriptProcessor(4096, 1, 1);
scriptProcessor.onaudioprocess = (e) => {
if (isRecording) {
recordedBuffers.push(e.inputBuffer.getChannelData(0).slice());
}
};
source.connect(scriptProcessor);
scriptProcessor.connect(audioContext.destination);
return true;
} catch (e) {
showError('无法访问麦克风');
return false;
}
}
// 开始录音
async function startRecording() {
if (!audioContext) {
const success = await initAudio();
if (!success) return;
}
recordedBuffers = [];
isRecording = true;
recordBtn.classList.add('recording');
recordBtn.querySelector('.icon').textContent = '⏹️';
recordBtn.querySelector('.text').textContent = '停止';
recordStatus.textContent = '正在录音...';
waveform.style.display = 'flex';
}
// 停止录音
function stopRecording() {
if (isRecording) {
isRecording = false;
const totalLength = recordedBuffers.reduce((acc, buf) => acc + buf.length, 0);
const mergedBuffer = new Float32Array(totalLength);
let offset = 0;
for (const buf of recordedBuffers) {
mergedBuffer.set(buf, offset);
offset += buf.length;
}
const wavBlob = createWavFile(mergedBuffer, 16000);
const duration = Math.round(totalLength / 16000);
recordBtn.classList.remove('recording');
recordBtn.querySelector('.icon').textContent = '🎤';
recordBtn.querySelector('.text').textContent = '录音';
recordStatus.textContent = '处理中...';
waveform.style.display = 'none';
sendAudio(wavBlob, duration);
}
}
// 发送音频
async function sendAudio(audioBlob, duration) {
showLoading();
const formData = new FormData();
formData.append('audio', audioBlob, 'recording.wav');
if (conversationId) formData.append('conversation_id', conversationId);
try {
const resp = await fetch(`${API_URL}/voice/chat`, {
method: 'POST',
body: formData
});
const data = await resp.json();
conversationId = data.conversation_id;
addMessage('user', audioBlob, duration);
// TTS 合成
let ttsAudioUrl = null;
if (currentTTSProvider !== 'none') {
ttsAudioUrl = await synthesizeTTS(data.reply);
}
addMessage('assistant', data.reply, ttsAudioUrl);
recordStatus.textContent = '点击按钮开始录音';
} catch (e) {
showError('发送失败: ' + e.message);
}
}
// 发送文字
async function sendText(text) {
if (!text.trim()) return;
showLoading();
const formData = new FormData();
formData.append('text', text);
if (conversationId) formData.append('conversation_id', conversationId);
try {
const resp = await fetch(`${API_URL}/voice/text`, {
method: 'POST',
body: formData
});
const data = await resp.json();
conversationId = data.conversation_id;
addMessage('user', text);
// TTS 合成
let ttsAudioUrl = null;
if (currentTTSProvider !== 'none') {
ttsAudioUrl = await synthesizeTTS(data.reply);
}
addMessage('assistant', data.reply, ttsAudioUrl);
textInput.value = '';
} catch (e) {
showError('发送失败: ' + e.message);
}
}
// TTS 合成
async function synthesizeTTS(text) {
try {
const formData = new FormData();
formData.append('text', text);
const resp = await fetch(`${API_URL}/tts/synthesize`, {
method: 'POST',
body: formData
});
const data = await resp.json();
return data.audio_url;
} catch (e) {
console.error('TTS 合成失败:', e);
return null;
}
}
// 添加消息
function addMessage(role, content, audioData = null) {
const hint = chatSection.querySelector('.hint');
if (hint) hint.remove();
const loading = chatSection.querySelector('.loading');
if (loading) loading.remove();
const msg = document.createElement('div');
msg.className = `message ${role}`;
if (role === 'user' && content instanceof Blob) {
const audioUrl = URL.createObjectURL(content);
msg.innerHTML = `
<div class="role">我</div>
<div class="content audio-content">
<button class="play-btn" onclick="playAudio('${audioUrl}', this)">
<span class="play-icon">▶️</span>
<span class="duration">${audioData}s</span>
</button>
</div>
`;
} else if (role === 'assistant') {
let audioHtml = '';
if (audioData) {
audioHtml = `<button class="play-btn tts-play-btn" onclick="playAudio('${audioData}', this)">
<span class="play-icon">🔊</span>
<span>播放回复</span>
</button>`;
}
msg.innerHTML = `
<div class="role">AI</div>
<div class="content">${content}${audioHtml}</div>
`;
} else {
msg.innerHTML = `<div class="role">我</div><div class="content">${content}</div>`;
}
chatSection.appendChild(msg);
chatSection.scrollTop = chatSection.scrollHeight;
}
// 播放音频
function playAudio(url, btn) {
const audio = new Audio(url);
const icon = btn.querySelector('.play-icon');
audio.onplay = () => {
icon.textContent = '🔊';
btn.classList.add('playing');
};
audio.onended = () => {
icon.textContent = url.startsWith('/audio') ? '🔊' : '▶️';
btn.classList.remove('playing');
};
audio.play();
}
// 显示加载
function showLoading() {
const hint = chatSection.querySelector('.hint');
if (hint) hint.remove();
const loading = document.createElement('div');
loading.className = 'loading';
loading.innerHTML = '<div class="loading-dot"></div><div class="loading-dot"></div><div class="loading-dot"></div>';
chatSection.appendChild(loading);
chatSection.scrollTop = chatSection.scrollHeight;
}
// 显示错误
function showError(message) {
const loading = chatSection.querySelector('.loading');
if (loading) loading.remove();
const error = document.createElement('div');
error.className = 'error-message';
error.textContent = message;
chatSection.appendChild(error);
setTimeout(() => error.remove(), 3000);
}
// 清除对话
async function clearChat() {
if (conversationId) {
await fetch(`${API_URL}/conversation/${conversationId}`, { method: 'DELETE' });
}
conversationId = null;
chatSection.innerHTML = '<div class="hint">开始对话吧！</div>';
}
// 事件绑定
ttsOptions.addEventListener('click', (e) => {
const option = e.target.closest('.tts-option');
if (option) {
selectProvider(option.dataset.provider);
}
});
voiceDropdown.addEventListener('change', () => {
currentVoice = voiceDropdown.value;
saveTTSSettings();
});
recordBtn.addEventListener('click', () => {
isRecording ? stopRecording() : startRecording();
});
sendTextBtn.addEventListener('click', () => sendText(textInput.value));
textInput.addEventListener('keypress', (e) => {
if (e.key === 'Enter') sendText(textInput.value);
});
clearBtn.addEventListener('click', clearChat);
// 启动
init();
</script>
</body>
</html>

223
tts_service.py Normal file
View File

@@ -0,0 +1,223 @@
"""
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_URL = os.getenv("CHATTTS_URL", "http://localhost:19020")
def __init__(self):
self._available = False # 暂不可用
async def synthesize(self, text: str) -> Tuple[str, str]:
"""
使用 ChatTTS 合成语音
TODO: 后续实现
"""
raise NotImplementedError("ChatTTS 尚未实现，请先部署 ChatTTS 服务")
def get_name(self) -> str:
return "ChatTTS"
def is_available(self) -> bool:
"""检查 ChatTTS 是否可用"""
# TODO: 后续实现检测逻辑
return self._available
def set_available(self, available: bool):
"""设置可用状态（部署后调用）"""
self._available = available
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()