Python3 微信支付(小程序支付)V3接口的实现

mchid = "xxxxxx"                         # 商户号 pay_key = "xxxxxx"                       # 商户秘钥V3 使用V3接口必须使用V3秘钥 serial_num = "xxxxxx"                    # 证书序列号 # ======================前三个参数在微信支付中可找到=============================== # ============ 商户号(mchid ) 在账户中心——商户信息——微信支付商户号 （是纯数字） ================== # ============= 商户秘钥(pay_key) 在账户中心——API安全——APIv3秘钥 （需手动设置） =================== # ============= 证书序列号(serial_num) 在账户中心——API安全——API证书 （需手动申请,通过后会有串证书序列号），申请完成后需要把证书下载到项目中，便于使用 =================== appid = "xxxxxx"                         # 微信小程序appid wx_secret ="xxxxxx"                      # 微信小程序秘钥 # ============= 微信小程序appid 在产品中心——AppID账号管理——添加关联的AppID  =================== WX_Pay_URL = "https://api.mch.weixin.qq.com/v3/pay/transactions/jsapi" # ============= 微信支付调用地址,用于请求接收 预支付交易会话标识: prepay_id =================== WX_Notify_URL = "https://127.0.0.1:8000" # ============= 接收微信支付回调地址,必须是https ===================         

import json import decimal import traceback import requests from django.http import HttpResponse def payment_view(request, *args, **kwargs): """ 微信支付（小程序） :param request: :param args: :param kwargs: :return: """ try: reqdata = json.loads(request.body) # 前端参数 jscode = reqdata["jscode"]  # 微信ID price = decimal.Decimal(reqdata["price"]).quantize(decimal.Decimal("0.00"))  # 充值金额,保留两位小数 nickname = reqdata["nickname"]  # 微信昵称/支付宝名称 前端获取到返给后端做记录,可要可不要的字段 paymode = reqdata["paymode"]  # 支付方式  1微信支付 remark = reqdata["remark"]  # 支付内容描述 # 根据jscode 获取openID rets = requests.get(url = "https://api.weixin.qq.com/sns/jscode2session?" \ "appid=%s&secret=%s&js_code=%s" \ "&grant_type=authorization_code" % (appid,wx_secret, js_code), timeout=3, verify=False) if not rets: return HttpResponse(general_error_msg(msg="未获取到微信信息")) # 0.获取支付的微信openid print(f"组织ID:{userinfo['orgid']}, jscode:{jscode}") wxuser = getappopenid(orgid, jscode) if wxuser: # session_key = wxuser["session_key"] openid = wxuser["openid"] else: return HttpResponse(general_error_msg(msg="未获取到微信用户信息")) # 1.以交易日期生成交易号 orderno = order_num() # 2.生成新交易记录 paystatus 支付状态  1成功 0待支付 -1支付失败 conorder.objects.create(orderno=orderno, openid=openid, openname=nickname, paymode=paymode,goodstotalprice=price, paystatus=0, remark=remark,createtime=get_now_time(1)) # 3.生成统一下单的报文body url = WX_Pay_URL body = { "appid": appid, "mchid": mchid, "description": remark, "out_trade_no": orderno, "notify_url": WX_Notify_URL + "/pay/notify",  # 后端接收回调通知的接口 "amount": {"total": int(price * 100), "currency": "CNY"},  # 正式上线price要*100，微信金额单位为分（必须整型）。 "payer": {"openid": openid}, } data = json.dumps(body) headers, random_str, time_stamps = make_headers_v3(mchid, serial_num, data=data, method='POST') # 10.发送请求获得prepay_id try: response = requests.post(url, data=data, headers=headers)  # 获取预支付交易会话标识（prepay_id） print("预支付交易会话标识", response) if response.status_code == 200: wechatpay_serial, wechatpay_timestamp, wechatpay_nonce, wechatpay_signature, certificate, serial_no = check_wx_cert( response, mchid, pay_key, serial_num) # 11.9签名验证 if wechatpay_serial == serial_no:  # 应答签名中的序列号同证书序列号应相同 print('serial_no match') try: data3 = f"{wechatpay_timestamp}\n{wechatpay_nonce}\n{response.text}\n" verify(data3, wechatpay_signature, certificate) print('The signature is valid.') # 12.生成调起支付API需要的参数并返回前端 res = { 'orderno': orderno,  # 订单号 'timeStamp': time_stamps, 'nonceStr': random_str, 'package': 'prepay_id=' + response.json()['prepay_id'], 'signType': "RSA", 'paySign': get_sign(f"{appid}\n{time_stamps}\n{random_str}\n{'prepay_id=' + response.json()['prepay_id']}\n"), } return HttpResponse(success_msg(msg="下单成功", total=0, data=res)) except Exception as e: log.error(f"证书序列号验签失败{e}, {traceback.format_exc()}") return HttpResponse(general_error_msg(msg="下单失败")) else: log.error(f"证书序列号比对失败【请求头中证书序列号：{wechatpay_serial}；本地存储证书序列号：{serial_no}；】") return HttpResponse(general_error_msg(msg="调起微信支付失败!")) else: log.error(f"获取预支付交易会话标识 接口报错【params：{data}；headers：{headers}；response：{response.text}】") return HttpResponse(general_error_msg(msg="调起微信支付失败!")) except Exception as e: log.error(f"调用微信支付接口超时【params：{data}；headers：{headers}；】：{e},{traceback.format_exc()}") return HttpResponse(general_error_msg(msg="微信支付超时!")) except Exception as e: log.error(f"微信支付接口报错：{e},{traceback.format_exc()}") return HttpResponse(general_error_msg(msg="微信支付接口报错!"))

import base64 import random import string import time import traceback from datetime import datetime import requests from BaseMethods.log import log from Crypto.PublicKey import RSA from Crypto.Signature import pkcs1_15 from Cryptodome.Hash import SHA256 from sqlalchemy.util import b64encode from cryptography.hazmat.primitives.ciphers.aead import AESGCM # 各包版本 # django-ratelimit==3.0.1 # SQLAlchemy~=1.4.44 # pycryptodome==3.16.0 # pycryptodomex==3.16.0 # cryptography~=38.0.4 # Django~=3.2.4 # 获取唯一标识 def get_uuid(utype=0): """ 唯一码 :param utype: :return: """ if utype == 0: return uuid.uuid1() elif utype == 1: return str(uuid.uuid1()) elif utype == 2: return str(uuid.uuid1().hex) elif utype == 3: return str((uuid.uuid5(uuid.NAMESPACE_DNS, str(uuid.uuid1()) + str(random.random())))) # 获取当前时间 def get_now_time(type=0): """ :param type: 类型0-5 :return: yyyy-mm-dd HH:MM:SS;y-m-d H:M:S.f;y-m-d;ymdHMS;y年m月d日h时M分S秒 """ if type == 0: return datetime.datetime.now() elif type == 1: return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") elif type == 2: return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f") elif type == 3: return datetime.datetime.now().strftime("%Y-%m-%d") elif type == 4: return datetime.datetime.now().strftime("%Y%m%d%H%M%S") elif type == 5: locale.setlocale(locale.LC_CTYPE, 'chinese') timestr = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") t = time.strptime(timestr, "%Y-%m-%d %H:%M:%S") result = (time.strftime("%Y年%m月%d日%H时%M分%S秒", t)) return result elif type == 6: return datetime.datetime.now().strftime("%Y%m%d") # 重构系统jargon类，用于处理时间格式报错问题 class DateEncoder(json.JSONEncoder): def default(self, obj): if isinstance(obj, datetime.datetime): return obj.strftime('%Y-%m-%d %H:%M:%S') elif isinstance(obj, datetime.date): return obj.strftime("%Y-%m-%d") elif isinstance(obj, Decimal): return float(obj) elif isinstance(obj, bytes): return str(obj, encoding='utf-8') elif isinstance(obj, uuid.UUID): return str(obj) elif isinstance(obj, datetime.time): return obj.strftime('%H:%M') elif isinstance(obj, datetime.timedelta): return str(obj) else: return json.JSONEncoder.default(self, obj) def decrypt(nonce, ciphertext, associated_data, pay_key): """ AES解密 :param nonce: :param ciphertext: :param associated_data: :param pay_key: :return: """ key = pay_key key_bytes = str.encode(key) nonce_bytes = str.encode(nonce) ad_bytes = str.encode(associated_data) data = base64.b64decode(ciphertext) aesgcm = AESGCM(key_bytes) return aesgcm.decrypt(nonce_bytes, data, ad_bytes) def order_num(): """ 生成订单号 :return: """ # 下单时间的年月日毫秒12+随机数8位 now_time = datetime.now() result = str(now_time.year) + str(now_time.month) + str(now_time.day) + str(now_time.microsecond) + str( random.randrange(10000000, 99999999)) return result def get_sign(sign_str): """ 定义生成签名的函数 :param sign_str: :return: """ try: with open(r'static/cret/apiclient_key.pem') as f: private_key = f.read() rsa_key = RSA.importKey(private_key) signer = pkcs1_15.new(rsa_key) digest = SHA256.new(sign_str.encode('utf-8')) # sign = b64encode(signer.sign(digest)).decode('utf-8') sign = b64encode(signer.sign(digest)) return sign except Exception as e: log.error("生成签名的函数方法报错【func：get_sign；sign_str：%s】：%s ==> %s" % (sign_str, e, traceback.format_exc())) def check_wx_cert(response, mchid, pay_key, serial_no): """ 微信平台证书 :param response: 请求微信支付平台所对应的的接口返回的响应值 :param mchid: 商户号 :param pay_key: 商户号秘钥 :param serial_no: 证书序列号 :return: """ wechatpay_serial, wechatpay_timestamp, wechatpay_nonce, wechatpay_signature, certificate = None, None, None, None, None try: # 11.应答签名验证 wechatpay_serial = response.headers['Wechatpay-Serial']  # 获取HTTP头部中包括回调报文的证书序列号 wechatpay_signature = response.headers['Wechatpay-Signature']  # 获取HTTP头部中包括回调报文的签名 wechatpay_timestamp = response.headers['Wechatpay-Timestamp']  # 获取HTTP头部中包括回调报文的时间戳 wechatpay_nonce = response.headers['Wechatpay-Nonce']  # 获取HTTP头部中包括回调报文的随机串 # 11.1.获取微信平台证书 （等于又把前面的跑一遍，实际上应是获得一次证书就存起来，不用每次都重新获取一次） url2 = "https://api.mch.weixin.qq.com/v3/certificates" # 11.2.生成证书请求随机串 random_str2 = ''.join(random.choice(string.ascii_uppercase + string.digits) for _ in range(32)) # 11.3.生成证书请求时间戳 time_stamps2 = str(int(time.time())) # 11.4.生成请求证书的签名串 data2 = "" sign_str2 = f"GET\n{'/v3/certificates'}\n{time_stamps2}\n{random_str2}\n{data2}\n" # 11.5.生成签名 sign2 = get_sign(sign_str2) # 11.6.生成HTTP请求头 headers2 = { "Content-Type": "application/json", "Accept": "application/json", "Authorization": 'WECHATPAY2-SHA256-RSA2048 ' + f'mchid="{mchid}",nonce_str="{random_str2}",signature="{sign2}",timestamp="{time_stamps2}",serial_no="{serial_no}"' } # 11.7.发送请求获得证书 response2 = requests.get(url2, headers=headers2)  # 只需要请求头 cert = response2.json() # 11.8.证书解密 nonce = cert["data"][0]['encrypt_certificate']['nonce'] ciphertext = cert["data"][0]['encrypt_certificate']['ciphertext'] associated_data = cert["data"][0]['encrypt_certificate']['associated_data'] serial_no = cert["data"][0]['serial_no'] certificate = decrypt(nonce, ciphertext, associated_data, pay_key) except Exception as e: log.error(f"微信平台证书验证报错：{e}；{traceback.format_exc()}") return wechatpay_serial, wechatpay_timestamp, wechatpay_nonce, wechatpay_signature, certificate, serial_no def verify(check_data, signature, certificate): """ 验签函数 :param check_data: :param signature: :param certificate: :return: """ key = RSA.importKey(certificate)  # 这里直接用了解密后的证书，但没有去导出公钥，似乎也是可以的。怎么导公钥还没搞懂。 verifier = pkcs1_15.new(key) hash_obj = SHA256.new(check_data.encode('utf8')) return verifier.verify(hash_obj, base64.b64decode(signature)) def make_headers_v3(mchid, serial_num, data='', method='GET'): """ 定义微信支付请求接口中请求头认证 :param mchid: 商户ID :param serial_num: 证书序列号 :param data: 请求体内容 :param method: 请求方法 :return: headers(请求头) """ # 4.定义生成签名的函数 get_sign(sign_str) # 5.生成请求随机串 random_str = ''.join(random.choice(string.ascii_uppercase + string.digits) for _ in range(32)) # 6.生成请求时间戳 time_stamps = str(int(time.time())) # 7.生成签名串 sign_str = f"{method}\n{'/v3/pay/transactions/jsapi'}\n{time_stamps}\n{random_str}\n{data}\n" # 8.生成签名 sign = get_sign(sign_str) # 9.生成HTTP请求头 headers = { 'Content-Type': 'application/json', 'Authorization': 'WECHATPAY2-SHA256-RSA2048 ' + f'mchid="{mchid}",nonce_str="{random_str}",signature="{sign}",timestamp="{time_stamps}",serial_no="{serial_num}"' } return headers, random_str, time_stamps

import decimal import json import traceback from django.http import HttpResponse def notify_view(request, *args, **kwargs): """ 支付完成之后的通知(微信官方返回的数据) :param request: :param args: :param kwargs: :return: """ try: # 1.获得支付通知的参数 body = request.body data = bytes.decode(body, 'utf-8') newdata = json.loads(data) # newdata = { #     "id": "9d40acfd-13cb-5175-a5aa-6c421f794952", #     "create_time": "2023-01-06T15:12:49+08:00", #     "resource_type": "encrypt-resource", #     "event_type": "TRANSACTION.SUCCESS", #     "summary": "\xe6\x94\xaf\xe4\xbb\x98\xe6\x88\x90\xe5\x8a\x9f", #     "resource": { #         "original_type": #         "transaction", #         "algorithm": "AEAD_AES_256_GCM", #         "ciphertext": "UF5gLXfe8qBv9qxQsf+/Mb6as+vbIhUS8Dm25qGIJIIdXTorUUjqZH1+" #                       "jMQxkxma/Gn9bOxeAoQWPEuIoJ2pB328Iv90jmHTrouoP3L60mjNgGJS8d3H8i1zAPBXCpP4mgvgRANWsw4pAWj1lFM5BZr4aP+" #                       "pNMc5TdwreGBG3rO9sbCLXsSRfW8pVZ7IfPnhPDTOWP3P1k5ikHedcRt4/HP69oDBEe5RSsD93wO/" #                       "lrIwycStVHyecBaliwpVMRnNnRCXqhlalNJ3NJ6jcgy32fP1J+L90ntwGyqMmZUS71P5TN1H0iH5rXNpRY9IF3pvN+" #                       "lei5IS86wEoVXkmEsPcJrHaabn7rghxuZoqwuauMIiMwBLllnEmgXfAbJA4FJy+" #                       "OLhZPrMWMkkiNCLcL069QlvhLXYi/0V9PQVTnvtA5RLarj26s4WSqTZ2I5VGHbTqSIZvZYK3F275KEbQsemYETl18xwZ+" #                       "WAuSrYaSKN/pKykK37vUGtT3FeIoJup2c6M8Ghull3OcVmqCOsgvU7/pNjl1rLKEJB6t/X9avcHv+feikwQBtBmd/b2qCeSrEpM7US", #         "associated_data": "transaction", #         "nonce": "cKEdw8eV9Bh0" #     } # } nonce = newdata['resource']['nonce'] ciphertext = newdata['resource']['ciphertext'] associated_data = newdata['resource']['associated_data'] try: payment = decrypt(nonce, ciphertext, associated_data, pay_key) break except Exception as e: print(e) if not payment: return HttpResponse({"code": "FAIL", "message": "失败"}, status=400) payment = eval(payment.decode('utf-8')) # payment = { #     "mchid": "xxxx", #     "appid": "xxxx", #     "out_trade_no": "20231654836163523608", #     "transaction_id": "4200001646202301065425000524", #     "trade_type": "JSAPI", #     "trade_state": "SUCCESS", #     "trade_state_desc": "\xe6\x94\xaf\xe4\xbb\x98\xe6\x88\x90\xe5\x8a\x9f", #     "bank_type": "OTHERS", #     "attach": "", #     "success_time": "2023-01-06T15:12:49+08:00", #     "payer": { #         "openid": "xxxxx" #     }, #     "amount": { #         "total": 1, #         "payer_total": 1, #         "currency": "CNY", #         "payer_currency": "CNY" #     } # } orderno = payment['out_trade_no'] zf_status = True if payment["trade_type"] == "SUCCESS" else False if zf_status: money = decimal.Decimal(int(payment["amount"]["payer_total"]) / 100).quantize(decimal.Decimal("0.00")) else: money = decimal.Decimal(0.0).quantize(decimal.Decimal("0.00")) # 7.回调报文签名验证 # 同第一篇签名验证的代码 wechatpay_serial, wechatpay_timestamp, wechatpay_nonce, wechatpay_signature, certificate = check_wx_cert(request, mchid, pay_key, serial_num) if wechatpay_serial == serial_num:  # 应答签名中的序列号同证书序列号应相同 # 8.获得回调报文中交易号后修改已支付订单状态 res = conorder.objects.filter(orderno=orderno, paystatus=-1).first() if res: res.paystatus = 1 res.save() else: res.paystatus = -1 res.save() # 9.项目业务逻辑 return HttpResponse({"code": "SUCCESS", "message": "成功"}) else: log.error(f"证书序列号比对失败【请求头中证书序列号：{wechatpay_serial}；本地存储证书序列号：{serial_num}；】") return HttpResponse({"code": "FAIL", "message": "失败"}, status=400) except Exception as e: log.error(f"微信回调接口报错：{e},{traceback.format_exc()}") return HttpResponse({"code": "FAIL", "message": "失败"}, status=400)