PuppyServer/aiDogProject_docker/aiDogProject/aiDogApp/generalRequest.py

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)