讯飞语音文字互转

Object/WsParam/WsParamRecognizeObject.py

@@ -0,0 +1,112 @@
+import websocket
+import datetime
+import hashlib
+import base64
+import hmac
+import json
+from urllib.parse import urlencode
+import time
+import ssl
+from wsgiref.handlers import format_date_time
+from datetime import datetime
+from time import mktime
+import _thread as thread
+
+"""
+调用讯飞模型 语音转文字
+"""
+
+class WsParamRecognize:
+    STATUS_FIRST_FRAME = 0
+    STATUS_CONTINUE_FRAME = 1
+    STATUS_LAST_FRAME = 2
+
+    def __init__(self, APPID, APISecret, APIKey, AudioFile):
+        self.APPID = APPID
+        self.APISecret = APISecret
+        self.APIKey = APIKey
+        self.AudioFile = AudioFile
+        self.all_sentences = ""
+
+        # 公共参数
+        self.CommonArgs = {"app_id": self.APPID}
+        # 业务参数
+        self.BusinessArgs = {"domain": "iat", "language": "zh_cn", "accent": "mandarin", "vinfo": 1, "vad_eos": 10000}
+
+    def create_url(self):
+        url = 'wss://ws-api.xfyun.cn/v2/iat'
+        now = datetime.now()
+        date = format_date_time(mktime(now.timetuple()))
+        signature_origin = "host: ws-api.xfyun.cn\n" + "date: " + date + "\nGET /v2/iat HTTP/1.1"
+        signature_sha = hmac.new(self.APISecret.encode('utf-8'), signature_origin.encode('utf-8'),
+                                 digestmod=hashlib.sha256).digest()
+        signature_sha = base64.b64encode(signature_sha).decode(encoding='utf-8')
+        authorization_origin = "api_key=\"%s\", algorithm=\"hmac-sha256\", headers=\"host date request-line\", signature=\"%s\"" % (
+            self.APIKey, signature_sha)
+        authorization = base64.b64encode(authorization_origin.encode('utf-8')).decode(encoding='utf-8')
+        v = {"authorization": authorization, "date": date, "host": "ws-api.xfyun.cn"}
+        url = url + '?' + urlencode(v)
+        return url
+
+    def on_message(self, ws, message):
+        try:
+            message = json.loads(message)
+            code = message["code"]
+            sid = message["sid"]
+            if code != 0:
+                errMsg = message["message"]
+                print("sid:%s call error:%s code is:%s" % (sid, errMsg, code))
+            else:
+                data = message["data"]["result"]["ws"]
+                for i in data:
+                    for w in i["cw"]:
+                        self.all_sentences += w["w"]
+        except Exception as e:
+            print("receive msg,but parse exception:", e)
+
+    def on_error(self, error):
+        print("### error:", error)
+
+    def on_close(self):
+        print("### closed ###")
+
+    def on_open(self, ws):
+        def run(*args):
+            frameSize = 8000
+            intervel = 0.04
+            status = self.STATUS_FIRST_FRAME
+
+            with open(self.AudioFile, "rb") as fp:
+                while True:
+                    buf = fp.read(frameSize)
+                    if not buf:
+                        status = self.STATUS_LAST_FRAME
+                    if status == self.STATUS_FIRST_FRAME:
+                        d = {"common": self.CommonArgs, "business": self.BusinessArgs,
+                             "data": {"status": 0, "format": "audio/L16;rate=16000",
+                                      "audio": str(base64.b64encode(buf), 'utf-8'), "encoding": "raw"}}
+                        ws.send(json.dumps(d))
+                        status = self.STATUS_CONTINUE_FRAME
+                    elif status == self.STATUS_CONTINUE_FRAME:
+                        ws.send(json.dumps({"data": {"status": 1, "format": "audio/L16;rate=16000",
+                                                     "audio": str(base64.b64encode(buf), 'utf-8'), "encoding": "raw"}}))
+                    elif status == self.STATUS_LAST_FRAME:
+                        ws.send(json.dumps({"data": {"status": 2, "format": "audio/L16;rate=16000",
+                                                     "audio": str(base64.b64encode(buf), 'utf-8'), "encoding": "raw"}}))
+                        time.sleep(1)
+                        break
+                    time.sleep(intervel)
+            ws.close()
+
+        thread.start_new_thread(run, ())
+
+    def start(self):
+        websocket.enableTrace(False)
+        wsUrl = self.create_url()
+        self.ws = websocket.WebSocketApp(wsUrl,
+                                         on_message=lambda ws, msg: self.on_message(ws, msg),
+                                         on_error=self.on_error,
+                                         on_close=self.on_close,
+                                         on_open=lambda ws: self.on_open(ws))
+        self.ws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE})
+        return self.all_sentences

Object/WsParam/WsParamSynthesizeObject.py

@@ -0,0 +1,116 @@
+import websocket
+import datetime
+import hashlib
+import base64
+import hmac
+import json
+from urllib.parse import urlencode
+import ssl
+from wsgiref.handlers import format_date_time
+from datetime import datetime
+from time import mktime
+import _thread as thread
+import os
+
+
+"""
+调用讯飞模型 文字转语音 
+"""
+
+class WsParamSynthesize:
+    def __init__(self, APPID, APIKey, APISecret, Text, output_file='demo.pcm'):
+        self.APPID = APPID
+        self.APIKey = APIKey
+        self.APISecret = APISecret
+        self.Text = Text
+        self.output_file = output_file
+
+        # 初始化其他需要的属性
+        self.CommonArgs = {"app_id": self.APPID}
+        self.BusinessArgs = {"aue": "raw", "auf": "audio/L16;rate=16000", "vcn": "xiaoyan", "tte": "utf8"}
+        self.Data = {"status": 2, "text": str(base64.b64encode(self.Text.encode('utf-8')), "UTF8")}
+
+    def create_url(self):
+        url = 'wss://tts-api.xfyun.cn/v2/tts'
+        # 生成RFC1123格式的时间戳
+        now = datetime.now()
+        date = format_date_time(mktime(now.timetuple()))
+
+        # 拼接字符串
+        signature_origin = "host: " + "ws-api.xfyun.cn" + "\n"
+        signature_origin += "date: " + date + "\n"
+        signature_origin += "GET " + "/v2/tts " + "HTTP/1.1"
+        # 进行hmac-sha256进行加密
+        signature_sha = hmac.new(self.APISecret.encode('utf-8'), signature_origin.encode('utf-8'),
+                                 digestmod=hashlib.sha256).digest()
+        signature_sha = base64.b64encode(signature_sha).decode(encoding='utf-8')
+
+        authorization_origin = "api_key=\"%s\", algorithm=\"%s\", headers=\"%s\", signature=\"%s\"" % (
+            self.APIKey, "hmac-sha256", "host date request-line", signature_sha)
+        authorization = base64.b64encode(authorization_origin.encode('utf-8')).decode(encoding='utf-8')
+        # 将请求的鉴权参数组合为字典
+        v = {
+            "authorization": authorization,
+            "date": date,
+            "host": "ws-api.xfyun.cn"
+        }
+        # 拼接鉴权参数，生成url
+        url = url + '?' + urlencode(v)
+        # print("date: ",date)
+        # print("v: ",v)
+        # 此处打印出建立连接时候的url,参考本demo的时候可取消上方打印的注释，比对相同参数时生成的url与自己代码生成的url是否一致
+        # print('websocket url :', url)
+        return url
+
+    def on_message(self, ws, message):
+        try:
+            print("Received message:", message)  # 打印接收到的消息
+            message = json.loads(message)
+            code = message["code"]
+            sid = message["sid"]
+            audio = message["data"]["audio"]
+            audio = base64.b64decode(audio)
+            status = message["data"]["status"]
+            print(message)
+            if status == 2:
+                print("WebSocket connection is closed.")
+                ws.close()
+            if code != 0:
+                errMsg = message["message"]
+                print(f"sid:{sid} call error:{errMsg} code is:{code}")
+            else:
+                print("Writing audio data to file.")
+                with open(self.output_file, 'ab') as f:
+                    f.write(audio)
+        except Exception as e:
+            print("Receive message, but parse exception:", e)
+
+    # on_error和on_close方法类似地修改，可以访问类实例的属性
+    def on_error(self, error):
+        print("### error:", error)
+
+    # 收到websocket关闭的处理
+    def on_close(self):
+        print("### closed ###")
+
+    def on_open(self, ws):
+        def run(*args):
+            print("WebSocket connection is open.")
+            d = {"common": self.CommonArgs, "business": self.BusinessArgs, "data": self.Data}
+            d = json.dumps(d)
+            print("Sending text data...")
+            ws.send(d)
+            if os.path.exists(self.output_file):
+                os.remove(self.output_file)
+
+        print("Starting new thread for sending data...")
+        thread.start_new_thread(run, ())
+
+    def start(self):
+        websocket.enableTrace(False)
+        self.ws = websocket.WebSocketApp(self.create_url(),
+                                         on_message=lambda ws, msg: self.on_message(ws, msg),
+                                         on_error=lambda msg: self.on_error(msg),
+                                         on_close=self.on_close,
+                                         on_open=lambda ws: self.on_open(ws))  # 使用 lambda 来确保 ws 参数传递
+        self.ws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE})