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

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

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

# 获取Token的函数
def get_token():
    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

# WebSocket服务地址
URL = "wss://nls-gateway-cn-beijing.aliyuncs.com/ws/v1"
APPKEY = "OKt6jogp6fRjHQVp"  # 你的Appkey

# 定义实时转写类
class TestSt:
    # 静态变量用于缓存Token
    token_cache = {
        "token": None,
        "expire_time": 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 = 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, message_receiver=None):
        self.is_closed = False
        self.lock = threading.Lock()

        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
        print("开始")

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

    def send_audio(self, audio_data):
        # if self.sr:
        #     # print("Sending audio data of length:", len(audio_data))
        #     self.sr.send_audio(audio_data)
            # print("Audio data sent.")
        if self.sr and not self.is_closed:
            with self.lock:
                try:
                    self.sr.send_audio(audio_data)
                except Exception as e:
                    print(f"Error sending audio: {e}")
                    self.close()
    def close(self):
        with self.lock:
            if not self.is_closed:
                self.is_closed = True
                try:
                    self.sr.stop()
                except Exception as e:
                    print(f"Error stopping ASR: {e}")

    def __test_run(self):
        print("Thread:{} start..".format(self.__id))
        self.sr = nls.NlsSpeechTranscriber(
            url=URL,
            token=self._Token,
            appkey=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]
        )
        self.sr.__setattr__("max_sentence_silence", 1200)
        self.sr.start(
            aformat="pcm",
            enable_intermediate_result=True,
            enable_punctuation_prediction=True,
            enable_inverse_text_normalization=True
        )
        print("ASR session started.")

    def test_on_sentence_begin(self, message, *args):
        # pass
        # print("test_on_sentence_begin:{}".format(message))
        if self.message_receiver:
            self.message_receiver(message, *args)

    def test_on_sentence_end(self, message, *args):
        # print("test_on_sentence_end:{}".format(message))
        if self.message_receiver:
            self.message_receiver(message, *args)

    def test_on_start(self, message, *args):
        # print("test_on_start:{}".format(message))
        pass

    def test_on_error(self, message, *args):
        # print("on_error args=>{}".format(args))
        if self.message_receiver:
            self.message_receiver(message, *args)

    def test_on_close(self, *args):
        # print("on_close: args=>{}".format(args))
        pass

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

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