本文最后更新于752 天前,其中的信息可能已经过时,如有错误请发送邮件到zzy1173071230@gmail.com
最近需要一个能实时语音转文字的功能
# -*- encoding:utf-8 -*-
import hashlib
import hmac
import base64
from socket import *
import json
import time
import threading
from websocket import create_connection
import websocket
from urllib.parse import quote
import logging
class Client():
def __init__(self):
# 讯飞语音识别服务的WebSocket地址
base_url = "ws://rtasr.xfyun.cn/v1/ws"
ts = str(int(time.time()))
tt = (app_id + ts).encode('utf-8')
md5 = hashlib.md5()
md5.update(tt)
baseString = md5.hexdigest()
baseString = bytes(baseString, encoding='utf-8')
apiKey = api_key.encode('utf-8')
signa = hmac.new(apiKey, baseString, hashlib.sha1).digest()
signa = base64.b64encode(signa)
signa = str(signa, 'utf-8')
self.end_tag = "{\"end\": true}"
# 创建WebSocket连接并启动接收线程
self.ws = create_connection(base_url + "?appid=" + app_id + "&ts=" + ts + "&signa=" + quote(signa))
self.trecv = threading.Thread(target=self.recv)
self.trecv.start()
def send(self, file_path):
file_object = open(file_path, 'rb')
try:
index = 1
while True:
# 读取语音文件的1280字节块并通过WebSocket发送
chunk = file_object.read(1280)
if not chunk:
break
self.ws.send(chunk)
index += 1
time.sleep(0.04)
finally:
file_object.close()
# 发送语音结束标志
self.ws.send(bytes(self.end_tag.encode('utf-8')))
print("send end tag success")
def recv(self):
try:
while self.ws.connected:
# 接收WebSocket返回的识别结果
result = str(self.ws.recv())
if len(result) == 0:
print("receive result end")
break
result_dict = json.loads(result)
# 解析结果
if result_dict["action"] == "started":
print("handshake success, result: " + result)
if result_dict["action"] == "result":
result_1 = result_dict
# result_2 = json.loads(result_1["cn"])
# result_3 = json.loads(result_2["st"])
# result_4 = json.loads(result_3["rt"])
print("rtasr result: " + result_1["data"])
if result_dict["action"] == "error":
print("rtasr error: " + result)
self.ws.close()
return
except websocket.WebSocketConnectionClosedException:
print("receive result end")
def close(self):
# 关闭WebSocket连接
self.ws.close()
print("connection closed")
if __name__ == '__main__':
logging.basicConfig()
# 请替换以下变量的值为实际的讯飞语音识别服务信息
app_id = ""
api_key = ""
file_path = r"./test_1.pcm"
# 创建Client实例并发送语音文件
client = Client()
client.send(file_path)
以上内容仅为实力demo,实测有待考证。 稍后布置腾讯云实际调用实例
