import os
|
import uuid
|
import json
|
import hmac
|
import hashlib
|
import base64
|
import time
|
import queue
|
import asyncio
|
import traceback
|
import websockets
|
|
from asyncio import Task
|
from urllib import parse
|
from datetime import datetime
|
from typing import Callable, Any
|
from config.logger import setup_logging
|
from core.utils.tts import MarkdownCleaner
|
from core.providers.tts.base import TTSProviderBase
|
from core.providers.tts.dto.dto import SentenceType, ContentType, InterfaceType
|
|
|
TAG = __name__
|
logger = setup_logging()
|
|
|
class AccessToken:
|
@staticmethod
|
def _encode_text(text):
|
encoded_text = parse.quote_plus(text)
|
return encoded_text.replace("+", "%20").replace("*", "%2A").replace("%7E", "~")
|
|
@staticmethod
|
def _encode_dict(dic):
|
keys = dic.keys()
|
dic_sorted = [(key, dic[key]) for key in sorted(keys)]
|
encoded_text = parse.urlencode(dic_sorted)
|
return encoded_text.replace("+", "%20").replace("*", "%2A").replace("%7E", "~")
|
|
@staticmethod
|
def create_token(access_key_id, access_key_secret, timeout):
|
parameters = {
|
"AccessKeyId": access_key_id,
|
"Action": "CreateToken",
|
"Format": "JSON",
|
"RegionId": "cn-shanghai", # 使用上海地域进行Token获取
|
"SignatureMethod": "HMAC-SHA1",
|
"SignatureNonce": str(uuid.uuid1()),
|
"SignatureVersion": "1.0",
|
"Timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
|
"Version": "2019-02-28",
|
}
|
|
query_string = AccessToken._encode_dict(parameters)
|
string_to_sign = (
|
"GET"
|
+ "&"
|
+ AccessToken._encode_text("/")
|
+ "&"
|
+ AccessToken._encode_text(query_string)
|
)
|
|
secreted_string = hmac.new(
|
bytes(access_key_secret + "&", encoding="utf-8"),
|
bytes(string_to_sign, encoding="utf-8"),
|
hashlib.sha1,
|
).digest()
|
signature = base64.b64encode(secreted_string)
|
signature = AccessToken._encode_text(signature)
|
|
full_url = "http://nls-meta.cn-shanghai.aliyuncs.com/?Signature=%s&%s" % (
|
signature,
|
query_string,
|
)
|
|
import requests
|
|
response = requests.get(full_url, timeout=timeout)
|
if response.ok:
|
root_obj = response.json()
|
key = "Token"
|
if key in root_obj:
|
token = root_obj[key]["Id"]
|
expire_time = root_obj[key]["ExpireTime"]
|
return token, expire_time
|
return None, None
|
|
|
class TTSProvider(TTSProviderBase):
|
TTS_PARAM_CONFIG = [
|
("ttsVolume", "volume", 0, 100, 50, int),
|
("ttsRate", "speech_rate", -500, 500, 0, int),
|
("ttsPitch", "pitch_rate", -500, 500, 0, int),
|
]
|
|
def __init__(self, config, delete_audio_file):
|
super().__init__(config, delete_audio_file)
|
|
# 设置为流式接口类型
|
self.interface_type = InterfaceType.DUAL_STREAM
|
|
# 基础配置
|
self.access_key_id = config.get("access_key_id")
|
self.access_key_secret = config.get("access_key_secret")
|
self.appkey = config.get("appkey")
|
self.format = config.get("format", "pcm")
|
self.report_on_last = True
|
|
# 音色配置 - CosyVoice大模型音色
|
if config.get("private_voice"):
|
self.voice = config.get("private_voice")
|
else:
|
self.voice = config.get("voice", "longxiaochun") # CosyVoice默认音色
|
|
# 音频参数配置
|
volume = config.get("volume", "50")
|
self.volume = int(volume) if volume else 50
|
|
speech_rate = config.get("speech_rate", "0")
|
self.speech_rate = int(speech_rate) if speech_rate else 0
|
|
pitch_rate = config.get("pitch_rate", "0")
|
self.pitch_rate = int(pitch_rate) if pitch_rate else 0
|
|
# 应用百分比调整（如果存在），否则使用公有化配置
|
self._apply_percentage_params(config)
|
|
# WebSocket配置
|
self.host = config.get("host", "nls-gateway-cn-beijing.aliyuncs.com")
|
# 如果配置的是内网地址（包含-internal.aliyuncs.com），则使用ws协议，默认是wss协议
|
if "-internal." in self.host:
|
self.ws_url = f"ws://{self.host}/ws/v1"
|
else:
|
# 默认使用wss协议
|
self.ws_url = f"wss://{self.host}/ws/v1"
|
self.ws = None
|
self._monitor_task = None
|
self.activate_session = False
|
self.last_active_time = None
|
|
# 专属tts设置
|
self.task_id = uuid.uuid4().hex
|
|
# Token管理
|
if self.access_key_id and self.access_key_secret:
|
self._refresh_token()
|
else:
|
self.token = config.get("token")
|
self.expire_time = None
|
|
def _refresh_token(self):
|
"""刷新Token并记录过期时间"""
|
if self.access_key_id and self.access_key_secret:
|
self.token, expire_time_str = AccessToken.create_token(
|
self.access_key_id, self.access_key_secret, self.tts_timeout
|
)
|
if not expire_time_str:
|
raise ValueError("无法获取有效的Token过期时间")
|
|
expire_str = str(expire_time_str).strip()
|
|
try:
|
if expire_str.isdigit():
|
expire_time = datetime.fromtimestamp(int(expire_str))
|
else:
|
expire_time = datetime.strptime(expire_str, "%Y-%m-%dT%H:%M:%SZ")
|
self.expire_time = expire_time.timestamp() - 60
|
except Exception as e:
|
raise ValueError(f"无效的过期时间格式: {expire_str}") from e
|
else:
|
self.expire_time = None
|
|
if not self.token:
|
raise ValueError("无法获取有效的访问Token")
|
|
def _is_token_expired(self):
|
"""检查Token是否过期"""
|
if not self.expire_time:
|
return False
|
return time.time() > self.expire_time
|
|
async def _ensure_connection(self):
|
"""确保WebSocket连接可用"""
|
try:
|
if self._is_token_expired():
|
logger.bind(tag=TAG).warning("Token已过期，正在自动刷新...")
|
self._refresh_token()
|
current_time = time.time()
|
if self.ws and current_time - self.last_active_time < 10:
|
# 10秒内才可以复用链接进行连续对话
|
self.task_id = uuid.uuid4().hex
|
logger.bind(tag=TAG).debug(f"使用已有链接..., task_id: {self.task_id}")
|
return self.ws
|
logger.bind(tag=TAG).debug("开始建立新连接...")
|
|
# 建立新连接前取消旧监听任务
|
await self._cancel_monitor_task()
|
|
self.ws = await websockets.connect(
|
self.ws_url,
|
additional_headers={"X-NLS-Token": self.token},
|
ping_interval=30,
|
ping_timeout=10,
|
close_timeout=10,
|
)
|
self.task_id = uuid.uuid4().hex
|
logger.bind(tag=TAG).debug(f"WebSocket连接建立成功, task_id: {self.task_id}")
|
self.last_active_time = time.time()
|
return self.ws
|
except Exception as e:
|
logger.bind(tag=TAG).error(f"建立连接失败: {str(e)}")
|
self.ws = None
|
self.last_active_time = None
|
raise
|
|
def tts_text_priority_thread(self):
|
"""流式文本处理线程"""
|
while not self.conn.stop_event.is_set():
|
try:
|
message = self.tts_text_queue.get(timeout=1)
|
|
if self.conn.client_abort:
|
logger.bind(tag=TAG).info("收到打断信息，终止TTS文本处理线程")
|
asyncio.run_coroutine_threadsafe(
|
self.finish_session(self.conn.sentence_id),
|
loop=self.conn.loop,
|
)
|
continue
|
|
# 过滤旧消息：检查sentence_id是否匹配
|
if message.sentence_id != self.conn.sentence_id:
|
continue
|
|
logger.bind(tag=TAG).debug(
|
f"收到TTS任务｜{message.sentence_type.name} ｜ {message.content_type.name} | 会话ID: {message.sentence_id}"
|
)
|
|
if message.sentence_type == SentenceType.FIRST:
|
# 重置流式处理状态
|
self.reset_stream_state()
|
# 初始化参数
|
try:
|
logger.bind(tag=TAG).debug("开始启动TTS会话...")
|
future = asyncio.run_coroutine_threadsafe(
|
self.start_session(self.task_id),
|
loop=self.conn.loop,
|
)
|
future.result(timeout=self.tts_timeout)
|
self.before_stop_play_files.clear()
|
logger.bind(tag=TAG).debug("TTS会话启动成功")
|
|
except Exception as e:
|
logger.bind(tag=TAG).error(f"启动TTS会话失败: {str(e)}")
|
continue
|
|
elif ContentType.TEXT == message.content_type:
|
if message.content_detail:
|
try:
|
logger.bind(tag=TAG).debug(
|
f"开始发送TTS文本: {message.content_detail}"
|
)
|
future = asyncio.run_coroutine_threadsafe(
|
self.text_to_speak(message.content_detail, None),
|
loop=self.conn.loop,
|
)
|
future.result(timeout=self.tts_timeout)
|
except Exception as e:
|
logger.bind(tag=TAG).error(f"发送TTS文本失败: {str(e)}")
|
continue
|
|
elif ContentType.FILE == message.content_type:
|
logger.bind(tag=TAG).info(
|
f"添加音频文件到待播放列表: {message.content_file}"
|
)
|
if message.content_file and os.path.exists(message.content_file):
|
# 先处理文件音频数据
|
self._process_audio_file_stream(message.content_file, callback=lambda audio_data: self.handle_audio_file(audio_data, message.content_detail))
|
if message.sentence_type == SentenceType.LAST:
|
try:
|
logger.bind(tag=TAG).debug("开始结束TTS会话...")
|
future = asyncio.run_coroutine_threadsafe(
|
self.finish_session(self.task_id),
|
loop=self.conn.loop,
|
)
|
future.result(timeout=self.tts_timeout)
|
except Exception as e:
|
logger.bind(tag=TAG).error(f"结束TTS会话失败: {str(e)}")
|
continue
|
|
except queue.Empty:
|
continue
|
except Exception as e:
|
logger.bind(tag=TAG).error(
|
f"处理TTS文本失败: {str(e)}, 类型: {type(e).__name__}, 堆栈: {traceback.format_exc()}"
|
)
|
|
async def text_to_speak(self, text, _):
|
try:
|
if self.ws is None:
|
logger.bind(tag=TAG).warning(f"WebSocket连接不存在，终止发送文本")
|
return
|
filtered_text = MarkdownCleaner.clean_markdown(text)
|
|
if filtered_text:
|
# 使用滑动窗口匹配处理跨分片的替换词
|
confirmed_texts, self._pending_prefix = self._match_stream_text(filtered_text)
|
|
# 发送每个确定的文本片段
|
for txt in confirmed_texts:
|
if txt and self.ws:
|
run_request = {
|
"header": {
|
"message_id": uuid.uuid4().hex,
|
"task_id": self.task_id,
|
"namespace": "FlowingSpeechSynthesizer",
|
"name": "RunSynthesis",
|
"appkey": self.appkey,
|
},
|
"payload": {"text": txt},
|
}
|
await self.ws.send(json.dumps(run_request))
|
self.last_active_time = time.time()
|
return
|
|
except Exception as e:
|
logger.bind(tag=TAG).error(f"发送TTS文本失败: {str(e)}")
|
if self.ws:
|
try:
|
await self.ws.close()
|
except:
|
pass
|
self.ws = None
|
raise
|
|
async def start_session(self, task_id):
|
logger.bind(tag=TAG).debug("开始会话～～")
|
try:
|
# 上个会话处于激活状态时关闭上个连接新建链接
|
if self.activate_session:
|
await self.close()
|
|
# 设置会话激活标志
|
self.activate_session = True
|
|
# 建立新连接
|
await self._ensure_connection()
|
|
# 启动监听任务
|
if self._monitor_task is None or self._monitor_task.done():
|
logger.bind(tag=TAG).debug("启动监听任务...")
|
self._monitor_task = asyncio.create_task(self._start_monitor_tts_response())
|
|
start_request = {
|
"header": {
|
"message_id": uuid.uuid4().hex,
|
"task_id": self.task_id,
|
"namespace": "FlowingSpeechSynthesizer",
|
"name": "StartSynthesis",
|
"appkey": self.appkey,
|
},
|
"payload": {
|
"voice": self.voice,
|
"format": self.format,
|
"sample_rate": self.conn.sample_rate,
|
"volume": self.volume,
|
"speech_rate": self.speech_rate,
|
"pitch_rate": self.pitch_rate,
|
"enable_subtitle": True,
|
},
|
}
|
await self.ws.send(json.dumps(start_request))
|
self.last_active_time = time.time()
|
logger.bind(tag=TAG).debug("会话启动请求已发送")
|
except Exception as e:
|
logger.bind(tag=TAG).error(f"启动会话失败: {str(e)}")
|
# 确保清理资源
|
await self.close()
|
raise
|
|
async def finish_session(self, task_id):
|
logger.bind(tag=TAG).debug(f"关闭会话～～{task_id}")
|
try:
|
if self.ws:
|
stop_request = {
|
"header": {
|
"message_id": uuid.uuid4().hex,
|
"task_id": self.task_id,
|
"namespace": "FlowingSpeechSynthesizer",
|
"name": "StopSynthesis",
|
"appkey": self.appkey,
|
}
|
}
|
await self.ws.send(json.dumps(stop_request))
|
logger.bind(tag=TAG).debug("会话结束请求已发送")
|
self.last_active_time = time.time()
|
|
except Exception as e:
|
logger.bind(tag=TAG).error(f"关闭会话失败: {str(e)}")
|
# 确保清理资源
|
await self.close()
|
raise
|
|
async def close(self):
|
"""资源清理"""
|
await super().close()
|
self.activate_session = False
|
await self._cancel_monitor_task()
|
|
if self.ws:
|
try:
|
await self.ws.close()
|
except:
|
pass
|
self.ws = None
|
self.last_active_time = None
|
|
async def _start_monitor_tts_response(self):
|
"""监听TTS响应 - 长期运行"""
|
try:
|
while not self.conn.stop_event.is_set():
|
try:
|
msg = await self.ws.recv()
|
self.last_active_time = time.time()
|
|
if isinstance(msg, str): # 文本控制消息
|
try:
|
data = json.loads(msg)
|
header = data.get("header", {})
|
event_name = header.get("name")
|
task_id = header.get("task_id")
|
|
# 只处理当前活跃会话的响应
|
if task_id and self.task_id != task_id:
|
if event_name in ["SynthesisCompleted", "TaskFailed"]:
|
logger.bind(tag=TAG).debug(f"收到残余下行结束响应重置会话状态～～")
|
self.activate_session = False
|
continue
|
|
if event_name == "SynthesisStarted":
|
logger.bind(tag=TAG).debug("TTS合成已启动")
|
self.tts_audio_queue.put(
|
(SentenceType.FIRST, [], None)
|
)
|
elif event_name == "SentenceEnd":
|
# 发送缓存的数据
|
tts_text = self.get_tts_text(self.conn.sentence_id)
|
if tts_text:
|
logger.bind(tag=TAG).info(
|
f"句子语音生成成功： {tts_text}"
|
)
|
self.tts_audio_queue.put(
|
(SentenceType.FIRST, [], tts_text)
|
)
|
self.clear_tts_text(self.conn.sentence_id)
|
elif event_name == "SynthesisCompleted":
|
logger.bind(tag=TAG).debug(f"会话结束～～")
|
self.activate_session = False
|
self._process_before_stop_play_files()
|
except json.JSONDecodeError:
|
logger.bind(tag=TAG).warning("收到无效的JSON消息")
|
# 二进制消息（音频数据）
|
elif isinstance(msg, (bytes, bytearray)):
|
self.opus_encoder.encode_pcm_to_opus_stream(msg, False, self.handle_opus)
|
except websockets.ConnectionClosed:
|
logger.bind(tag=TAG).warning("WebSocket连接已关闭")
|
break
|
except Exception as e:
|
logger.bind(tag=TAG).error(
|
f"处理TTS响应时出错: {e}\n{traceback.format_exc()}"
|
)
|
break
|
# 连接异常时关闭WebSocket
|
if self.ws:
|
try:
|
await self.ws.close()
|
except:
|
pass
|
self.ws = None
|
# 监听任务退出时清理引用
|
finally:
|
self.activate_session = False
|
self._monitor_task = None
|
|
def audio_to_opus_data_stream(
|
self, audio_file_path, callback: Callable[[Any], Any] = None
|
):
|
"""重写父类方法：使用独立的临时编码器处理音频文件，避免与TTS流式编码器并发冲突。
|
双流式TTS中，monitor任务在event loop线程接收TTS音频并使用self.opus_encoder编码，
|
同时tts_text_priority_thread处理音乐文件也使用self.opus_encoder，
|
共享的encoder.buffer非线程安全，并发访问会导致SILK resampler断言失败。
|
"""
|
from core.utils.util import audio_to_data_stream
|
|
return audio_to_data_stream(
|
audio_file_path,
|
is_opus=True,
|
callback=callback,
|
sample_rate=self.conn.sample_rate,
|
opus_encoder=None,
|
)
|
|
async def _cancel_monitor_task(self):
|
"""取消监听任务"""
|
if self._monitor_task and not self._monitor_task.done():
|
self._monitor_task.cancel()
|
try:
|
await self._monitor_task
|
except asyncio.CancelledError:
|
pass
|
except Exception as e:
|
logger.bind(tag=TAG).warning(f"取消监听任务错误: {e}")
|
self._monitor_task = None
|
|
def to_tts(self, text: str) -> list:
|
"""非流式TTS处理，用于测试及保存音频文件的场景"""
|
try:
|
# 创建新的事件循环
|
loop = asyncio.new_event_loop()
|
asyncio.set_event_loop(loop)
|
|
# 存储音频数据
|
audio_data = []
|
|
async def _generate_audio():
|
# 刷新Token（如果需要）
|
if self._is_token_expired():
|
self._refresh_token()
|
|
# 建立WebSocket连接
|
ws = await websockets.connect(
|
self.ws_url,
|
additional_headers={"X-NLS-Token": self.token},
|
ping_interval=30,
|
ping_timeout=10,
|
close_timeout=10,
|
)
|
try:
|
# 发送StartSynthesis请求
|
start_request = {
|
"header": {
|
"message_id": uuid.uuid4().hex,
|
"task_id": self.task_id,
|
"namespace": "FlowingSpeechSynthesizer",
|
"name": "StartSynthesis",
|
"appkey": self.appkey,
|
},
|
"payload": {
|
"voice": self.voice,
|
"format": self.format,
|
"sample_rate": self.conn.sample_rate,
|
"volume": self.volume,
|
"speech_rate": self.speech_rate,
|
"pitch_rate": self.pitch_rate,
|
"enable_subtitle": True,
|
},
|
}
|
await ws.send(json.dumps(start_request))
|
|
# 等待SynthesisStarted响应
|
synthesis_started = False
|
while not synthesis_started:
|
msg = await ws.recv()
|
if isinstance(msg, str):
|
data = json.loads(msg)
|
header = data.get("header", {})
|
if header.get("name") == "SynthesisStarted":
|
synthesis_started = True
|
logger.bind(tag=TAG).debug("TTS合成已启动")
|
elif header.get("name") == "TaskFailed":
|
error_info = data.get("payload", {}).get(
|
"error_info", {}
|
)
|
error_code = error_info.get("error_code")
|
error_message = error_info.get(
|
"error_message", "未知错误"
|
)
|
raise Exception(
|
f"启动合成失败: {error_code} - {error_message}"
|
)
|
|
# 发送文本合成请求
|
filtered_text = MarkdownCleaner.clean_markdown(text)
|
if self._correct_words_pattern:
|
filtered_text = self._correct_words_pattern.sub(lambda m: self.correct_words[m.group(0)], filtered_text)
|
run_request = {
|
"header": {
|
"message_id": uuid.uuid4().hex,
|
"task_id": self.task_id,
|
"namespace": "FlowingSpeechSynthesizer",
|
"name": "RunSynthesis",
|
"appkey": self.appkey,
|
},
|
"payload": {"text": filtered_text},
|
}
|
await ws.send(json.dumps(run_request))
|
|
# 发送停止合成请求
|
stop_request = {
|
"header": {
|
"message_id": uuid.uuid4().hex,
|
"task_id": self.task_id,
|
"namespace": "FlowingSpeechSynthesizer",
|
"name": "StopSynthesis",
|
"appkey": self.appkey,
|
}
|
}
|
await ws.send(json.dumps(stop_request))
|
|
# 接收音频数据
|
synthesis_completed = False
|
while not synthesis_completed:
|
msg = await ws.recv()
|
if isinstance(msg, (bytes, bytearray)):
|
self.opus_encoder.encode_pcm_to_opus_stream(
|
msg,
|
end_of_stream=False,
|
callback=lambda opus: audio_data.append(opus)
|
)
|
elif isinstance(msg, str):
|
data = json.loads(msg)
|
header = data.get("header", {})
|
event_name = header.get("name")
|
if event_name == "SynthesisCompleted":
|
synthesis_completed = True
|
logger.bind(tag=TAG).debug("TTS合成完成")
|
elif event_name == "TaskFailed":
|
error_info = data.get("payload", {}).get(
|
"error_info", {}
|
)
|
error_code = error_info.get("error_code")
|
error_message = error_info.get(
|
"error_message", "未知错误"
|
)
|
raise Exception(
|
f"合成失败: {error_code} - {error_message}"
|
)
|
finally:
|
try:
|
await ws.close()
|
except:
|
pass
|
|
loop.run_until_complete(_generate_audio())
|
loop.close()
|
|
return audio_data
|
except Exception as e:
|
logger.bind(tag=TAG).error(f"生成音频数据失败: {str(e)}")
|
return []
|
