voice-gateway-service/src/core/voip/asr.py

#!/usr/bin/env python3
# encoding:utf-8

import os
import json
import threading
import traceback

import nls  # 引入阿里云语音识别库
from aliyunsdkcore.client import AcsClient
from aliyunsdkcore.request import CommonRequest
import time


# 定义实时转写类
class TestSt:
    # 静态变量用于缓存Token
    token_cache = {
        "token": None,
        "expire_time": None
    }

    # 获取Token的函数
    @classmethod
    def get_token(cls):
        ak_id = "LTAI5tQ2HmiHCygZkt5BYrYR"
        ak_secret = "KhmxTd14SUcXafpFk5yofA43FoeM99"
        client = AcsClient(ak_id, ak_secret, "cn-shanghai")
        request = CommonRequest()
        request.set_method('POST')
        request.set_domain('nls-meta.cn-shanghai.aliyuncs.com')
        request.set_version('2019-02-28')
        request.set_action_name('CreateToken')

        try:
            response = client.do_action_with_exception(request)
            jss = json.loads(response)
            if 'Token' in jss and 'Id' in jss['Token']:
                token = jss['Token']['Id']
                expire_time = jss['Token']['ExpireTime']
                print(f"Token: {token}, ExpireTime: {expire_time}")
                return token, int(expire_time)  # 返回Token和过期时间
            else:
                print("Token获取失败，响应内容: ", response)
        except Exception as e:
            print(f"获取Token时发生错误: {e}")
        return None, None

    @classmethod
    def get_cached_token(cls):
        # 检查是否已有缓存的Token且未过期s):
        #         # 检查是否已有缓存的Token且未
        current_time = int(time.time())
        # if cls.token_cache["token"] and cls.token_cache["expire_time"]:
        if cls.token_cache["token"] and cls.token_cache["expire_time"] - current_time > 60:
            # if current_time < cls.token_cache["expire_time"]:
            #     print("使用缓存的Token")
            return cls.token_cache["token"]

        # 如果没有缓存Token或者Token已过期，重新获取
        new_token, expire_time = cls.get_token()
        if new_token:
            cls.token_cache["token"] = new_token
            cls.token_cache["expire_time"] = expire_time
            print("获取新的Token")
            return new_token
        else:
            print("无法获取Token")
            return None

    def __init__(self, tid, logger, message_receiver=None):
        # self.is_closed = False
        # self.lock = threading.Lock()
        self.logger = logger
        self.__event = threading.Event()
        self.__th = threading.Thread(target=self.__test_run)
        self.__id = tid
        self.message_receiver = message_receiver
        self._Token = self.get_cached_token()
        self.sr = None

        # WebSocket服务地址
        self.URL = "wss://nls-gateway-cn-beijing.aliyuncs.com/ws/v1"
        self.APPKEY = "OKt6jogp6fRjHQVp"  # 你的Appkey
        self.logger.debug("开始")

    def start(self):
        self.__th.start()

    def send_audio(self, audio_data):
        if self.sr:
            self.sr.send_audio(audio_data)

    def close(self):
        try:
            self.sr.stop()
        except Exception as e:
            self.logger.debug(f"[{self.__id}]Error stopping ASR: {e}")

    def __test_run(self):
        self.logger.debug("Thread:%s start..",self.__id)
        nls.enableTrace(True)
        count = 0
        self.__event.clear()
        while not self.__event.is_set():
            self.sr = nls.NlsSpeechTranscriber(
                url=self.URL,
                token=self._Token,
                appkey=self.APPKEY,
                on_sentence_begin=self.test_on_sentence_begin,
                on_sentence_end=self.test_on_sentence_end,
                on_start=self.test_on_start,
                on_result_changed=self.test_on_result_chg,
                on_completed=self.test_on_completed,
                on_error=self.test_on_error,
                on_close=self.test_on_close,
                callback_args=[self.__id]
            )
            try:
                self.sr.start(
                    aformat="pcm",
                    sample_rate=8000,
                    enable_intermediate_result=True,
                    enable_punctuation_prediction=True,
                    enable_inverse_text_normalization=True,
                    ex={'max_sentence_silence': 2000, 'disfluency': True, 'enable_words': True}
                )
                # _res = self.sr.ctrl(ex={'max_sentence_silence': 6000, 'disfluency': True,'enable_words': True })
                self.logger.debug(f"[{self.__id}]ASR session started. {count}")
                self.__event.wait(timeout=.5)
                self.logger.debug(f"[{self.__id}]ASR session started. {count}")
            except Exception as e:
                traceback.print_exc()
                self.logger.debug(f"[{self.__id}]ASR session start exception. {e}")

            count = count + 1

    def test_on_sentence_begin(self, message, *args):
        self.logger.debug("[%s]test_on_sentence_begin:%s", self.__id, message)
        if self.message_receiver:
            self.message_receiver(message, *args)

    def test_on_sentence_end(self, message, *args):
        self.logger.debug("[%s]test_on_sentence_end:%s", self.__id, message)
        if self.message_receiver:
            self.message_receiver(message, *args)

    def test_on_start(self, message, *args):
        self.__event.set()
        self.logger.debug("[%s]test_on_start:%s", self.__id, message)
        pass

    def test_on_error(self, message, *args):
        self.logger.debug("on_error args=>%s", args)
        if not self.__event.is_set():
            self.__event.set()
        if self.message_receiver:
            self.message_receiver(message, *args)

    def test_on_close(self, *args):
        self.logger.debug("on_close: args=>%s", args)
        if not self.__event.is_set():
            self.__event.set()
        pass

    def test_on_result_chg(self, message, *args):
        # self.logger.debug("test_on_chg:{}".format(message))
        if self.message_receiver:
            self.message_receiver(message, *args)

    def test_on_completed(self, message, *args):
        # self.logger.debug("on_completed:args=>{} message=>{}".format(args, message))
        pass


# 讯飞ASR实时转写
class XfAsr:
    def __init__(self, tid, logger, message_receiver=None):
        import hashlib
        import hmac
        import base64
        # from socket import *
        # import json, time, threading
        from websocket import create_connection
        # import websocket
        from urllib.parse import quote

        app_id = "1ec1097b"
        api_key = "60b7d2d8d172b065b1c3e723e5ba0696"
        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}"
        self.tid = tid
        self.logger = logger
        self.message_receiver = message_receiver
        self.ws = create_connection(base_url + "?appid=" + app_id + "&ts=" + ts + "&signa=" + quote(signa))
        self.trecv = threading.Thread(target=self.recv)

    def start(self):
        self.trecv.start()

    # def send(self, file_path):
    #     file_object = open(file_path, 'rb')
    #     try:
    #         index = 1
    #         while True:
    #             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 send_audio(self, chunk):
        self.ws.send(chunk)

    def close(self):
        self.ws.send(bytes(self.end_tag.encode('utf-8')))
        self.ws.close()
        print("connection closed")

    def recv(self):
        try:
            while self.ws.connected:
                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 Exception as e:
            traceback.print_exc()
            print("receive result end", e)