python實現微信付款碼支付（刷卡支付）（純python）
阿新 • • 發佈：2018-12-08
參考連結https://github.com/Jolly23/wx_pay_python
https://pay.weixin.qq.com/wiki/doc/api/micropay.php?chapter=9_10&index=1
use.py from wx_pay import WxPay, WxPayError import sys wx_pay = WxPay( wx_app_id='', # 微信平臺appid wx_mch_id='', # 微信支付商戶號 wx_mch_key='', # wx_mch_key 微信支付重要金鑰，請登入微信支付商戶平臺，在 賬戶中心-API安全-設定API金鑰設定 wx_notify_url='http://www.example.com/pay/weixin/notify' # wx_notify_url 接受微信付款訊息通知地址（通常比自己把支付成功訊號寫在js裡要安全得多，推薦使用這個來接收微信支付成功通知） # wx_notify_url 開發詳見https://pay.weixin.qq.com/wiki/doc/api/jsapi.php?chapter=9_7 ) for line in sys.stdin: for value in line.split():#這兩個for迴圈是為了得到掃碼器掃到的二維碼資訊 wx_pay.swiping_card_payment( body=u'test', # 例如：綜合超市 total_fee=1, # total_fee 單位是 分， 100 = 1元, 單使用者 單筆上限／當日上限：2W／2W auth_code=value, # 掃碼支付授權碼，裝置讀取使用者微信中的條碼或者二維碼資訊（注：使用者刷卡條形碼規則：18位純數字，以10、11、12、13、14、15開頭） spbill_create_ip='', # 呼叫微信企業付款介面伺服器公網IP地址219.223.207.240 10.12.9.1 )
這裡實現微信付款碼支付只用到了wx_pay.py中的swiping_card_payment 函式，想實現其他付款方式可呼叫該py檔案中其他函式。執行此程式碼，只需將 wx_app_id='', # 微信平臺appid wx_mch_id='', # 微信支付商戶號 wx_mch_key='',填進去，並將自己電腦的ip地址賦給spbill_create_ip即可
wx_pay.py # -*- coding: utf-8 -*- import hashlib import random import string import time import urllib.request as urllib2 import urllib import requests try: from flask import request except ImportError: request = None try: from xml.etree import cElementTree as ETree except ImportError: from xml.etree import ElementTree as ETree class WxPayError(Exception): def __init__(self, msg): super(WxPayError, self).__init__(msg) class WxPay(object): def __init__(self, wx_app_id, wx_mch_id, wx_mch_key, wx_notify_url): self.opener = urllib2.build_opener(urllib2.HTTPSHandler()) self.WX_APP_ID = wx_app_id self.WX_MCH_ID = wx_mch_id self.WX_MCH_KEY = wx_mch_key self.WX_NOTIFY_URL = wx_notify_url @staticmethod def user_ip_address(): return request.remote_addr if request else None @staticmethod def nonce_str(length=32): char = string.ascii_letters + string.digits return "".join(random.choice(char) for _ in range(length)) @staticmethod def to_utf8(raw): return raw.encode("utf-8") if isinstance(raw, str) else raw @staticmethod def to_dict(content): raw = {} root = ETree.fromstring(content) for child in root: raw[child.tag] = child.text return raw @staticmethod def random_num(length): digit_list = list(string.digits) random.shuffle(digit_list) return ''.join(digit_list[:length]) def sign(self, raw): """ 生成簽名 參考微信簽名生成演算法 https://pay.weixin.qq.com/wiki/doc/api/jsapi.php?chapter=4_3 """ raw = [(k, str(raw[k]) if isinstance(raw[k], (int, float)) else raw[k]) for k in sorted(raw.keys())] s = "&".join("=".join(kv) for kv in raw if kv[1]) s += "&key={0}".format(self.WX_MCH_KEY) return hashlib.md5(self.to_utf8(s)).hexdigest().upper() def check(self, raw): """ 驗證簽名是否正確 """ sign = raw.pop("sign") # print("1:"+sign) # print("2:"+self.sign(raw)) return sign == self.sign(raw) # def to_xml(self, raw): # s = "" # for k, v in raw.items(): # s += "<{0}>{1}</{0}>".format(k, self.to_utf8(v), k) # return "<xml>{0}</xml>".format(s) def to_xml(self, data): """ 將 dict 物件轉換成微信支付互動所需的 XML 格式資料 :param data: dict 物件 :return: xml 格式資料 """ xml = [] for k in sorted(data.keys()): v = data.get(k) if k == 'detail' and not v.startswith('<![CDATA['): v = '<![CDATA[{}]]>'.format(v) xml.append('<{key}>{value}</{key}>'.format(key=k, value=v)) return '<xml>{}</xml>'.format(''.join(xml)) def fetch(self, url, data): req = urllib2.Request(url, data=self.to_xml(data)) try: resp = self.opener.open(req, timeout=20) except urllib.error.URLError as e: resp = e print(type(resp)) re_info = resp.read() try: return self.to_dict(re_info) except ETree.ParseError: return re_info def fetch_with_ssl(self, url, data, api_client_cert_path, api_client_key_path): #print(self.to_xml(data)) req = requests.post(url, data=self.to_xml(data), cert=(api_client_cert_path, api_client_key_path)) return self.to_dict(req.content) def reply(self, msg, ok=True): code = "SUCCESS" if ok else "FAIL" return self.to_xml(dict(return_code=code, return_msg=msg)) def unified_order(self, **data): """ 統一下單 詳細規則參考 https://pay.weixin.qq.com/wiki/doc/api/jsapi.php?chapter=9_1 :param data: out_trade_no, body, total_fee, trade_type out_trade_no: 商戶訂單號 body: 商品描述 total_fee: 標價金額, 整數, 單位 分 trade_type: 交易型別 user_ip 在flask框架下可以自動填寫, 非flask框架需傳入spbill_create_ip :return: 統一下單生成結果 """ url = "https://api.mch.weixin.qq.com/pay/unifiedorder" # 必填引數 if "out_trade_no" not in data: raise WxPayError(u"缺少統一支付介面必填引數out_trade_no") if "body" not in data: raise WxPayError(u"缺少統一支付介面必填引數body") if "total_fee" not in data: raise WxPayError(u"缺少統一支付介面必填引數total_fee") if "trade_type" not in data: raise WxPayError(u"缺少統一支付介面必填引數trade_type") # 關聯引數 if data["trade_type"] == "JSAPI" and "openid" not in data: raise WxPayError(u"trade_type為JSAPI時，openid為必填引數") if data["trade_type"] == "NATIVE" and "product_id" not in data: raise WxPayError(u"trade_type為NATIVE時，product_id為必填引數") user_ip = self.user_ip_address() if not user_ip and "spbill_create_ip" not in data: raise WxPayError(u"當前未使用flask框架，缺少統一支付介面必填引數spbill_create_ip") data.setdefault("appid", self.WX_APP_ID) data.setdefault("mch_id", self.WX_MCH_ID) data.setdefault("notify_url", self.WX_NOTIFY_URL) data.setdefault("nonce_str", self.nonce_str()) data.setdefault("spbill_create_ip", user_ip) data.setdefault("sign", self.sign(data)) raw = self.fetch(url, data) if raw["return_code"] == "FAIL": raise WxPayError(raw["return_msg"]) err_msg = raw.get("err_code_des") if err_msg: raise WxPayError(err_msg) return raw def js_pay_api(self, **kwargs): """ 生成給JavaScript呼叫的資料 詳細規則參考 https://pay.weixin.qq.com/wiki/doc/api/jsapi.php?chapter=7_7&index=6 :param kwargs: openid, body, total_fee openid: 使用者openid body: 商品名稱 total_fee: 標價金額, 整數, 單位 分 out_trade_no: 商戶訂單號, 若未傳入則自動生成 :return: 生成微信JS介面支付所需的資訊 """ kwargs.setdefault("trade_type", "JSAPI") if "out_trade_no" not in kwargs: kwargs.setdefault("out_trade_no", self.nonce_str()) raw = self.unified_order(**kwargs) package = "prepay_id={0}".format(raw["prepay_id"]) timestamp = int(time.time()) nonce_str = self.nonce_str() raw = dict(appId=self.WX_APP_ID, timeStamp=timestamp, nonceStr=nonce_str, package=package, signType="MD5") sign = self.sign(raw) return dict(package=package, appId=self.WX_APP_ID, timeStamp=timestamp, nonceStr=nonce_str, sign=sign) def order_query(self, **data): """ 訂單查詢 詳細規則參考 https://pay.weixin.qq.com/wiki/doc/api/jsapi.php?chapter=9_2 :param data: out_trade_no, transaction_id至少填一個 out_trade_no: 商戶訂單號 transaction_id: 微信訂單號 :return: 訂單查詢結果 """ url = "https://api.mch.weixin.qq.com/pay/orderquery" if "out_trade_no" not in data and "transaction_id" not in data: raise WxPayError(u"訂單查詢介面中，out_trade_no、transaction_id至少填一個") data.setdefault("appid", self.WX_APP_ID) data.setdefault("mch_id", self.WX_MCH_ID) data.setdefault("nonce_str", self.nonce_str()) data.setdefault("sign", self.sign(data)) raw = self.fetch(url, data) if raw["return_code"] == "FAIL": raise WxPayError(raw["return_msg"]) return raw def close_order(self, out_trade_no): """ 關閉訂單 詳細規則參考 https://pay.weixin.qq.com/wiki/doc/api/jsapi.php?chapter=9_3 :param out_trade_no: 商戶訂單號 :return: 申請關閉訂單結果 """ url = "https://api.mch.weixin.qq.com/pay/closeorder" data = { 'out_trade_no': out_trade_no, 'appid': self.WX_APP_ID, 'mch_id': self.WX_MCH_ID, 'nonce_str': self.nonce_str(), } data["sign"] = self.sign(data) raw = self.fetch(url, data) if raw["return_code"] == "FAIL": raise WxPayError(raw["return_msg"]) return raw def refund(self, api_cert_path, api_key_path, **data): """ 申請退款 詳細規則參考 https://pay.weixin.qq.com/wiki/doc/api/jsapi.php?chapter=9_4 :param api_cert_path: 微信支付商戶證書路徑，此證書(apiclient_cert.pem)需要先到微信支付商戶平臺獲取，下載後儲存至伺服器 :param api_key_path: 微信支付商戶證書路徑，此證書(apiclient_key.pem)需要先到微信支付商戶平臺獲取，下載後儲存至伺服器 :param data: out_trade_no、transaction_id至少填一個, out_refund_no, total_fee, refund_fee out_trade_no: 商戶訂單號 transaction_id: 微信訂單號 out_refund_no: 商戶退款單號（若未傳入則自動生成） total_fee: 訂單金額 refund_fee: 退款金額 :return: 退款申請返回結果 """ url = "https://api.mch.weixin.qq.com/secapi/pay/refund" if "out_trade_no" not in data and "transaction_id" not in data: raise WxPayError(u"訂單查詢介面中，out_trade_no、transaction_id至少填一個") if "total_fee" not in data: raise WxPayError(u"退款申請介面中，缺少必填引數total_fee") if "refund_fee" not in data: raise WxPayError(u"退款申請介面中，缺少必填引數refund_fee") if "out_refund_no" not in data: data.setdefault("out_refund_no", self.nonce_str()) data.setdefault("appid", self.WX_APP_ID) data.setdefault("mch_id", self.WX_MCH_ID) data.setdefault("op_user_id", self.WX_MCH_ID) data.setdefault("nonce_str", self.nonce_str()) data.setdefault("sign", self.sign(data)) raw = self.fetch_with_ssl(url, data, api_cert_path, api_key_path) if raw["return_code"] == "FAIL": raise WxPayError(raw["return_msg"]) return raw def refund_query(self, **data): """ 查詢退款 提交退款申請後，通過呼叫該介面查詢退款狀態。退款有一定延時， 用零錢支付的退款20分鐘內到賬，銀行卡支付的退款3個工作日後重新查詢退款狀態。 詳細規則參考 https://pay.weixin.qq.com/wiki/doc/api/jsapi.php?chapter=9_5 :param data: out_refund_no、out_trade_no、transaction_id、refund_id四個引數必填一個 out_refund_no: 商戶退款單號 out_trade_no: 商戶訂單號 transaction_id: 微信訂單號 refund_id: 微信退款單號 :return: 退款查詢結果 """ url = "https://api.mch.weixin.qq.com/pay/refundquery" if "out_refund_no" not in data and "out_trade_no" not in data \ and "transaction_id" not in data and "refund_id" not in data: raise WxPayError(u"退款查詢介面中，out_refund_no、out_trade_no、transaction_id、refund_id四個引數必填一個") data.setdefault("appid", self.WX_APP_ID) data.setdefault("mch_id", self.WX_MCH_ID) data.setdefault("nonce_str", self.nonce_str()) data.setdefault("sign", self.sign(data)) raw = self.fetch(url, data) if raw["return_code"] == "FAIL": raise WxPayError(raw["return_msg"]) return raw def download_bill(self, bill_date, bill_type=None): """ 下載對賬單 詳細規則參考 https://pay.weixin.qq.com/wiki/doc/api/jsapi.php?chapter=9_6 :param bill_date: 對賬單日期 :param bill_type: 賬單型別(ALL-當日所有訂單資訊，[預設]SUCCESS-當日成功支付的訂單, REFUND-當日退款訂單) :return: 資料流形式賬單 """ url = "https://api.mch.weixin.qq.com/pay/downloadbill" data = { 'bill_date': bill_date, 'bill_type': bill_type if bill_type else 'SUCCESS', 'appid': self.WX_APP_ID, 'mch_id': self.WX_MCH_ID, 'nonce_str': self.nonce_str() } data['sign'] = self.sign(data) raw = self.fetch(url, data) return raw def send_red_pack(self, api_cert_path, api_key_path, **data): """ 發給使用者微信紅包 詳細規則參考 https://pay.weixin.qq.com/wiki/doc/api/tools/cash_coupon.php?chapter=13_4&index=3 :param api_cert_path: 微信支付商戶證書路徑，此證書(apiclient_cert.pem)需要先到微信支付商戶平臺獲取，下載後儲存至伺服器 :param api_key_path: 微信支付商戶證書路徑，此證書(apiclient_key.pem)需要先到微信支付商戶平臺獲取，下載後儲存至伺服器 :param data: send_name, re_openid, total_amount, wishing, client_ip, act_name, remark send_name: 商戶名稱 例如: 天虹百貨 re_openid: 使用者openid total_amount: 付款金額 wishing: 紅包祝福語 例如: 感謝您參加猜燈謎活動，祝您元宵節快樂！ client_ip: 呼叫介面的機器Ip地址, 注：此地址為伺服器地址 act_name: 活動名稱 例如: 猜燈謎搶紅包活動 remark: 備註 例如: 猜越多得越多，快來搶！ :return: 紅包發放結果 """ url = "https://api.mch.weixin.qq.com/mmpaymkttransfers/sendredpack" if "send_name" not in data: raise WxPayError(u"向用戶傳送紅包介面中，缺少必填引數send_name") if "re_openid" not in data: raise WxPayError(u"向用戶傳送紅包介面中，缺少必填引數re_openid") if "total_amount" not in data: raise WxPayError(u"向用戶傳送紅包介面中，缺少必填引數total_amount") if "wishing" not in data: raise WxPayError(u"向用戶傳送紅包介面中，缺少必填引數wishing") if "client_ip" not in data: raise WxPayError(u"向用戶傳送紅包介面中，缺少必填引數client_ip") if "act_name" not in data: raise WxPayError(u"向用戶傳送紅包介面中，缺少必填引數act_name") if "remark" not in data: raise WxPayError(u"向用戶傳送紅包介面中，缺少必填引數remark") data.setdefault("wxappid", self.WX_APP_ID) data.setdefault("mch_id", self.WX_MCH_ID) data.setdefault("nonce_str", self.nonce_str()) data.setdefault("mch_billno", u'{0}{1}{2}'.format( self.WX_MCH_ID, time.strftime('%Y%m%d', time.localtime(time.time())), self.random_num(10) )) data.setdefault("total_num", 1) data.setdefault("scene_id", 'PRODUCT_4') data.setdefault("sign", self.sign(data)) raw = self.fetch_with_ssl(url, data, api_cert_path, api_key_path) if raw["return_code"] == "FAIL": raise WxPayError(raw["return_msg"]) return raw def enterprise_payment(self, api_cert_path, api_key_path, **data): """ 使用企業對個人付款功能 詳細規則參考 https://pay.weixin.qq.com/wiki/doc/api/tools/mch_pay.php?chapter=14_2 :param api_cert_path: 微信支付商戶證書路徑，此證書(apiclient_cert.pem)需要先到微信支付商戶平臺獲取，下載後儲存至伺服器 :param api_key_path: 微信支付商戶證書路徑，此證書(apiclient_key.pem)需要先到微信支付商戶平臺獲取，下載後儲存至伺服器 :param data: openid, check_name, re_user_name, amount, desc, spbill_create_ip openid: 使用者openid check_name: 是否校驗使用者姓名 re_user_name: 如果 check_name 為True，則填寫，否則不帶此引數 amount: 金額: 企業付款金額，單位為分 desc: 企業付款描述資訊 spbill_create_ip: 呼叫介面的機器Ip地址, 注：此地址為伺服器地址 :return: 企業轉賬結果 """ url = "https://api.mch.weixin.qq.com/mmpaymkttransfers/promotion/transfers" if "openid" not in data: raise WxPayError(u"企業付款申請介面中，缺少必填引數openid") if "check_name" not in data: raise WxPayError(u"企業付款申請介面中，缺少必填引數check_name") if data['check_name'] and "re_user_name" not in data: raise WxPayError(u"企業付款申請介面中，缺少必填引數re_user_name") if "amount" not in data: raise WxPayError(u"企業付款申請介面中，缺少必填引數amount") if "desc" not in data: raise WxPayError(u"企業付款申請介面中，缺少必填引數desc") if "spbill_create_ip" not in data: raise WxPayError(u"企業付款申請介面中，缺少必填引數spbill_create_ip") data.setdefault("mch_appid", self.WX_APP_ID) data.setdefault("mchid", self.WX_MCH_ID) data.setdefault("nonce_str", self.nonce_str()) data.setdefault("partner_trade_no", u'{0}{1}{2}'.format( self.WX_MCH_ID, time.strftime('%Y%m%d', time.localtime(time.time())), self.random_num(10) )) data['check_name'] = 'FORCE_CHECK' if data['check_name'] else 'NO_CHECK' data.setdefault("sign", self.sign(data)) raw = self.fetch_with_ssl(url, data, api_cert_path, api_key_path) if raw["return_code"] == "FAIL": raise WxPayError(raw["return_msg"]) return raw def swiping_card_payment(self, **data): """ 提交刷卡支付 詳細規則參考 https://pay.weixin.qq.com/wiki/doc/api/micropay.php?chapter=9_10&index=1 :param data: body, out_trade_no, total_fee, auth_code, (可選引數 device_info, detail, goods_tag, limit_pay) body: 商品描述 *out_trade_no: 商戶訂單號 total_fee: 標價金額, 整數, 單位 分 auth_code: 微信支付二維碼掃描結果 *device_info: 終端裝置號(商戶自定義，如門店編號) user_ip 在flask框架下可以自動填寫, 非flask框架需傳入spbill_create_ip :return: 統一下單生成結果 """ url = "https://api.mch.weixin.qq.com/pay/micropay".encode() # 必填引數 if "body" not in data: raise WxPayError(u"缺少刷卡支付介面必填引數body") if "total_fee" not in data: raise WxPayError(u"缺少刷卡支付介面必填引數total_fee") if "out_trade_no" not in data: data.setdefault("out_trade_no", self.nonce_str()) user_ip = self.user_ip_address() if not user_ip and "spbill_create_ip" not in data: raise WxPayError(u"當前未使用flask框架，缺少刷卡支付介面必填引數spbill_create_ip") data.setdefault("appid", self.WX_APP_ID) data.setdefault("mch_id", self.WX_MCH_ID) data.setdefault("nonce_str", self.nonce_str()) data.setdefault("spbill_create_ip", user_ip) data.setdefault("sign", self.sign(data)) #raw = self.fetch(url, data) raw = self.fetch_with_ssl(url,data,api_client_cert_path=r'C:\Users\zhangqian\Desktop\cert\apiclient_cert.pem', api_client_key_path=r'C:\Users\zhangqian\Desktop\cert\apiclient_key.pem') if raw["return_code"] == "FAIL": raise WxPayError(raw["return_msg"]) err_msg = raw.get("err_code_des") if err_msg: raise WxPayError(err_msg) return raw