#!/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()