dsProject/dsLightRag/KeDaXunFei/XunFeiAudioEvaluator.py

import base64
import datetime
import hashlib
import hmac
import json
import ssl
import time
import xml.etree.ElementTree as ET
from datetime import datetime
from time import mktime
from urllib.parse import urlencode
from wsgiref.handlers import format_date_time
import websocket

from Config.Config import XF_APPID, XF_APISECRET, XF_APIKEY


class XunFeiAudioEvaluator:
    """讯飞语音评测类"""
    
    def __init__(self, appid, api_key, api_secret, audio_file):
        self.appid = appid
        self.api_key = api_key
        self.api_secret = api_secret
        self.audio_file = audio_file
        self.host_url = "ws://ise-api.xfyun.cn/v2/open-ise"
        self.websocket_url = ""
        self.evaluation_results = {}
    
    def generate_auth_url(self):
        """生成鉴权URL"""
        now_time = datetime.now()
        now_date = format_date_time(mktime(now_time.timetuple()))
        
        origin_base = "host: " + "ise-api.xfyun.cn" + "\n"
        origin_base += "date: " + now_date + "\n"
        origin_base += "GET " + "/v2/open-ise " + "HTTP/1.1"
        
        # SHA256加密
        signature_sha = hmac.new(
            self.api_secret.encode('utf-8'), 
            origin_base.encode('utf-8'),
            digestmod=hashlib.sha256
        ).digest()
        signature_sha = base64.b64encode(signature_sha).decode(encoding='utf-8')
        
        authorization_origin = f'api_key="{self.api_key}", algorithm="hmac-sha256", headers="host date request-line", signature="{signature_sha}"'
        authorization = base64.b64encode(authorization_origin.encode('utf-8')).decode(encoding='utf-8')
        
        dict_data = {
            "authorization": authorization,
            "date": now_date,
            "host": "ise-api.xfyun.cn"
        }
        
        return self.host_url + '?' + urlencode(dict_data)
    
    def on_message(self, ws, message):
        """WebSocket消息处理"""
        print(f"Received message: {message}")
        data = json.loads(message)
        status = data["data"]["status"]
        
        if status == 2:
            # 解析评测结果
            xml_data = base64.b64decode(data["data"]["data"])
            xml_content = xml_data.decode("utf-8")
            #print(xml_content)
            
            # 解析XML并提取得分信息
            self.parse_evaluation_results(xml_content)
            ws.close()
    
    def on_error(self, ws, error):
        """错误处理"""
        print(f"Error: {error},{ws}")
    
    def on_close(self, ws, reason, res):
        """连接关闭处理"""
        print(f"WebSocket connection closed,{ws}")
    
    def on_open(self, ws):
        """连接建立处理"""
        print(f"WebSocket connection opened,{ws},ws连接建立成功...")
        
        # 发送初始参数
        send_dict = {
            "common": {
                "app_id": self.appid
            },
            "business": {
                "category": "read_sentence",
                "rstcd": "utf8",
                "sub": "ise",
                "group": "pupil",
                "ent": "en_vip",
                "tte": "utf-8",
                "cmd": "ssb",
                "auf": "audio/L16;rate=16000",
                "aue": "lame",
                "text": '\uFEFF' + "[content]\nnice to meet you."
            },
            "data": {
                "status": 0,
                "data": ""
            }
        }
        ws.send(json.dumps(send_dict))
        
        # 发送音频数据
        with open(self.audio_file, "rb") as file_flag:
            while True:
                buffer = file_flag.read(1280)
                if not buffer:
                    # 发送最后一帧
                    my_dict = {
                        "business": {
                            "cmd": "auw", 
                            "aus": 4, 
                            "aue": "lame"
                        },
                        "data": {
                            "status": 2, 
                            "data": str(base64.b64encode(buffer).decode())
                        }
                    }
                    ws.send(json.dumps(my_dict))
                    #print("发送最后一帧")
                    time.sleep(1)
                    break
                
                # 发送中间音频帧
                send_dict = {
                    "business": {
                        "cmd": "auw",
                        "aus": 1,
                        "aue": "lame"
                    },
                    "data": {
                        "status": 1,
                        "data": str(base64.b64encode(buffer).decode()),
                        "data_type": 1,
                        "encoding": "raw"
                    }
                }
                ws.send(json.dumps(send_dict))
                time.sleep(0.04)
    
    def parse_evaluation_results(self, xml_content):
        """解析评测结果XML并提取得分信息"""
        try:
            root = ET.fromstring(xml_content)
            
            # 查找read_chapter节点
            read_chapter = root.find('.//read_chapter')
            if read_chapter is not None:
                self.evaluation_results = {
                    'accuracy_score': float(read_chapter.get('accuracy_score', 0)),
                    'fluency_score': float(read_chapter.get('fluency_score', 0)),
                    'integrity_score': float(read_chapter.get('integrity_score', 0)),
                    'standard_score': float(read_chapter.get('standard_score', 0)),
                    'total_score': float(read_chapter.get('total_score', 0)),
                    'word_count': int(read_chapter.get('word_count', 0)),
                    'is_rejected': read_chapter.get('is_rejected', 'false') == 'true'
                }
                
                # 提取句子级别得分
                sentence = read_chapter.find('.//sentence')
                if sentence is not None:
                    self.evaluation_results['sentence'] = {
                        'accuracy_score': float(sentence.get('accuracy_score', 0)),
                        'fluency_score': float(sentence.get('fluency_score', 0)),
                        'total_score': float(sentence.get('total_score', 0))
                    }
                
                # 提取单词级别得分
                words = []
                for word in read_chapter.findall('.//word'):
                    word_data = {
                        'content': word.get('content', ''),
                        'total_score': float(word.get('total_score', 0)),
                        'dp_message': int(word.get('dp_message', 0))
                    }
                    words.append(word_data)
                
                self.evaluation_results['words'] = words
                
        except ET.ParseError as e:
            print(f"XML解析错误: {e}")
    
    def get_evaluation_summary(self):
        """获取评测结果摘要"""
        if not self.evaluation_results:
            return "暂无评测结果"
        
        summary = "=== 语音评测结果摘要 ===\n"
        summary += f"总得分: {self.evaluation_results.get('total_score', 0):.4f}\n"
        summary += f"准确度得分: {self.evaluation_results.get('accuracy_score', 0):.4f}\n"
        summary += f"流畅度得分: {self.evaluation_results.get('fluency_score', 0):.4f}\n"
        summary += f"完整度得分: {self.evaluation_results.get('integrity_score', 0):.4f}\n"
        summary += f"标准度得分: {self.evaluation_results.get('standard_score', 0):.4f}\n"
        summary += f"单词数量: {self.evaluation_results.get('word_count', 0)}\n"
        summary += f"是否被拒绝: {'是' if self.evaluation_results.get('is_rejected', False) else '否'}\n"
        
        if 'sentence' in self.evaluation_results:
            sentence = self.evaluation_results['sentence']
            summary += f"\n=== 句子级别得分 ===\n"
            summary += f"句子准确度: {sentence.get('accuracy_score', 0):.4f}\n"
            summary += f"句子流畅度: {sentence.get('fluency_score', 0):.4f}\n"
            summary += f"句子总分: {sentence.get('total_score', 0):.4f}\n"
        
        if 'words' in self.evaluation_results:
            summary += f"\n=== 单词级别得分 ===\n"
            for i, word in enumerate(self.evaluation_results['words']):
                dp_msg = self._get_dp_message_description(word['dp_message'])
                summary += f"{i+1}. {word['content']}: {word['total_score']:.4f} ({dp_msg})\n"
        
        return summary
    
    def _get_dp_message_description(self, dp_message):
        """获取dp_message描述"""
        descriptions = {
            0: "正常",
            16: "漏读",
            32: "增读", 
            64: "回读",
            128: "替换"
        }
        return descriptions.get(dp_message, f"未知({dp_message})")
    
    def run_evaluation(self):
        """运行评测"""
        start_time = datetime.now()
        websocket.enableTrace(False)
        
        ws_url = self.generate_auth_url()
        ws_entity = websocket.WebSocketApp(
            ws_url, 
            on_message=self.on_message, 
            on_error=self.on_error, 
            on_close=self.on_close,
            on_open=self.on_open
        )
        
        ws_entity.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE})
        
        end_time = datetime.now()
        evaluation_time = end_time - start_time
        print(f"评测耗时： {evaluation_time}")
        
        return self.evaluation_results, evaluation_time


# 使用示例
if __name__ == '__main__':
    # 配置参数
    # appid = "5b83f8d6"
    # api_secret = "604fa6cb9c5ab664a0d153fe0ccc6802"
    # api_key = "5beb887923204000bfcb402046bb05a6"
    appid = XF_APPID
    api_secret = XF_APISECRET
    api_key = XF_APIKEY
    audio_file = "./1.mp3"
    
    # 创建评测器实例
    evaluator = XunFeiAudioEvaluator(appid, api_key, api_secret, audio_file)
    
    # 运行评测
    results, eval_time = evaluator.run_evaluation()
    
    # 输出评测结果摘要
    print("\n" + "="*50)
    print(evaluator.get_evaluation_summary())
    print(f"总评测时间: {eval_time}")
    print("="*50)