'commit'
This commit is contained in:
269
dsLightRag/KeDaXunFei/XunFeiAudioEvaluator.py
Normal file
269
dsLightRag/KeDaXunFei/XunFeiAudioEvaluator.py
Normal file
@@ -0,0 +1,269 @@
|
||||
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
|
||||
|
||||
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"
|
||||
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)
|
||||
Reference in New Issue
Block a user