update：格式化代码

2026-07-28 21:13:51 +08:00 · 2025-04-01 14:27:17 +08:00
parent d9d973b6aa 7c389a0641
commit 61db1192e2
7 changed files with 507 additions and 222 deletions
@@ -24,52 +24,65 @@ class AccessToken:
    @staticmethod
    def _encode_text(text):
        encoded_text = parse.quote_plus(text)
-        return encoded_text.replace('+', '%20').replace('*', '%2A').replace('%7E', '~')
-    
+        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', '~')
-    
+        return encoded_text.replace("+", "%20").replace("*", "%2A").replace("%7E", "~")
+
    @staticmethod
    def create_token(access_key_id, access_key_secret):
-        parameters = {'AccessKeyId': access_key_id,
-                      'Action': 'CreateToken',
-                      'Format': 'JSON',
-                      'RegionId': 'cn-shanghai',
-                      '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'}
+        parameters = {
+            "AccessKeyId": access_key_id,
+            "Action": "CreateToken",
+            "Format": "JSON",
+            "RegionId": "cn-shanghai",
+            "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)
-        print('规范化的请求字符串: %s' % query_string)
+        print("规范化的请求字符串: %s" % query_string)
        # 构造待签名字符串
-        string_to_sign = 'GET' + '&' + AccessToken._encode_text('/') + '&' + AccessToken._encode_text(query_string)
-        print('待签名的字符串: %s' % string_to_sign)
+        string_to_sign = (
+            "GET"
+            + "&"
+            + AccessToken._encode_text("/")
+            + "&"
+            + AccessToken._encode_text(query_string)
+        )
+        print("待签名的字符串: %s" % string_to_sign)
        # 计算签名
-        secreted_string = hmac.new(bytes(access_key_secret + '&', encoding='utf-8'),
-                                   bytes(string_to_sign, encoding='utf-8'),
-                                   hashlib.sha1).digest()
+        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)
-        print('签名: %s' % signature)
+        print("签名: %s" % signature)
        # 进行URL编码
        signature = AccessToken._encode_text(signature)
-        print('URL编码后的签名: %s' % signature)
+        print("URL编码后的签名: %s" % signature)
        # 调用服务
-        full_url = 'http://nls-meta.cn-shanghai.aliyuncs.com/?Signature=%s&%s' % (signature, query_string)
-        print('url: %s' % full_url)
+        full_url = "http://nls-meta.cn-shanghai.aliyuncs.com/?Signature=%s&%s" % (
+            signature,
+            query_string,
+        )
+        print("url: %s" % full_url)
        # 提交HTTP GET请求
        response = requests.get(full_url)
        if response.ok:
            root_obj = response.json()
-            key = 'Token'
+            key = "Token"
            if key in root_obj:
-                token = root_obj[key]['Id']
-                expire_time = root_obj[key]['ExpireTime']
+                token = root_obj[key]["Id"]
+                expire_time = root_obj[key]["ExpireTime"]
                return token, expire_time
        print(response.text)
        return None, None
@@ -77,22 +90,23 @@ class AccessToken:

 class TTSProvider(TTSProviderBase):

-
    def __init__(self, config, delete_audio_file):
        super().__init__(config, delete_audio_file)
-        
+
        # 新增空值判断逻辑
        access_key_id = config.get("access_key_id")
        access_key_secret = config.get("access_key_secret")
        if access_key_id and access_key_secret:
            # 使用密钥对生成临时token
-            token, expire_time = AccessToken.create_token(access_key_id, access_key_secret)
+            token, expire_time = AccessToken.create_token(
+                access_key_id, access_key_secret
+            )
        else:
            # 直接使用预生成的长期token
            token = config.get("token")
            expire_time = None
-        
-        print('token: %s, expire time(s): %s' % (token, expire_time))
+
+        print("token: %s, expire time(s): %s" % (token, expire_time))

        self.appkey = config.get("appkey")
        self.token = token
@@ -105,12 +119,13 @@ class TTSProvider(TTSProviderBase):

        self.host = config.get("host", "nls-gateway-cn-shanghai.aliyuncs.com")
        self.api_url = f"https://{self.host}/stream/v1/tts"
-        self.header = {
-            "Content-Type": "application/json"
-        }
+        self.header = {"Content-Type": "application/json"}

    def generate_filename(self, extension=".wav"):
-        return os.path.join(self.output_file, f"tts-{__name__}{datetime.now().date()}@{uuid.uuid4().hex}{extension}")
+        return os.path.join(
+            self.output_file,
+            f"tts-{__name__}{datetime.now().date()}@{uuid.uuid4().hex}{extension}",
+        )

    async def text_to_speak(self, u_id, text, is_last_text=False, is_first_text=False):
        request_json = {
@@ -122,25 +137,34 @@ class TTSProvider(TTSProviderBase):
            "voice": self.voice,
            "volume": self.volume,
            "speech_rate": self.speech_rate,
-            "pitch_rate": self.pitch_rate
+            "pitch_rate": self.pitch_rate,
        }

        print(self.api_url, json.dumps(request_json, ensure_ascii=False))
        tmp_file = self.generate_filename()
        try:
-            resp = requests.post(self.api_url, json.dumps(request_json), headers=self.header)
+            resp = requests.post(
+                self.api_url, json.dumps(request_json), headers=self.header
+            )
            # 检查返回请求数据的mime类型是否是audio/***，是则保存到指定路径下；返回的是binary格式的
-            if resp.headers['Content-Type'].startswith('audio/'):
-                with open(tmp_file, 'wb') as f:
+            if resp.headers["Content-Type"].startswith("audio/"):
+                with open(tmp_file, "wb") as f:
                    f.write(resp.content)
            else:
-                raise Exception(f"{__name__} status_code: {resp.status_code} response: {resp.content}")
+                raise Exception(
+                    f"{__name__} status_code: {resp.status_code} response: {resp.content}"
+                )
            # 使用 pydub 读取临时文件
            audio = AudioSegment.from_file(tmp_file, format="wav")
            audio = audio.set_channels(1).set_frame_rate(16000)
            opus_datas = self.wav_to_opus_data_audio_raw(audio.raw_data)
-            yield TTSMessageDTO(u_id=u_id, msg_type=MsgType.TTS_TEXT_RESPONSE, content=opus_datas,
-                                tts_finish_text=text, sentence_type=SentenceType.SENTENCE_START)
+            yield TTSMessageDTO(
+                u_id=u_id,
+                msg_type=MsgType.TTS_TEXT_RESPONSE,
+                content=opus_datas,
+                tts_finish_text=text,
+                sentence_type=SentenceType.SENTENCE_START,
+            )
            # 用完后删除临时文件
            try:
                os.remove(tmp_file)
@@ -35,11 +35,25 @@ class TTSProviderBase(ABC):
        self.stop_event = threading.Event()

        self.tts_text_buff = []
-        self.punctuations = ("。", "？", "！", "；", "：", ".", "?", "!", ";", ":", " ", ",", "，")
+        self.punctuations = (
+            "。",
+            "？",
+            "！",
+            "；",
+            "：",
+            ".",
+            "?",
+            "!",
+            ";",
+            ":",
+            " ",
+            ",",
+            "，",
+        )
        self.tts_request = False
        self.processed_chars = 0
        self.stream = False
-        self.last_to_opus_raw = b''
+        self.last_to_opus_raw = b""

        # 启动tts_text_queue监听线程
        # 线程任务相关
@@ -51,7 +65,9 @@ class TTSProviderBase(ABC):

    async def open_audio_channels(self):
        # 启动tts_text_queue监听线程
-        tts_priority = threading.Thread(target=self._tts_text_priority_thread, daemon=True)
+        tts_priority = threading.Thread(
+            target=self._tts_text_priority_thread, daemon=True
+        )
        tts_priority.start()

    async def stop_listen_resource(self):
@@ -67,15 +83,19 @@ class TTSProviderBase(ABC):
    def _get_segment_text(self):
        # 合并当前全部文本并处理未分割部分
        full_text = "".join(self.tts_text_buff)
-        current_text = full_text[self.processed_chars:]  # 从未处理的位置开始
+        current_text = full_text[self.processed_chars :]  # 从未处理的位置开始
        last_punct_pos = -1
        for punct in self.punctuations:
            pos = current_text.rfind(punct)
-            if (pos != -1 and last_punct_pos == -1) or (pos != -1 and pos < last_punct_pos):
+            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)
+            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)  # 更新已处理字符位置
            return segment_text
        else:
@@ -98,11 +118,21 @@ class TTSProviderBase(ABC):
    async def finish_session(self, session_id):
        pass

-    async def tts_one_sentence(self,text):
+    async def tts_one_sentence(self, text):
        uuid_str = str(uuid.uuid4()).replace("-", "")
-        self.tts.tts_text_queue.put(TTSMessageDTO(u_id=uuid_str, msg_type=MsgType.START_TTS_REQUEST, content=''))
-        self.tts.tts_text_queue.put(TTSMessageDTO(u_id=uuid_str, msg_type=MsgType.TTS_TEXT_REQUEST, content=text))
-        self.tts.tts_text_queue.put(TTSMessageDTO(u_id=uuid_str, msg_type=MsgType.STOP_TTS_REQUEST, content=text))
+        self.tts.tts_text_queue.put(
+            TTSMessageDTO(u_id=uuid_str, msg_type=MsgType.START_TTS_REQUEST, content="")
+        )
+        self.tts.tts_text_queue.put(
+            TTSMessageDTO(
+                u_id=uuid_str, msg_type=MsgType.TTS_TEXT_REQUEST, content=text
+            )
+        )
+        self.tts.tts_text_queue.put(
+            TTSMessageDTO(
+                u_id=uuid_str, msg_type=MsgType.STOP_TTS_REQUEST, content=text
+            )
+        )

    def _enable_two_way_tts(self):
        while not self.stop_event.is_set():
@@ -114,27 +144,36 @@ class TTSProviderBase(ABC):
                    self.tts_request = True
                    self.u_id = ttsMessageDTO.u_id
                    # 开启session
-                    future = asyncio.run_coroutine_threadsafe(self.start_session(ttsMessageDTO.u_id), loop=self.loop)
+                    future = asyncio.run_coroutine_threadsafe(
+                        self.start_session(ttsMessageDTO.u_id), loop=self.loop
+                    )
                    future.result()
                    # await self.start_session(ttsMessageDTO.u_id)
                elif self.tts_request and msg_type == MsgType.TTS_TEXT_REQUEST:
                    future = asyncio.run_coroutine_threadsafe(
-                        self.text_to_speak(u_id=ttsMessageDTO.u_id, text=ttsMessageDTO.content), loop=self.loop)
+                        self.text_to_speak(
+                            u_id=ttsMessageDTO.u_id, text=ttsMessageDTO.content
+                        ),
+                        loop=self.loop,
+                    )
                    future.result()
                elif msg_type == MsgType.STOP_TTS_REQUEST:
                    self.tts_request = False
-                    future = asyncio.run_coroutine_threadsafe(self.finish_session(ttsMessageDTO.u_id), loop=self.loop)
+                    future = asyncio.run_coroutine_threadsafe(
+                        self.finish_session(ttsMessageDTO.u_id), loop=self.loop
+                    )
                    future.result()

-
            except Exception as e:
                logger.bind(tag=TAG).error(f"Failed to process TTS text: {e}")
                # 报错了。要关闭说话
                self.tts_audio_queue.put(
                    TTSMessageDTO(
-                        u_id=self.u_id, msg_type=MsgType.STOP_TTS_RESPONSE, content=[],
-                        tts_finish_text='',
-                        sentence_type=None
+                        u_id=self.u_id,
+                        msg_type=MsgType.STOP_TTS_RESPONSE,
+                        content=[],
+                        tts_finish_text="",
+                        sentence_type=None,
                    )
                )
                traceback.print_exc()
@@ -162,25 +201,43 @@ class TTSProviderBase(ABC):
                        if segment_text:
                            # 修改部分：创建协程对象
                            # 修改部分：创建协程对象
-                            tts_generator = self.text_to_speak(ttsMessageDTO.u_id, segment_text,
-                                                               True if msg_type == MsgType.STOP_TTS_REQUEST else False,
-                                                               True if msg_type == MsgType.START_TTS_REQUEST else False)
-                            future = asyncio.run_coroutine_threadsafe(self.process_generator(tts_generator), self.loop)
+                            tts_generator = self.text_to_speak(
+                                ttsMessageDTO.u_id,
+                                segment_text,
+                                True if msg_type == MsgType.STOP_TTS_REQUEST else False,
+                                (
+                                    True
+                                    if msg_type == MsgType.START_TTS_REQUEST
+                                    else False
+                                ),
+                            )
+                            future = asyncio.run_coroutine_threadsafe(
+                                self.process_generator(tts_generator), self.loop
+                            )
                            self.active_tasks.add(future)
                        if self.active_tasks:
+
                            async def wrap_future(future):
                                return await asyncio.wrap_future(future)

-                            wrapped_tasks = [wrap_future(task) for task in self.active_tasks]
-                            done, _ = loop.run_until_complete(asyncio.wait(wrapped_tasks))
+                            wrapped_tasks = [
+                                wrap_future(task) for task in self.active_tasks
+                            ]
+                            done, _ = loop.run_until_complete(
+                                asyncio.wait(wrapped_tasks)
+                            )
                            self.active_tasks -= done

                        # 发送合成结束
-                        self.tts_audio_queue.put(TTSMessageDTO(u_id=ttsMessageDTO.u_id,
-                                                               msg_type=MsgType.STOP_TTS_RESPONSE,
-                                                               content=[],
-                                                               tts_finish_text='',
-                                                               sentence_type=SentenceType.SENTENCE_END))
+                        self.tts_audio_queue.put(
+                            TTSMessageDTO(
+                                u_id=ttsMessageDTO.u_id,
+                                msg_type=MsgType.STOP_TTS_RESPONSE,
+                                content=[],
+                                tts_finish_text="",
+                                sentence_type=SentenceType.SENTENCE_END,
+                            )
+                        )

                    segment_text = self._get_segment_text()
                    if segment_text:
@@ -189,25 +246,31 @@ class TTSProviderBase(ABC):
                            ttsMessageDTO.u_id,
                            segment_text,
                            msg_type == MsgType.STOP_TTS_REQUEST,
-                            msg_type == MsgType.START_TTS_REQUEST
+                            msg_type == MsgType.START_TTS_REQUEST,
                        )
                        # 提交协程到事件循环
                        tts_generator_future = asyncio.run_coroutine_threadsafe(
-                            self.process_generator(tts_generator),
-                            loop
+                            self.process_generator(tts_generator), loop
                        )
                        self.active_tasks.add(tts_generator_future)
                        if len(self.active_tasks) >= self.max_workers:
                            # 等待所有任务完成
                            try:
+
                                async def wrap_future(future):
                                    return await asyncio.wrap_future(future)

-                                wrapped_tasks = [wrap_future(task) for task in self.active_tasks]
-                                done, _ = loop.run_until_complete(asyncio.wait(wrapped_tasks))
+                                wrapped_tasks = [
+                                    wrap_future(task) for task in self.active_tasks
+                                ]
+                                done, _ = loop.run_until_complete(
+                                    asyncio.wait(wrapped_tasks)
+                                )
                                self.active_tasks -= done
                            except Exception as e:
-                                logger.bind(tag=TAG).error(f"Failed to process TTS text: {e}")
+                                logger.bind(tag=TAG).error(
+                                    f"Failed to process TTS text: {e}"
+                                )
                                traceback.print_exc()
                else:
                    pass
@@ -227,10 +290,14 @@ class TTSProviderBase(ABC):
                asyncio.run(self.text_to_speak(text, tmp_file))
                if not os.path.exists(tmp_file):
                    max_repeat_time = max_repeat_time - 1
-                    logger.bind(tag=TAG).error(f"语音生成失败: {text}:{tmp_file}，再试{max_repeat_time}次")
+                    logger.bind(tag=TAG).error(
+                        f"语音生成失败: {text}:{tmp_file}，再试{max_repeat_time}次"
+                    )

            if max_repeat_time > 0:
-                logger.bind(tag=TAG).info(f"语音生成成功: {text}:{tmp_file}，重试{5 - max_repeat_time}次")
+                logger.bind(tag=TAG).info(
+                    f"语音生成成功: {text}:{tmp_file}，重试{5 - max_repeat_time}次"
+                )

            return tmp_file
        except Exception as e:
@@ -256,7 +323,7 @@ class TTSProviderBase(ABC):
        # 获取文件后缀名
        file_type = os.path.splitext(audio_file_path)[1]
        if file_type:
-            file_type = file_type.lstrip('.')
+            file_type = file_type.lstrip(".")
        audio = AudioSegment.from_file(audio_file_path, format=file_type)

        # 转换为单声道/16kHz采样率/16位小端编码（确保与编码器匹配）
@@ -279,11 +346,11 @@ class TTSProviderBase(ABC):
        # 按帧处理所有音频数据（包括最后一帧可能补零）
        for i in range(0, len(raw_data), frame_size * 2):  # 16bit=2bytes/sample
            # 获取当前帧的二进制数据
-            chunk = raw_data[i:i + frame_size * 2]
+            chunk = raw_data[i : i + frame_size * 2]

            # 如果最后一帧不足，补零
            if len(chunk) < frame_size * 2:
-                chunk += b'\x00' * (frame_size * 2 - len(chunk))
+                chunk += b"\x00" * (frame_size * 2 - len(chunk))

            # 转换为numpy数组处理
            np_frame = np.frombuffer(chunk, dtype=np.int16)
@@ -295,7 +362,9 @@ class TTSProviderBase(ABC):
        return opus_datas, duration

    def get_audio_from_tts(self, data_bytes, src_rate, to_rate=16000):
-        tts_speech = torch.from_numpy(np.array(np.frombuffer(data_bytes, dtype=np.int16))).unsqueeze(dim=0)
+        tts_speech = torch.from_numpy(
+            np.array(np.frombuffer(data_bytes, dtype=np.int16))
+        ).unsqueeze(dim=0)
        with io.BytesIO() as bf:
            torchaudio.save(bf, tts_speech, src_rate, format="wav")
            audio = AudioSegment.from_file(bf, format="wav")
@@ -304,7 +373,7 @@ class TTSProviderBase(ABC):

    def wav_to_opus_data_audio_raw(self, raw_data_var, is_end=False):
        raw_data = self.last_to_opus_raw + raw_data_var
-        self.last_to_opus_raw = b''
+        self.last_to_opus_raw = b""
        # 初始化Opus编码器
        encoder = opuslib_next.Encoder(16000, 1, opuslib_next.APPLICATION_AUDIO)

@@ -316,7 +385,7 @@ class TTSProviderBase(ABC):
        # 按帧处理所有音频数据（包括最后一帧可能补零）
        for i in range(0, len(raw_data), frame_size * 2):  # 16bit=2bytes/sample
            # 获取当前帧的二进制数据
-            chunk = raw_data[i:i + frame_size * 2]
+            chunk = raw_data[i : i + frame_size * 2]

            # 如果最后一帧不足，补零
            # 缓存记录一下
@@ -326,7 +395,7 @@ class TTSProviderBase(ABC):
                break
            if len(chunk) < frame_size * 2 and is_end:
                logger.bind(tag=TAG).info("是最后一句了，补零")
-                chunk += b'\x00' * (frame_size * 2 - len(chunk))
+                chunk += b"\x00" * (frame_size * 2 - len(chunk))

            # 转换为numpy数组处理
            np_frame = np.frombuffer(chunk, dtype=np.int16)
@@ -13,17 +13,23 @@ from core.providers.tts.dto.dto import TTSMessageDTO, MsgType, SentenceType
 TAG = __name__
 logger = setup_logging()

+
 class TTSProvider(TTSProviderBase):
    def __init__(self, config, delete_audio_file):
        super().__init__(config, delete_audio_file)
        self.voice = config.get("voice")

    def generate_filename(self, extension=".mp3"):
-        return os.path.join(self.output_file, f"tts-{datetime.now().date()}@{uuid.uuid4().hex}{extension}")
+        return os.path.join(
+            self.output_file,
+            f"tts-{datetime.now().date()}@{uuid.uuid4().hex}{extension}",
+        )

    async def text_to_speak(self, u_id, text, is_last_text=False, is_first_text=False):
        try:
-            communicate = edge_tts.Communicate(text, voice=self.voice)  # Use your preferred voice
+            communicate = edge_tts.Communicate(
+                text, voice=self.voice
+            )  # Use your preferred voice
            tmp_file = self.generate_filename()
            await communicate.save(tmp_file)

@@ -31,7 +37,13 @@ class TTSProvider(TTSProviderBase):
            audio = AudioSegment.from_file(tmp_file, format="mp3")
            audio = audio.set_channels(1).set_frame_rate(16000)
            opus_datas = self.wav_to_opus_data_audio_raw(audio.raw_data)
-            yield TTSMessageDTO(u_id=u_id, msg_type=MsgType.TTS_TEXT_RESPONSE, content=opus_datas, tts_finish_text=text,sentence_type=SentenceType.SENTENCE_START)
+            yield TTSMessageDTO(
+                u_id=u_id,
+                msg_type=MsgType.TTS_TEXT_RESPONSE,
+                content=opus_datas,
+                tts_finish_text=text,
+                sentence_type=SentenceType.SENTENCE_START,
+            )
            # 用完后删除临时文件
            try:
                os.remove(tmp_file)
@@ -40,6 +52,10 @@ class TTSProvider(TTSProviderBase):
                pass
        except Exception as e:
            logger.bind(tag=TAG).error(f"TTSProvider text_to_speak error: {e}")
-            yield TTSMessageDTO(u_id=u_id, msg_type=MsgType.TTS_TEXT_RESPONSE, content=[], tts_finish_text=text,sentence_type=SentenceType.SENTENCE_START)
-
-
+            yield TTSMessageDTO(
+                u_id=u_id,
+                msg_type=MsgType.TTS_TEXT_RESPONSE,
+                content=[],
+                tts_finish_text=text,
+                sentence_type=SentenceType.SENTENCE_START,
+            )