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

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

import os
import json
import base64
import threading
import traceback
from datetime import datetime
from src.core.callcenter import registry

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(self.convert_message(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(self.convert_message(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(self.convert_message(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(self.convert_message(message), *args)

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

    def convert_message(self, message):
        final_result = {}
        message = json.loads(message)
        if message["header"]["status"] == 20000000:
            if message["header"]["name"] == "SentenceBegin":
                final_result['name'] = 'SentenceBegin'
            if message["header"]["name"] == "SentenceEnd":
                result = message["payload"]["result"]
                # self.logger.info("asr返回内容Result:%s", result)
                final_result['name'] = 'SentenceEnd'
                final_result['result'] = result
            elif message["header"]["name"] == "TranscriptionResultChanged":
                final_result['name'] = 'TranscriptionResultChanged'
        else:
            final_result['name'] = 'TranscriptionResultError'
            final_result['status'] = message['header']['status']
            final_result['result'] = ''
            self.logger.info(f"Status is not {message['header']['status']}")
            registry.ASR_ERRORS.labels(message['header']['status']).inc()

        self.logger.error("aliyun.Asr.recv: call_id:%s, final_result: %s", self.__id, final_result)
        return final_result


# 讯飞ASR实时转写
class XfAsr:

    def __init__(self, tid, logger, message_receiver=None):
        self.end_tag = "{\"end\": true}"
        self.tid = tid
        self.logger = logger
        self.message_receiver = message_receiver
        self.ws = self.new_connection()
        self.logger.info(f"xunfei.Asr: call_id: {tid}, ws.connected:{self.ws.connected}")
        self.trecv = threading.Thread(target=self.recv)

    def new_connection(self, base_url = "ws://rtasr.xfyun.cn/v1/ws", app_id = "1ec1097b", api_key = "60b7d2d8d172b065b1c3e723e5ba0696"):
        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
        # import logging

        # logging.basicConfig()

        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')

        count = 10
        _ws = None
        while count > 0:
            try:
                _ws = create_connection(base_url + "?appid=" + app_id + "&ts=" + ts + "&signa=" + quote(signa))
                break
            except Exception as e:
                count -= 1
                self.logger.info("new_connection:exception, call_id: %s, count=%s, message=%s", self.tid, count, e)
                time.sleep(.010)
        return _ws

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

    def send_audio(self, chunk):
        # self.logger.debug('xunfei.Asr.send_audio: call_id: %s, chunk:%s, %s', self.tid, len(chunk), chunk)
        if self.ws:
            self.ws.send(chunk)

    def close(self):
        if self.ws:
            self.ws.send(bytes(self.end_tag.encode('utf-8')))
            self.ws.close()
        self.logger.info("connection closed call_id: %s", self.tid)

    def recv(self):
        try:
            self.logger.info(f"xunfei.Asr.recv: call_id: {self.tid}, ws.connected:{self.ws.connected}")
            while self.ws and self.ws.connected:
                message = str(self.ws.recv())
                if len(message) == 0:
                    self.logger.info("xunfei.Asr.recv: receive result end call_id: %s", self.tid)
                    break

                self.logger.info("xunfei.Asr.recv: call_id: %s, message :%s", self.tid, message)
                if self.message_receiver:
                    self.message_receiver(self.convert_message(message))

        except Exception as e:
            traceback.print_exc()
            self.logger.error("xunfei.Asr.recv: receive result end, call_id: %s, message: %s", self.tid, e)

    def convert_message(self, message):
        final_result = {}
        result_dict = json.loads(message)
        if result_dict["code"] == "0":
            if result_dict["action"] == "started":
                final_result['name'] = 'SentenceBegin'
            elif result_dict["action"] == "result":
                result_dict = json.loads(message)
                result_1 = json.loads(result_dict["data"])
                st = result_1["cn"]["st"]
                rt = st["rt"]
                if st['type'] == "1":
                    final_result['name'] = 'TranscriptionResultChanged'
                if st['type'] == "0":
                    final_result['name'] = 'SentenceEnd'
                final_result['result'] = ''.join(cw["w"] for item in rt for ws in item["ws"] for cw in ws["cw"]).strip()
            elif result_dict["action"] == "error":
                self.logger.error("xunfei.Asr.recv: call_id: %s, action is error: %s", self.tid, message)
                final_result['name'] = 'TranscriptionResultError'
                final_result['result'] = message
                if self.ws:
                    self.ws.close()
        else:
            self.logger.error("xunfei.Asr.recv: call_id: %s, Status is not: %s", self.tid, result_dict["code"])
            final_result['name'] = 'TranscriptionResultError'
            final_result['status'] = result_dict["code"]
            registry.ASR_ERRORS.labels(result_dict["code"]).inc()
            if self.ws:
                self.ws.close()
        self.logger.error("xunfei.Asr.recv: call_id: %s, final_result: %s", self.tid, final_result)
        return final_result


class XunfeiAsr:

    STATUS_FIRST_FRAME = 0  # 第一帧的标识
    STATUS_CONTINUE_FRAME = 1  # 中间帧标识
    STATUS_LAST_FRAME = 2  # 最后一帧的标识

    def __init__(self, tid, logger, message_receiver=None):
        self.__id = tid
        self.logger = logger
        self.message_receiver = message_receiver

        self.connected = False
        self.status = XunfeiAsr.STATUS_FIRST_FRAME  # 音频的状态信息，标识音频是第一帧，还是中间帧、最后一帧
        self.pre_result = ""

        self.APPID = '1ec1097b'
        self.APIKey = '7237c29d862daa6fd805f788ed70c409'
        self.APISecret = 'YTk1YzAyMmQwYjQ3ZDJkYWQyZGQwMDFm'
        # 公共参数(common)
        self.CommonArgs = {"app_id": self.APPID}
        # 业务参数(business)，更多个性化参数可在官网查看
        self.BusinessArgs = {"domain": "iat", "language": "zh_cn", "accent": "mandarin", "vinfo":1,"vad_eos":2000, "dwa":"wpgs"}

        # self.__event = threading.Event()
        self.__th = threading.Thread(target=self.__run)

    def create_url(self):
        import hashlib
        import hmac
        from urllib.parse import urlencode
        from wsgiref.handlers import format_date_time
        from time import mktime

        url = 'wss://ws-api.xfyun.cn/v2/iat'
        # 生成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/iat " + "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 __run(self):
        import ssl
        import websocket

        websocket.enableTrace(False)
        try:
            # 测试时候在此处正确填写相关信息即可运行
            time1 = datetime.now()
            ws_url = self.create_url()
            self.logger.info("xunfei.Asr.call_id:%s, ws_url:%s", self.__id, ws_url)
            self.ws = websocket.WebSocketApp(ws_url, on_message=self.on_message, on_error=self.on_error, on_close=self.on_close)
            self.ws.on_open = self.on_open
            self.connected = False
            self.ws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE})
            time_cost = (datetime.now() - time1)
            self.logger.info(f"xunfei.Asr.started. call_id:{self.__id}, timeCost:{time_cost}")
        except Exception as e:
            traceback.print_exc()
            self.logger.info(f"[{self.__id}]ASR session start exception. {e}")

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

    def send_audio(self, audio_data=None):
        if not self.connected:
            self.logger.info('xunfei.Asr.send_audio:ws_is_None: call_id: %s, chunk:%s, %s', self.__id, len(audio_data), audio_data)
            return

        if not audio_data:
            self.status = XunfeiAsr.STATUS_LAST_FRAME
        buf = bytes(audio_data) if audio_data else bytes()
        # self.logger.info('xunfei.Asr.send_audio: call_id: %s, status:%s, chunk:%s, %s', self.__id, status, (len(buf) if buf else 0) , buf)

        # 第一帧处理
        # 发送第一帧音频，带business 参数
        # appid 必须带上，只需第一帧发送
        if self.status == XunfeiAsr.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"}}
            d = json.dumps(d)
            self.ws.send(d)
            self.status = XunfeiAsr.STATUS_CONTINUE_FRAME
        # 中间帧处理
        elif self.status == XunfeiAsr.STATUS_CONTINUE_FRAME:
            d = {"data": {"status": 1, "format": "audio/L16;rate=16000",
                          "audio": str(base64.b64encode(buf), 'utf-8'),
                          "encoding": "raw"}}
            self.ws.send(json.dumps(d))
        # 最后一帧处理
        elif self.status == XunfeiAsr.STATUS_LAST_FRAME:
            d = {"data": {"status": 2, "format": "audio/L16;rate=16000",
                          "audio": str(base64.b64encode(buf), 'utf-8'),
                          "encoding": "raw"}}
            self.ws.send(json.dumps(d))
            time.sleep(1)

    def close(self):
        try:
            self.send_audio()
            self.ws.close()
        except Exception as e:
            self.logger.info(f"[{self.__id}]Error stopping ASR: {e}")

    # 收到websocket连接建立的处理
    def on_open(self, ws):
        self.connected = True
        self.logger.info("xunfei.Asr.open: call_id: %s", self.__id)
        # self.__event.set()

    # 收到websocket消息的处理
    def on_message(self, ws, message):
        try:
            self.logger.info("xunfei.Asr.recv: call_id: %s, message :%s", self.__id, message)
            if self.message_receiver:
                self.message_receiver(self.convert_message(message))
        except Exception as e:
            self.logger.error("receive msg, but parse exception call_id:%s, message:%s, error:%s", self.__id, message, e)

    # 收到websocket错误的处理
    def on_error(self, ws, error):
        self.logger.error("xunfei.Asr.recv::error, call_id:%s, msg:%s", self.__id, error)
        # if not self.__event.is_set():
        #     self.__event.set()

    # 收到websocket关闭的处理
    def on_close(self, ws, a, b):
        self.connected = False
        self.logger.error("xunfei.Asr.recv::close, call_id:%s", self.__id)
        # if not self.__event.is_set():
        #     self.__event.set()

    def convert_message(self, message):
        final_result = {}
        message = json.loads(message)

        if message["code"] == 0:
            data = message["data"]["result"]["ws"]
            result = ""
            for i in data:
                for w in i["cw"]:
                    result += w["w"]

            status = message["data"]["status"]
            if status == 0:
                final_result['name'] = 'SentenceBegin'
            elif status == 1:
                final_result['name'] = 'TranscriptionResultChanged'
                final_result['result'] = result
                self.pre_result = result
            elif status == 2:
                final_result['status'] = 'SentenceEnd'
                final_result['result'] = self.pre_result + result
                self.pre_result = ""
        else:
            final_result['name'] = 'TranscriptionResultError'
            final_result['status'] = message['code']
            final_result['result'] = message['message']
            self.logger.info("call_id:%s, sid:%s call error:%s code is:%s" % (self.__id, message["sid"], message["message"], message["code"]))
            registry.ASR_ERRORS.labels(message['code']).inc()

        self.logger.error("xunfei.Asr.recv: call_id:%s, final_result: %s", self.__id, final_result)
        return final_result