import os
|
import re
|
import math
|
import uuid
|
import queue
|
import asyncio
|
import threading
|
import traceback
|
import concurrent.futures
|
|
from core.utils import p3
|
from datetime import datetime
|
from core.utils import textUtils
|
from typing import Callable, Any
|
from abc import ABC, abstractmethod
|
from config.logger import setup_logging
|
from core.utils import opus_encoder_utils
|
from core.utils.tts import MarkdownCleaner, convert_percentage_to_range
|
from core.utils.output_counter import add_device_output
|
from core.handle.reportHandle import enqueue_tts_report
|
from core.handle.sendAudioHandle import sendAudioMessage
|
from core.utils.util import audio_bytes_to_data_stream, audio_to_data_stream
|
from core.providers.tts.dto.dto import (
|
TTSMessageDTO,
|
SentenceType,
|
ContentType,
|
InterfaceType,
|
)
|
|
TAG = __name__
|
logger = setup_logging()
|
|
|
class TTSProviderBase(ABC):
|
def __init__(self, config, delete_audio_file):
|
self.interface_type = InterfaceType.NON_STREAM
|
self.conn = None
|
self.delete_audio_file = delete_audio_file
|
self.audio_file_type = "wav"
|
self.output_file = config.get("output_dir", "tmp/")
|
self.tts_timeout = float(config.get("tts_timeout", 15))
|
if not math.isfinite(self.tts_timeout) or self.tts_timeout <= 0:
|
raise ValueError("tts_timeout must be a positive finite number")
|
self.tts_text_queue = queue.Queue()
|
self.tts_audio_queue = queue.Queue()
|
self.tts_audio_first_sentence = True
|
self.before_stop_play_files = []
|
self.report_on_last = False
|
# sentence_id 到文本的映射，用于流式TTS获取正确的字幕文本
|
self._sentence_text_map = {}
|
# 加载替换词，用于一次性正则替换
|
raw_words = config.get("correct_words", [])
|
self.correct_words = {}
|
for item in raw_words:
|
parts = item.split("|", 1)
|
if len(parts) == 2:
|
self.correct_words[parts[0]] = parts[1]
|
# 构建正则表达式，使用最长匹配优先（排序后转义拼接）
|
if self.correct_words:
|
# 按key长度降序排列，长的先匹配，避免短词部分干扰
|
sorted_keys = sorted(self.correct_words.keys(), key=len, reverse=True)
|
pattern_str = "|".join(re.escape(k) for k in sorted_keys)
|
self._correct_words_pattern = re.compile(pattern_str)
|
# 构建反向替换正则，用于将TTS服务返回的替换后文本还原为原始文本（字幕显示）
|
reverse_map = {v: k for k, v in self.correct_words.items()}
|
sorted_reverse_keys = sorted(reverse_map.keys(), key=len, reverse=True)
|
reverse_pattern_str = "|".join(re.escape(k) for k in sorted_reverse_keys)
|
self._reverse_words_pattern = re.compile(reverse_pattern_str)
|
self._reverse_words_map = reverse_map
|
# 流式滑动窗口：按首字分组的替换词字典，用于快速查找
|
self._words_by_first_char = {}
|
for key in sorted_keys: # 使用已按长度降序排列的keys，确保长词优先匹配
|
first_char = key[0] if key else ""
|
if first_char not in self._words_by_first_char:
|
self._words_by_first_char[first_char] = []
|
self._words_by_first_char[first_char].append(key)
|
else:
|
self._correct_words_pattern = None
|
self._reverse_words_pattern = None
|
self._reverse_words_map = None
|
|
# 流式滑动窗口：待匹配的缓存文本
|
self._pending_prefix = ""
|
self.tts_text_buff = []
|
self.punctuations = (
|
"。",
|
"？",
|
"?",
|
"！",
|
"!",
|
"；",
|
";",
|
"：",
|
)
|
self.first_sentence_punctuations = (
|
"，",
|
"~",
|
"、",
|
",",
|
"。",
|
"？",
|
"?",
|
"！",
|
"!",
|
"；",
|
";",
|
"：",
|
)
|
self.tts_stop_request = False
|
self.processed_chars = 0
|
self.is_first_sentence = True
|
|
def generate_filename(self, extension=".wav"):
|
return os.path.join(
|
self.output_file,
|
f"tts-{datetime.now().date()}@{uuid.uuid4().hex}{extension}",
|
)
|
|
def handle_opus(self, opus_data: bytes):
|
self.last_stream_had_audio = True
|
if not getattr(self, "_logged_first_opus", False):
|
self._logged_first_opus = True
|
logger.bind(tag=TAG).info(f"首帧opus已入队 bytes={len(opus_data)}")
|
logger.bind(tag=TAG).debug(f"推送数据到队列里面帧数～～ {len(opus_data)}")
|
self.tts_audio_queue.put((SentenceType.MIDDLE, opus_data, None, getattr(self, 'current_sentence_id', None)))
|
|
def _run_text_to_speak(self, text, output_file=None):
|
"""Generate TTS with a hard timeout so LAST/stop is not blocked forever.
|
|
Edge TTS (and some HTTP TTS APIs) can hang past asyncio.wait_for cancel.
|
Run synthesis in a daemon thread and abandon it if the join times out.
|
"""
|
result = {"data": None, "error": None}
|
|
def worker():
|
try:
|
result["data"] = asyncio.run(self.text_to_speak(text, output_file))
|
except Exception as e:
|
result["error"] = e
|
|
worker_thread = threading.Thread(target=worker, name="tts-synth", daemon=True)
|
worker_thread.start()
|
worker_thread.join(timeout=self.tts_timeout)
|
if worker_thread.is_alive():
|
logger.bind(tag=TAG).error(
|
f"TTS timed out after {self.tts_timeout}s: {text}"
|
)
|
raise TimeoutError(f"TTS timed out after {self.tts_timeout}s")
|
if result["error"] is not None:
|
raise result["error"]
|
return result["data"]
|
|
def handle_audio_file(self, file_audio: bytes, text):
|
self.before_stop_play_files.append((file_audio, text))
|
|
def to_tts_stream(self, text, opus_handler: Callable[[bytes], None] = None) -> None:
|
# 保留原始文本用于显示/上报
|
original_text = text
|
text = MarkdownCleaner.clean_markdown(text)
|
# 使用正则一次性替换，避免重复遍历和部分匹配问题
|
if self._correct_words_pattern:
|
text = self._correct_words_pattern.sub(lambda m: self.correct_words[m.group(0)], text)
|
# 先合成 opus，再下发 sentence_start。先发字幕再合成时，设备会进入说话中却收不到音频。
|
max_repeat_time = 2
|
opus_frames = {"n": 0}
|
pending_opus = []
|
self.last_stream_had_audio = False
|
|
def _count_opus(data):
|
opus_frames["n"] += 1
|
pending_opus.append(bytes(data) if not isinstance(data, bytes) else data)
|
|
def _flush_ready_audio():
|
sid = getattr(self, "current_sentence_id", None)
|
self.tts_audio_queue.put((SentenceType.FIRST, None, original_text, sid))
|
for frame in pending_opus:
|
if opus_handler:
|
opus_handler(frame)
|
else:
|
self.tts_audio_queue.put((SentenceType.MIDDLE, frame, None, sid))
|
self.last_stream_had_audio = True
|
pending_opus.clear()
|
|
counted_handler = _count_opus
|
if self.delete_audio_file:
|
# 需要删除文件的直接转为音频数据
|
audio_bytes = None
|
while max_repeat_time > 0:
|
try:
|
audio_bytes = self._run_text_to_speak(text, None)
|
if audio_bytes:
|
encode_rate = getattr(self.opus_encoder, "sample_rate", None) or self.conn.sample_rate
|
audio_bytes_to_data_stream(
|
audio_bytes,
|
file_type=self.audio_file_type,
|
is_opus=True,
|
callback=counted_handler,
|
sample_rate=encode_rate,
|
opus_encoder=self.opus_encoder,
|
)
|
break
|
else:
|
max_repeat_time -= 1
|
except Exception as e:
|
logger.bind(tag=TAG).warning(
|
f"语音生成失败{5 - max_repeat_time + 1}次: {original_text}，错误: {e}"
|
)
|
max_repeat_time -= 1
|
if max_repeat_time > 0 and opus_frames["n"] > 0:
|
logger.bind(tag=TAG).info(
|
f"语音生成成功: {original_text}，opus帧={opus_frames['n']}，重试{5 - max_repeat_time}次"
|
)
|
_flush_ready_audio()
|
elif max_repeat_time > 0:
|
logger.bind(tag=TAG).error(
|
f"语音生成无opus帧: {original_text}，bytes={0 if not audio_bytes else len(audio_bytes)}"
|
)
|
else:
|
logger.bind(tag=TAG).error(
|
f"语音生成失败: {original_text}，请检查网络或服务是否正常"
|
)
|
return None
|
else:
|
tmp_file = self.generate_filename()
|
try:
|
while not os.path.exists(tmp_file) and max_repeat_time > 0:
|
try:
|
self._run_text_to_speak(text, tmp_file)
|
except Exception as e:
|
logger.bind(tag=TAG).warning(
|
f"语音生成失败{5 - max_repeat_time + 1}次: {original_text}，错误: {e}"
|
)
|
# 未执行成功，删除文件
|
if os.path.exists(tmp_file):
|
os.remove(tmp_file)
|
max_repeat_time -= 1
|
|
if max_repeat_time > 0:
|
logger.bind(tag=TAG).info(
|
f"语音生成成功: {original_text}:{tmp_file}，重试{5 - max_repeat_time}次"
|
)
|
self._process_audio_file_stream(tmp_file, callback=counted_handler)
|
if opus_frames["n"] > 0:
|
_flush_ready_audio()
|
else:
|
logger.bind(tag=TAG).error(
|
f"语音生成无opus帧: {original_text}:{tmp_file}"
|
)
|
else:
|
logger.bind(tag=TAG).error(
|
f"语音生成失败: {original_text}，请检查网络或服务是否正常"
|
)
|
except Exception as e:
|
logger.bind(tag=TAG).error(f"Failed to generate TTS file: {e}")
|
return None
|
|
def to_tts(self, text):
|
# 保留原始文本用于日志/显示
|
original_text = text
|
text = MarkdownCleaner.clean_markdown(text)
|
if self._correct_words_pattern:
|
text = self._correct_words_pattern.sub(lambda m: self.correct_words[m.group(0)], text)
|
max_repeat_time = 5
|
if self.delete_audio_file:
|
# 需要删除文件的直接转为音频数据
|
while max_repeat_time > 0:
|
try:
|
audio_bytes = self._run_text_to_speak(text, None)
|
if audio_bytes:
|
audio_datas = []
|
audio_bytes_to_data_stream(
|
audio_bytes,
|
file_type=self.audio_file_type,
|
is_opus=True,
|
callback=lambda data: audio_datas.append(data),
|
sample_rate=self.conn.sample_rate,
|
)
|
return audio_datas
|
else:
|
max_repeat_time -= 1
|
except Exception as e:
|
logger.bind(tag=TAG).warning(
|
f"语音生成失败{5 - max_repeat_time + 1}次: {original_text}，错误: {e}"
|
)
|
max_repeat_time -= 1
|
if max_repeat_time > 0:
|
logger.bind(tag=TAG).info(
|
f"语音生成成功: {original_text}，重试{5 - max_repeat_time}次"
|
)
|
else:
|
logger.bind(tag=TAG).error(
|
f"语音生成失败: {original_text}，请检查网络或服务是否正常"
|
)
|
return None
|
else:
|
tmp_file = self.generate_filename()
|
try:
|
while not os.path.exists(tmp_file) and max_repeat_time > 0:
|
try:
|
self._run_text_to_speak(text, tmp_file)
|
except Exception as e:
|
logger.bind(tag=TAG).warning(
|
f"语音生成失败{5 - max_repeat_time + 1}次: {original_text}，错误: {e}"
|
)
|
# 未执行成功，删除文件
|
if os.path.exists(tmp_file):
|
os.remove(tmp_file)
|
max_repeat_time -= 1
|
|
if max_repeat_time > 0:
|
logger.bind(tag=TAG).info(
|
f"语音生成成功: {original_text}:{tmp_file}，重试{5 - max_repeat_time}次"
|
)
|
else:
|
logger.bind(tag=TAG).error(
|
f"语音生成失败: {original_text}，请检查网络或服务是否正常"
|
)
|
|
return tmp_file
|
except Exception as e:
|
logger.bind(tag=TAG).error(f"Failed to generate TTS file: {e}")
|
return None
|
|
@abstractmethod
|
async def text_to_speak(self, text, output_file):
|
pass
|
|
def audio_to_pcm_data_stream(
|
self, audio_file_path, callback: Callable[[Any], Any] = None
|
):
|
"""音频文件转换为PCM编码"""
|
return audio_to_data_stream(audio_file_path, is_opus=False, callback=callback, sample_rate=self.conn.sample_rate, opus_encoder=None)
|
|
def audio_to_opus_data_stream(
|
self, audio_file_path, callback: Callable[[Any], Any] = None
|
):
|
"""音频文件转换为Opus编码"""
|
return audio_to_data_stream(audio_file_path, is_opus=True, callback=callback, sample_rate=self.conn.sample_rate, opus_encoder=self.opus_encoder)
|
|
def tts_one_sentence(
|
self,
|
conn,
|
content_type,
|
content_detail=None,
|
content_file=None,
|
sentence_id=None,
|
):
|
"""发送一句话"""
|
if not sentence_id:
|
if conn.sentence_id:
|
sentence_id = conn.sentence_id
|
else:
|
sentence_id = str(uuid.uuid4().hex)
|
conn.sentence_id = sentence_id
|
# 对于单句的文本，进行分段处理
|
segments = re.split(r"([。！？!?；;\n])", content_detail)
|
for seg in segments:
|
self.tts_text_queue.put(
|
TTSMessageDTO(
|
sentence_id=sentence_id,
|
sentence_type=SentenceType.MIDDLE,
|
content_type=content_type,
|
content_detail=seg,
|
content_file=content_file,
|
)
|
)
|
|
async def open_audio_channels(self, conn):
|
self.conn = conn
|
if not hasattr(self, "_open_lock"):
|
self._open_lock = threading.Lock()
|
|
with self._open_lock:
|
encoder = getattr(self, "opus_encoder", None)
|
encoder_rate = getattr(encoder, "sample_rate", None) if encoder is not None else None
|
if encoder is None or encoder_rate != conn.sample_rate:
|
self.opus_encoder = opus_encoder_utils.OpusEncoderUtils(
|
sample_rate=conn.sample_rate, channels=1, frame_size_ms=60
|
)
|
|
if getattr(self, "_channels_opened", False):
|
play = getattr(self, "audio_play_priority_thread", None)
|
text_th = getattr(self, "tts_priority_thread", None)
|
if play and play.is_alive() and text_th and text_th.is_alive():
|
return
|
|
self.tts_priority_thread = threading.Thread(
|
target=self.tts_text_priority_thread, daemon=True
|
)
|
self.tts_priority_thread.start()
|
self.audio_play_priority_thread = threading.Thread(
|
target=self._audio_play_priority_thread, daemon=True
|
)
|
self.audio_play_priority_thread.start()
|
self._channels_opened = True
|
|
def store_tts_text(self, sentence_id, text):
|
"""存储指定 sentence_id 对应的文本，用于流式TTS获取正确的字幕文本
|
|
Args:
|
sentence_id: 会话ID
|
text: 要存储的文本
|
"""
|
if sentence_id and text:
|
self._sentence_text_map[sentence_id] = text
|
# 只保留最近 5 个，防止内存泄漏
|
if len(self._sentence_text_map) > 5:
|
oldest = next(iter(self._sentence_text_map))
|
del self._sentence_text_map[oldest]
|
|
def get_tts_text(self, sentence_id):
|
"""获取指定 sentence_id 对应的文本
|
|
Args:
|
sentence_id: 会话ID
|
|
Returns:
|
str: 对应的文本，如果不存在返回 None
|
"""
|
return self._sentence_text_map.get(sentence_id)
|
|
def clear_tts_text(self, sentence_id):
|
"""清除指定 sentence_id 的文本
|
|
Args:
|
sentence_id: 会话ID
|
"""
|
if sentence_id in self._sentence_text_map:
|
del self._sentence_text_map[sentence_id]
|
|
def _restore_original_text(self, text):
|
if not self._reverse_words_pattern or not text:
|
return text
|
return self._reverse_words_pattern.sub(
|
lambda m: self._reverse_words_map[m.group(0)], text
|
)
|
|
# 这里默认是非流式的处理方式
|
# 流式处理方式请在子类中重写
|
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文本处理线程")
|
continue
|
# 过滤旧消息：检查sentence_id是否匹配
|
if message.sentence_id != self.conn.sentence_id:
|
continue
|
if message.sentence_type == SentenceType.FIRST:
|
self.current_sentence_id = message.sentence_id
|
self.tts_stop_request = False
|
self.processed_chars = 0
|
self.tts_text_buff = []
|
self.is_first_sentence = True
|
self.tts_audio_first_sentence = True
|
self._logged_first_opus = False
|
self.last_stream_had_audio = False
|
elif ContentType.TEXT == message.content_type:
|
self.tts_text_buff.append(message.content_detail)
|
segment_text = self._get_segment_text()
|
if segment_text:
|
try:
|
self.to_tts_stream(segment_text, opus_handler=self.handle_opus)
|
except Exception as e:
|
logger.bind(tag=TAG).error(f"TTS分段合成失败: {e}")
|
elif ContentType.FILE == message.content_type:
|
self._process_remaining_text_stream(opus_handler=self.handle_opus)
|
tts_file = message.content_file
|
if tts_file and os.path.exists(tts_file):
|
self._process_audio_file_stream(
|
tts_file, callback=self.handle_opus
|
)
|
if message.sentence_type == SentenceType.LAST:
|
try:
|
self._process_remaining_text_stream(opus_handler=self.handle_opus)
|
except Exception as e:
|
logger.bind(tag=TAG).error(
|
f"处理剩余TTS文本失败: {e}",
|
)
|
if getattr(self, "last_stream_had_audio", False):
|
self.tts_audio_queue.put(
|
(message.sentence_type, [], message.content_detail, message.sentence_id)
|
)
|
else:
|
logger.bind(tag=TAG).error("跳过TTS stop：本句没有opus音频")
|
|
except queue.Empty:
|
continue
|
except Exception as e:
|
logger.bind(tag=TAG).error(
|
f"处理TTS文本失败: {str(e)}, 类型: {type(e).__name__}, 堆栈: {traceback.format_exc()}"
|
)
|
continue
|
|
def _audio_play_priority_thread(self):
|
# 需要上报的文本和音频列表
|
enqueue_text = None
|
enqueue_audio = []
|
while not self.conn.stop_event.is_set():
|
text = None
|
try:
|
try:
|
item = self.tts_audio_queue.get(timeout=0.1)
|
if len(item) == 4:
|
sentence_type, audio_datas, text, sentence_id = item
|
else:
|
sentence_type, audio_datas, text = item
|
sentence_id = None
|
except queue.Empty:
|
if self.conn.stop_event.is_set():
|
break
|
continue
|
|
if self.conn.client_abort:
|
logger.bind(tag=TAG).debug("收到打断信号，跳过当前音频数据")
|
enqueue_text, enqueue_audio = None, []
|
continue
|
|
# 收到下一个文本开始或会话结束时进行上报
|
if sentence_type is not SentenceType.MIDDLE:
|
if self.report_on_last:
|
# 累积模式：适用于全程只有一个语音流的TTS（如seed-tts-2.0）
|
# FIRST时只记录文本，音频持续累积，仅在LAST时统一上报
|
if text:
|
enqueue_text = text
|
if sentence_type == SentenceType.LAST:
|
enqueue_tts_report(self.conn, enqueue_text, enqueue_audio)
|
enqueue_audio = []
|
enqueue_text = None
|
else:
|
# 非累积模式：每个句子分别上报
|
if enqueue_text is not None:
|
enqueue_tts_report(self.conn, enqueue_text, enqueue_audio)
|
enqueue_audio = []
|
enqueue_text = text
|
|
# 收集上报音频数据
|
if isinstance(audio_datas, bytes):
|
enqueue_audio.append(audio_datas)
|
|
# 发送音频
|
future = asyncio.run_coroutine_threadsafe(
|
sendAudioMessage(self.conn, sentence_type, audio_datas, text, sentence_id),
|
self.conn.loop,
|
)
|
future.result(timeout=20)
|
|
# 记录输出和报告
|
if self.conn.max_output_size > 0 and text:
|
add_device_output(self.conn.headers.get("device-id"), len(text))
|
|
except Exception as e:
|
logger.bind(tag=TAG).error(f"audio_play_priority_thread: {text} {e}")
|
|
async def start_session(self, session_id):
|
pass
|
|
async def finish_session(self, session_id):
|
pass
|
|
async def close(self):
|
"""资源清理方法"""
|
self._sentence_text_map.clear()
|
if hasattr(self, "ws") and self.ws:
|
await self.ws.close()
|
|
def _get_segment_text(self):
|
# 合并当前全部文本并处理未分割部分
|
full_text = "".join(self.tts_text_buff)
|
current_text = full_text[self.processed_chars :] # 从未处理的位置开始
|
last_punct_pos = -1
|
|
# 根据是否是第一句话选择不同的标点符号集合
|
punctuations_to_use = (
|
self.first_sentence_punctuations
|
if self.is_first_sentence
|
else self.punctuations
|
)
|
|
for punct in punctuations_to_use:
|
pos = current_text.rfind(punct)
|
if (pos != -1 and last_punct_pos == -1) or (
|
pos != -1 and pos < last_punct_pos
|
):
|
last_punct_pos = pos
|
|
if last_punct_pos != -1:
|
segment_text_raw = current_text[: last_punct_pos + 1]
|
segment_text = textUtils.get_string_no_punctuation_or_emoji(
|
segment_text_raw
|
)
|
self.processed_chars += len(segment_text_raw) # 更新已处理字符位置
|
|
# 如果是第一句话，在找到第一个逗号后，将标志设置为False
|
if self.is_first_sentence:
|
self.is_first_sentence = False
|
|
return segment_text
|
elif self.tts_stop_request and current_text:
|
segment_text = current_text
|
self.is_first_sentence = True # 重置标志
|
return segment_text
|
else:
|
return None
|
|
def _process_audio_file_stream(
|
self, tts_file, callback: Callable[[Any], Any]
|
) -> None:
|
"""处理音频文件并转换为指定格式
|
|
Args:
|
tts_file: 音频文件路径
|
callback: 文件处理函数
|
"""
|
if tts_file.endswith(".p3"):
|
p3.decode_opus_from_file_stream(tts_file, callback=callback)
|
elif self.conn.audio_format == "pcm":
|
self.audio_to_pcm_data_stream(tts_file, callback=callback)
|
else:
|
self.audio_to_opus_data_stream(tts_file, callback=callback)
|
|
if (
|
self.delete_audio_file
|
and tts_file is not None
|
and os.path.exists(tts_file)
|
and tts_file.startswith(self.output_file)
|
):
|
os.remove(tts_file)
|
|
def _process_before_stop_play_files(self):
|
for audio_datas, text in self.before_stop_play_files:
|
self.tts_audio_queue.put((SentenceType.MIDDLE, audio_datas, text, getattr(self, 'current_sentence_id', None)))
|
self.before_stop_play_files.clear()
|
self.tts_audio_queue.put((SentenceType.LAST, [], None, getattr(self, 'current_sentence_id', None)))
|
|
def _process_remaining_text_stream(
|
self, opus_handler: Callable[[bytes], None] = None
|
):
|
"""处理剩余的文本并生成语音
|
|
Returns:
|
bool: 是否成功处理了文本
|
"""
|
full_text = "".join(self.tts_text_buff)
|
remaining_text = full_text[self.processed_chars :]
|
if remaining_text:
|
segment_text = textUtils.get_string_no_punctuation_or_emoji(remaining_text)
|
if segment_text:
|
self.to_tts_stream(segment_text, opus_handler=opus_handler)
|
self.processed_chars += len(full_text)
|
return True
|
return False
|
|
def _apply_percentage_params(self, config):
|
"""根据子类定义的 TTS_PARAM_CONFIG 批量应用百分比参数"""
|
for config_key, attr_name, min_val, max_val, base_val, transform in self.TTS_PARAM_CONFIG:
|
if config_key in config:
|
val = convert_percentage_to_range(config[config_key], min_val, max_val, base_val)
|
setattr(self, attr_name, transform(val) if transform else val)
|
|
def _match_stream_text(self, text):
|
"""流式文本滑动窗口匹配，用于处理跨分片的替换词
|
|
Args:
|
text: 输入的文本片段
|
|
Returns:
|
tuple: (确定的文本列表, 剩余待匹配的前缀)
|
"""
|
if not self.correct_words or not text:
|
return [text] if text else [], ""
|
|
result = []
|
pending = self._pending_prefix
|
i = 0
|
|
while i < len(text):
|
char = text[i]
|
|
# 尝试：pending + 当前字符 是否能匹配替换词
|
test_text = pending + char
|
|
matched = False
|
# 遍历可能匹配的替换词
|
candidates = self._words_by_first_char.get(pending[0], []) if pending else self._words_by_first_char.get(char, [])
|
for key in candidates:
|
if test_text == key:
|
# 完整匹配，替换后发送
|
result.append(self.correct_words[key])
|
pending = ""
|
matched = True
|
break
|
elif key.startswith(test_text):
|
# 是替换词的前缀，继续等待
|
pending = test_text
|
matched = True
|
break
|
|
if matched:
|
i += 1
|
continue
|
|
# 没有匹配到更长的词，pending 的内容确定可以发送
|
if pending:
|
result.append(pending)
|
pending = ""
|
|
# 检查当前字符是否是某个替换词的开头
|
if char in self._words_by_first_char:
|
pending = char
|
else:
|
result.append(char)
|
|
i += 1
|
|
return result, pending
|
|
def reset_stream_state(self):
|
"""重置流式处理状态，用于会话开始时清理残留状态"""
|
self._pending_prefix = ""
|
