voice-gateway/voip/asr.py

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且未过期
        if cls.token_cache["token"] and cls.token_cache["expire_time"]:
            current_time = int(time.time())
            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):
        self.__th = threading.Thread(target=self.__test_run)
        self.__id = tid
        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.")

    def __test_run(self):
        print("Thread:{} start..".format(self.__id))
        self.sr = nls.NlsSpeechTranscriber(
            url=URL,
            token=self._Token,
            appkey=APPKEY,
            on_error=self.test_on_error,
            on_result_changed=self.test_on_result_chg,
            on_completed=self.test_on_completed
        )
        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_error(self, message, *args):
        print(f"Error occurred: {message}")
    def test_on_result_chg(self, message, *args):
        print(f"Result changed: {message}")

    def test_on_completed(self, message, *args):
        print(f"Recognition completed: {message}")