idcard_reader_web/apps/limit/biz/year_limit_query.py

#!/usr/bin/python
# -*- coding: UTF-8 -*-
#
# 年度额度查询
# author: Scott Chen
# date: 2019-12-13

import json
import time
import requests
from apps.limit.biz.signature import Signature

with open('apps/limit/biz/config.json', 'r', encoding='utf-8') as json_file:
    config_json = json.load(json_file)


class YearLimitQuery:
    def __init__(self, name, id_no):
        self.__name = name
        self.__id_no = id_no
        self.__secret_key = ""

    def __build_merch_info(self):
        merch_param = {}
        merch_param["merchId"] = config_json["merchId"]
        merch_param["merchName"] = config_json["merchName"]
        merch_param["thirdPartyMerchCode"] = config_json["thirdPartyMerchCode"]
        merch_param["thirdPartyMerchName"] = config_json["thirdPartyMerchName"]
        merch_param["name"] = self.__name
        merch_param["idCard"] = self.__id_no
        merch_param["isCheckIdCard"] = config_json["isCheckIdCard"]

        self.__secret_key = config_json["secret_key"]
        print("原始参数集：{}".format(merch_param))
        return merch_param

    def __build_req_params(self):
        merch_param = self.__build_merch_info()
        req_params = {
            "merchId": merch_param["merchId"],
            "data": str(json.dumps(merch_param)),
            "timestamp": str(int(time.time()))
        }
        return req_params

    def __sign(self):
        req_params = self.__build_req_params()
        instance = Signature(req_params, self.__secret_key)
        ret_sign = instance.sign_with_md5()
        return ret_sign

    # 额度查询请求并返回结果
    def __req(self, req_params):
        result = {}
        try:
            req_url = config_json["year_limit_url"]
            # GET方法，必须是params=str(dict or list or tuple or bytes)
            # POST方法，必须是data=dict or list or tuple or bytes or file
            headers = { "Content-Type": "application/json;charset=UTF-8" }
            response = requests.request("POST", req_url, headers=headers, data=json.dumps(req_params), timeout=20)
            if response.text:
                result = json.loads(response.text)
                print("API返回数据：{}".format(result))
                result = self.__parse_response(result)
                # 返回额度查询结果
                return result
            else:
                print("response.text is empty")
                result['error'] = '[20000]请求查询API失败'
                return result
        except Exception as e:
            print(e)
            result['error'] = '[20001]请求查询API异常'
            return result

    # 额度查询数据处理
    def __parse_response(self, data):
        result = {}
        # 处理返回数据
        if data["code"] == '0':
            limit_data = data["data"]["rows"][0]
            if str(limit_data["success"]).lower() == "true":
                result["totalAmount"] = limit_data["result"]["totalAmount"]
                result["innerbalance"] = limit_data["result"]["innerbalance"]
                print("【{}，{}】查询额度为：{}".format(self.__name, self.__id_no, data))
            else:
                print("海关返回【{}，{}】额度数据不成功".format(self.__name, self.__id_no))
                result['error'] = '[20002]查询额度不成功'
        else:
            print("【{}，{}】额度查询失败：{}".format(self.__name, self.__id_no, data["msg"]))
            result['error'] = '[20003]额度查询失败'
        return result

    def query(self):
        ret = self.__sign()
        return self.__req(ret)


# 单元测试
if __name__ == '__main__':
    yq = YearLimitQuery("张三", "xxxxx")
    yq.query()