python3 Mock开发

  1 import datetime
  2 import json
  3 import re
  4 import pymongo
  5 import binascii
  6 from Crypto.Cipher import DES
  7 
  8 from flask import Flask, request, Response
  9 
 10 app = Flask(__name__)
 11 
 12 app.config[‘JSON_AS_ASCII‘] = False
 13 
 14 
 15 def des_decode(data, key):
 16     """aes解密
 17     :param key:
 18     :param data:
 19     """
 20     cipher = DES.new(key)
 21     result2 = binascii.a2b_hex(data)  # 十六进制还原成二进制
 22     decrypted = cipher.decrypt(result2)
 23 
 24     return decrypted
 25 
 26 
 27 def mongodb_connect():
 28     """创建数据库及集合
 29     """
 30     myclient = pymongo.MongoClient("mongodb://1.1.1.1:20/")
 31     mydb = myclient["test"]
 32     mycol = mydb["test"]
 33 
 34     return mycol
 35 
 36 
 37 def mongodb_update_apply(*args):
 38     """根据申请信息更新表中的数据
 39     """
 40     mycol = mongodb_connect()
 41     mycol.update({"tran_no": args[0]},
 42                  {"$set": {"code": args[1], "data": args[2], "msg": args[3], "apply_status": args[4]}})
 43 
 44 
 45 def mongodb_update_verification(*args):
 46     """根据确认信息更新表中的数据
 47     """
 48     mycol = mongodb_connect()
 49     mycol.update({"tran_no": args[0]},
 50                  {"$set": {"code": args[1], "msg": args[2], "apply_status": args[3]}})
 51 
 52 
 53 def apply_info(tran_no, code, data, msg, apply_status):
 54     """申请响应结果
 55     :param tran_no:
 56     :param msg:
 57     :param data:
 58     :param code:
 59     :param apply_status:
 60     """
 61     result = {
 62         "code": code,
 63         "data": data,
 64         "msg": msg
 65     }
 66 
 67     mongodb_update_apply(tran_no, code, data, msg, apply_status)
 68 
 69     return result
 70 
 71 
 72 
 73 @app.route(‘/bxdb/xmbksq‘, methods=[‘POST‘, ‘GET‘])
 74 def bxdb_xmbksq():
 75     """签约接口"""
 76     json_data = request.form
 77     data = json.loads(json_data[‘uploadData‘])[‘data‘]
 78     data1 = des_decode(data, ‘123456‘)
 79     data2 = re.findall(‘({.+})‘, data1.decode(‘utf-8‘))
 80     request_data = json.loads(data2[0])
 81 
 82     tran_no = request_data[‘tranno‘]
 83     name = request_data[‘name‘]
 84     id_no = request_data[‘idNo‘]
 85     bank_card_no = request_data[‘bankCardNo‘]
 86     mobile = request_data[‘mobile‘]
 87 
 88     date_str = datetime.datetime.now().strftime(‘%Y-%m-%d %H:%M:%S‘)
 89 
 90     mycol = mongodb_connect()
 91 
 92     mongodb_result = mycol.find_one(
 93         {"id_no": id_no, "name": name, "bank_card_no": bank_card_no, "mobile": mobile})
 94 
 95     if mongodb_result:
 96         apply_status = mongodb_result[‘apply_status‘]
 97         if apply_status != ‘COMPLETED‘:
 98             mycol.update(
 99                 {{"id_no": id_no, "name": name, "bank_card_no": bank_card_no, "mobile": mobile}},
100                 {"$set": {"tran_no": tran_no}})
101             result = apply_info(tran_no, ‘0‘, ‘01‘, ‘交易成功‘, ‘SUBMITTED‘)
102         else:
103             result = apply_info(tran_no, ‘0‘, ‘02‘, ‘该卡已签约绑定‘, ‘COMPLETED‘)
104     else:
105         mydict = {"tran_no": tran_no, "id_no": id_no, "name": name, "bank_card_no": bank_card_no, "mobile": mobile,
106                   "last_modified_time": datetime.datetime.strptime(date_str, ‘%Y-%m-%d %H:%M:%S‘),
107                   "created_time": datetime.datetime.strptime(date_str, ‘%Y-%m-%d %H:%M:%S‘)}
108         mycol.insert_one(mydict)
109 
110         if len(mobile) > 11:
111             result = apply_info(tran_no, ‘1‘, ‘‘, ‘银行预留手机号不正确！‘, ‘FAILED‘)
112         else:
113             result = apply_info(tran_no, ‘0‘, ‘01‘, ‘交易成功‘, ‘SUBMITTED‘)
114 
115     response = Response(json.dumps(result, ensure_ascii=False), mimetype=‘application/json‘)
116 
117     return response
118 
119 
120 if __name__ == ‘__main__‘:
121     app.run(host=‘0.0.0.0‘, port=9015)