import base64 import hashlib import hmac import json from datetime import datetime from http import HTTPMethod from time import mktime from urllib.parse import urlencode from wsgiref.handlers import format_date_time import requests # 填写在开放平台申请的APPID、APIKey、APISecret # 相应编码音频base64编码后数据(不超过4M) 说话人辨认(Speaker Identification)和说话人确认(Speaker Verification) class Gen_req_url(object): """生成请求的url""" def sha256base64(self, data): sha256 = hashlib.sha256() sha256.update(data) digest = base64.b64encode(sha256.digest()).decode(encoding='utf-8') return digest def parse_url(self, requset_url): stidx = requset_url.index("://") host = requset_url[stidx + 3:] # self.schema = requset_url[:stidx + 3] edidx = host.index("/") if edidx <= 0: raise Exception("invalid request url:" + requset_url) self.path = host[edidx:] self.host = host[:edidx] # build websocket auth request url def assemble_ws_auth_url(self, requset_url, api_key, api_secret, method="GET"): self.parse_url(requset_url) now = datetime.now() date = format_date_time(mktime(now.timetuple())) # date = "Thu, 12 Dec 2019 01:57:27 GMT" signature_origin = "host: {}\ndate: {}\n{} {} HTTP/1.1".format(self.host, date, method, self.path) signature_sha = hmac.new(api_secret.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\"" % ( api_key, "hmac-sha256", "host date request-line", signature_sha) authorization = base64.b64encode(authorization_origin.encode('utf-8')).decode(encoding='utf-8') values = { "host": self.host, "date": date, "authorization": authorization } return requset_url + "?" + urlencode(values) def gen_req_body(apiname, APPId,groupId,featureId,file_path=None): """ 生成请求的body :param apiname :param APPId: Appid :param file_name: 文件路径 :return: """ if apiname == 'createFeature': with open(file_path, "rb") as f: audioBytes = f.read() body = { "header": { "app_id": APPId, "status": 3 }, # 定义参数，解密返回值后比对一致 "parameter": { "s782b4996": { "func": "createFeature", "groupId": groupId, "featureId": featureId, "featureInfo": "aiDog_featureInfo_".featureId, "createFeatureRes": { "encoding": "utf8", "compress": "raw", "format": "json" } } }, "payload": { "resource": { "encoding": "lame", "sample_rate": 16000, "channels": 1, "bit_depth": 16, "status": 3, "audio": str(base64.b64encode(audioBytes), 'UTF-8') } } } elif apiname == 'createGroup': body = { "header": { "app_id": APPId, "status": 3 }, "parameter": { "s782b4996": { "func": "createGroup", "groupId": groupId, "groupName": "aiDog_groupName_".groupId, "groupInfo": "aiDog_groupInfo_".groupId, "createGroupRes": { "encoding": "utf8", "compress": "raw", "format": "json" } } } } elif apiname == 'deleteFeature': body = { "header": { "app_id": APPId, "status": 3 }, "parameter": { "s782b4996": { "func": "deleteFeature", "groupId": groupId, "featureId": featureId, "deleteFeatureRes": { "encoding": "utf8", "compress": "raw", "format": "json" } } } } elif apiname == 'queryFeatureList': body = { "header": { "app_id": APPId, "status": 3 }, "parameter": { "s782b4996": { "func": "queryFeatureList", "groupId": groupId, "queryFeatureListRes": { "encoding": "utf8", "compress": "raw", "format": "json" } } } } elif apiname == 'searchFea': with open(file_path, "rb") as f: audioBytes = f.read() body = { "header": { "app_id": APPId, "status": 3 }, "parameter": { "s782b4996": { "func": "searchFea", "groupId": groupId, "topK": 1, "searchFeaRes": { "encoding": "utf8", "compress": "raw", "format": "json" } } }, "payload": { "resource": { "encoding": "lame", "sample_rate": 16000, "channels": 1, "bit_depth": 16, "status": 3, "audio": str(base64.b64encode(audioBytes), 'UTF-8') } } } elif apiname == 'searchScoreFea': with open(file_path, "rb") as f: audioBytes = f.read() body = { "header": { "app_id": APPId, "status": 3 }, "parameter": { "s782b4996": { "func": "searchScoreFea", "groupId": groupId, "dstFeatureId": featureId, "searchScoreFeaRes": { "encoding": "utf8", "compress": "raw", "format": "json" } } }, "payload": { "resource": { "encoding": "lame", "sample_rate": 16000, "channels": 1, "bit_depth": 16, "status": 3, "audio": str(base64.b64encode(audioBytes), 'UTF-8') } } } elif apiname == 'updateFeature': with open(file_path, "rb") as f: audioBytes = f.read() body = { "header": { "app_id": APPId, "status": 3 }, "parameter": { "s782b4996": { "func": "updateFeature", "groupId": groupId, "featureId": featureId, "featureInfo": "aiDog_featureInfo_".featureId, "updateFeatureRes": { "encoding": "utf8", "compress": "raw", "format": "json" } } }, "payload": { "resource": { "encoding": "lame", "sample_rate": 16000, "channels": 1, "bit_depth": 16, "status": 3, "audio": str(base64.b64encode(audioBytes), 'UTF-8') } } } elif apiname == 'deleteGroup': body = { "header": { "app_id": APPId, "status": 3 }, "parameter": { "s782b4996": { "func": "deleteGroup", "groupId": groupId, "deleteGroupRes": { "encoding": "utf8", "compress": "raw", "format": "json" } } } } else: raise Exception( "输入的apiname不在[createFeature, createGroup, deleteFeature, queryFeatureList, searchFea, searchScoreFea,updateFeature]内，请检查") return body def req_url(api_name, APPId, APIKey, APISecret,featureId,groupId,file_path=None): """ 开始请求 :param APPId: APPID :param APIKey: APIKEY :param APISecret: APISecret :param file_path: body里的文件路径 :return: """ res = { "code":"", "status": "", "msg": "", "response_time": "" } gen_req_url = Gen_req_url() body = gen_req_body(apiname=api_name, APPId=APPId, file_path=file_path,featureId=featureId,groupId=groupId) request_url = gen_req_url.assemble_ws_auth_url(requset_url='https://api.xf-yun.com/v1/private/s782b4996', method="POST", api_key=APIKey, api_secret=APISecret) headers = {'content-type': "application/json", 'host': 'api.xf-yun.com', 'appid': '$APPID'} response = requests.post(request_url, data=json.dumps(body), headers=headers) tempResult = json.loads(response.content.decode('utf-8')) # print(tempResult) res['code'] = tempResult.get('header').get('code') if res['code'] ==0: fb = tempResult.get('payload').get(api_name + 'Res').get('text') res['status'] = tempResult.get('header').get('message') res["msg"] = str(base64.b64decode(fb), 'UTF-8') res['response_time'] =datetime.strftime(datetime.today(), '%Y-%m-%d %H:%M:%S') print(tempResult) print(str(base64.b64decode(fb),'UTF-8')) print(res) return res else: res['status'] = tempResult.get('header').get('message') res["msg"] = 'you wrong!' res['response_time'] = datetime.strftime(datetime.today(), '%Y-%m-%d %H:%M:%S') return res """ * 1.声纹识别接口,请填写在讯飞开放平台-控制台-对应能力页面获取的APPID、APIKey、APISecret * 2.groupId要先创建,然后再在createFeature里使用,不然会报错23005,修改时需要注意保持统一 * 3.音频base64编码后数据(不超过4M),音频格式需要16K、16BIT的MP3音频。 * 4.主函数只提供调用示例,其他参数请到对应类去更改,以适应实际的应用场景。 """ if __name__ == '__main__': APPId = "4522d502" APISecret = "Zjc5MjJhMzBmMDYxNTg4MTNlMTg1MmQw" APIKey = "f7a9f0ceae3ff7ebfc0c89efeb18810d" # file_path = '示例音频/讯飞开放平台.mp3'#读取数据库路径下音源 file_path='audio_cache/voice_JTYNZP9O_1726111463.mp3' # 执行逻辑 # 先从客户端获取音频，使用音频文件格式转换工具ffmpeg # 并根据uid生成音频特征(一个名字对应一个特征)，提交后创建特征库并添加特征id，（每个用户的8个声音指令=8个特征id，所有数据都在特征列表中） # 同时该音频存储到数据库对应uid下，下次提交音频的时候，根据用户id查询存储在数据库的音纹， # 提交数据库的音纹给检索出特征列表判断是否有数据，如有，则判断新音纹执行特征比对方法，返回结果中，取出text解密出得分0.0~1.0，给到前端做逻辑判断执行 # 用户移除，则删掉对应音纹 # apiname取值: # 1.创建声纹特征库 createGroup # req_url(api_name='createGroup', APPId=APPId, # APIKey=APIKey, APISecret=APISecret, file_path=file_path) # 2.添加音频特征 createFeature # req_url(api_name='createFeature', APPId=APPId, # APIKey=APIKey, APISecret=APISecret, file_path=file_path) # 3.查询特征列表 queryFeatureList # req_url(api_name='queryFeatureList', APPId=APPId, # APIKey=APIKey, APISecret=APISecret, file_path=file_path) # 4.特征比对1:1 searchScoreFea req_url(api_name='searchScoreFea', APPId=APPId, APIKey=APIKey, APISecret=APISecret, file_path=file_path) # 5.特征比对1:N searchFea # req_url(api_name='searchFea', APPId=APPId, # APIKey=APIKey, APISecret=APISecret, file_path=file_path) # 6.更新音频特征 updateFeature # req_url(api_name='updateFeature', APPId=APPId, # APIKey=APIKey, APISecret=APISecret, file_path=file_path) # 7.删除指定特征 deleteFeature # req_url(api_name='deleteFeature', APPId=APPId, # APIKey=APIKey, APISecret=APISecret, file_path=file_path) # 8.删除声纹特征库 deleteGroup # req_url(api_name='deleteGroup', APPId=APPId, # APIKey=APIKey, APISecret=APISecret, file_path=file_path)