refactor: 豆包接口校验发送demo验证

apps/setting/models_provider/impl/volcanic_engine_model_provider/model/stt.py

@@ -11,6 +11,7 @@ import base64
 import gzip
 import hmac
 import json
+import os
 import uuid
 import wave
 from enum import Enum
@@ -144,6 +145,7 @@ def parse_response(res):
         result['code'] = code
         payload_size = int.from_bytes(payload[4:8], "big", signed=False)
         payload_msg = payload[8:]
+        print(f"Error code: {code}, message: {payload_msg}")
     if payload_msg is None:
         return result
     if message_compression == GZIP:
@@ -321,14 +323,9 @@ class VolcanicEngineSpeechToText(MaxKBBaseModel, BaseSpeechToText):
         return result['payload_msg']['result'][0]['text']
 
     def check_auth(self):
-        header = self.token_auth()
-
-        async def check():
-            async with websockets.connect(self.volcanic_api_url, extra_headers=header, max_size=1000000000,
-                                          ssl=ssl_context) as ws:
-                pass
-
-        asyncio.run(check())
+        cwd = os.path.dirname(os.path.abspath(__file__))
+        with open(f'{cwd}/iat_mp3_16k.mp3', 'rb') as f:
+            self.speech_to_text(f)
 
     def speech_to_text(self, file):
         data = file.read()

apps/setting/models_provider/impl/volcanic_engine_model_provider/model/tts.py

@@ -69,14 +69,7 @@ class VolcanicEngineTextToSpeech(MaxKBBaseModel, BaseTextToSpeech):
         )
 
     def check_auth(self):
-        header = self.token_auth()
-
-        async def check():
-            async with websockets.connect(self.volcanic_api_url, extra_headers=header, ping_interval=None,
-                                          ssl=ssl_context) as ws:
-                pass
-
-        asyncio.run(check())
+        self.text_to_speech('你好')
 
     def text_to_speech(self, text):
         request_json = {
@@ -159,7 +152,7 @@ class VolcanicEngineTextToSpeech(MaxKBBaseModel, BaseTextToSpeech):
                 if message_compression == 1:
                     error_msg = gzip.decompress(error_msg)
                 error_msg = str(error_msg, "utf-8")
-                break
+                raise Exception(f"Error code: {code}, message: {error_msg}")
             elif message_type == 0xc:
                 msg_size = int.from_bytes(payload[:4], "big", signed=False)
                 payload = payload[4:]