這篇文章主要介紹了Python使用微信SDK實現(xiàn)的微信支付功能,結合實例形式分析了Python調(diào)用微信SDK接口實現(xiàn)微信支付功能的具體步驟與相關操作技巧,需要的朋友可以參考下
本文實例講述了Python使用微信SDK實現(xiàn)的微信支付功能。分享給大家供大家參考，具體如下：
最近一段時間一直在搞微信平臺開發(fā)，v3.37版本微信支付接口變化賊大，所以就看著php的demo移植為Python版，為了保持一致，所以接口方法基本都沒有變，這樣的好處就是不用寫demo了，看著微信官方的demo照葫蘆畫瓢就可以了。
還可以點擊此處本站下載。
我主要測試了JsApi調(diào)用方式，其它的調(diào)用方式并沒有測試，如果你發(fā)現(xiàn)了bug，請多多pull request，我將不甚感激。
方便觀看，代碼貼于此。
#coding:utf-8
"""
Created on 2014-11-24
@author: http://blog.csdn.net/yueguanghaidao
* 微信支付幫助庫
* ====================================================
* 接口分三種類型：
* 【請求型接口】--Wxpay_client_
* 統(tǒng)一支付接口類--UnifiedOrder
* 訂單查詢接口--OrderQuery
* 退款申請接口--Refund
* 退款查詢接口--RefundQuery
* 對賬單接口--DownloadBill
* 短鏈接轉換接口--ShortUrl
* 【響應型接口】--Wxpay_server_
* 通用通知接口--Notify
* Native支付——請求商家獲取商品信息接口--NativeCall
* 【其他】
* 靜態(tài)鏈接二維碼--NativeLink
* JSAPI支付--JsApi
* =====================================================
* 【CommonUtil】常用工具：
* trimString()，設置參數(shù)時需要用到的字符處理函數(shù)
* createNoncestr()，產(chǎn)生隨機字符串，不長于32位
* formatBizQueryParaMap(),格式化參數(shù)，簽名過程需要用到
* getSign(),生成簽名
* arrayToXml(),array轉xml
* xmlToArray(),xml轉 array
* postXmlCurl(),以post方式提交xml到對應的接口url
* postXmlSSLCurl(),使用證書，以post方式提交xml到對應的接口url
"""
import json
import time
import random
import urllib2
import hashlib
import threading
from urllib import quote
import xml.etree.ElementTree as ET
try:
import pycurl
from cStringIO import StringIO
except ImportError:
pycurl = None
class WxPayConf_pub(object):
"""配置賬號信息"""
#=======【基本信息設置】=====================================
#微信公眾號身份的唯一標識。審核通過后，在微信發(fā)送的郵件中查看
APPID = "wx8888888888888888"
#JSAPI接口中獲取openid，審核后在公眾平臺開啟開發(fā)模式后可查看
APPSECRET = "48888888888888888888888888888887"
#受理商ID，身份標識
MCHID = "18888887"
#商戶支付密鑰Key。審核通過后，在微信發(fā)送的郵件中查看
KEY = "48888888888888888888888888888886"
#=======【異步通知url設置】===================================
#異步通知url，商戶根據(jù)實際開發(fā)過程設定
NOTIFY_URL = "http://******.com/payback"
#=======【JSAPI路徑設置】===================================
#獲取access_token過程中的跳轉uri，通過跳轉將code傳入jsapi支付頁面
JS_API_CALL_URL = "http://******.com/pay/?showwxpaytitle=1"
#=======【證書路徑設置】=====================================
#證書路徑,注意應該填寫絕對路徑
SSLCERT_PATH = "/******/cacert/apiclient_cert.pem"
SSLKEY_PATH = "/******/cacert/apiclient_key.pem"
#=======【curl超時設置】===================================
CURL_TIMEOUT = 30
#=======【HTTP客戶端設置】===================================
HTTP_CLIENT = "CURL" # ("URLLIB", "CURL")
class Singleton(object):
"""單例模式"""
_instance_lock = threading.Lock()
def new(cls, *args, **kwargs):
if not hasattr(cls, "_instance"):
with cls._instance_lock:
if not hasattr(cls, "_instance"):
impl = cls.configure() if hasattr(cls, "configure") else cls
instance = super(Singleton, cls).new(impl, *args, **kwargs)
instance.init(*args, **kwargs)
cls._instance = instance
return cls._instance
class UrllibClient(object):
"""使用urlib2發(fā)送請求"""
def get(self, url, second=30):
return self.postXml(None, url, second)
def postXml(self, xml, url, second=30):
"""不使用證書"""
data = urllib2.urlopen(url, xml, timeout=second).read()
return data
def postXmlSSL(self, xml, url, second=30):
"""使用證書"""
raise TypeError("please use CurlClient")
class CurlClient(object):
"""使用Curl發(fā)送請求"""
def init(self):
self.curl = pycurl.Curl()
self.curl.setopt(pycurl.SSL_VERIFYHOST, False)
self.curl.setopt(pycurl.SSL_VERIFYPEER, False)
#設置不輸出header
self.curl.setopt(pycurl.HEADER, False)
def get(self, url, second=30):
return self.postXmlSSL(None, url, second=second, cert=False, post=False)
def postXml(self, xml, url, second=30):
"""不使用證書"""
return self.postXmlSSL(xml, url, second=second, cert=False, post=True)
def postXmlSSL(self, xml, url, second=30, cert=True, post=True):
"""使用證書"""
self.curl.setopt(pycurl.URL, url)
self.curl.setopt(pycurl.TIMEOUT, second)
#設置證書
#使用證書：cert 與 key 分別屬于兩個.pem文件
#默認格式為PEM，可以注釋
if cert:
self.curl.setopt(pycurl.SSLKEYTYPE, "PEM")
self.curl.setopt(pycurl.SSLKEY, WxPayConf_pub.SSLKEY_PATH)
self.curl.setopt(pycurl.SSLCERTTYPE, "PEM")
self.curl.setopt(pycurl.SSLCERT, WxPayConf_pub.SSLKEY_PATH)
#post提交方式
if post:
self.curl.setopt(pycurl.POST, True)
self.curl.setopt(pycurl.POSTFIELDS, xml)
buff = StringIO()
self.curl.setopt(pycurl.WRITEFUNCTION, buff.write)
self.curl.perform()
return buff.getvalue()
class HttpClient(Singleton):
@classmethod
def configure(cls):
if pycurl is not None and WxPayConf_pub.HTTP_CLIENT != "URLLIB":
return CurlClient
else:
return UrllibClient
class Common_util_pub(object):
"""所有接口的基類"""
def trimString(self, value):
if value is not None and len(value) == 0:
value = None
return value
def createNoncestr(self, length = 32):
"""產(chǎn)生隨機字符串，不長于32位"""
chars = "abcdefghijklmnopqrstuvwxyz0123456789"
strs = []
for x in range(length):
strs.append(chars[random.randrange(0, len(chars))])
return "".join(strs)
def formatBizQueryParaMap(self, paraMap, urlencode):
"""格式化參數(shù)，簽名過程需要使用"""
slist = sorted(paraMap)
buff = []
for k in slist:
v = quote(paraMap[k]) if urlencode else paraMap[k]
buff.append("{0}={1}".format(k, v))
return "&".join(buff)
def getSign(self, obj):
"""生成簽名"""
#簽名步驟一：按字典序排序參數(shù),formatBizQueryParaMap已做
String = self.formatBizQueryParaMap(obj, False)
#簽名步驟二：在string后加入KEY
String = "{0}&key={1}".format(String,WxPayConf_pub.KEY)
#簽名步驟三：MD5加密
String = hashlib.md5(String).hexdigest()
#簽名步驟四：所有字符轉為大寫
result_ = String.upper()
return result_
def arrayToXml(self, arr):
"""array轉xml"""
xml = ["<xml>"]
for k, v in arr.iteritems():
if v.isdigit():
xml.append("<{0}>{1}</{0}>".format(k, v))
else:
xml.append("<{0}><![CDATA[{1}]]></{0}>".format(k, v))
xml.append("</xml>")
return "".join(xml)
def xmlToArray(self, xml):
"""將xml轉為array"""
array_data = {}
root = ET.fromstring(xml)
for child in root:
value = child.text
array_data[child.tag] = value
return array_data
def postXmlCurl(self, xml, url, second=30):
"""以post方式提交xml到對應的接口url"""
return HttpClient().postXml(xml, url, second=second)
def postXmlSSLCurl(self, xml, url, second=30):
"""使用證書，以post方式提交xml到對應的接口url"""
return HttpClient().postXmlSSL(xml, url, second=second)
class JsApi_pub(Common_util_pub):
"""JSAPI支付——H5網(wǎng)頁端調(diào)起支付接口"""
code = None #code碼，用以獲取openid
openid = None #用戶的openid
parameters = None #jsapi參數(shù)，格式為json
prepay_id = None #使用統(tǒng)一支付接口得到的預支付id
curl_timeout = None #curl超時時間
def init(self, timeout=WxPayConf_pub.CURL_TIMEOUT):
self.curl_timeout = timeout
def createOauthUrlForCode(self, redirectUrl):
"""生成可以獲得code的url"""
urlObj = {}
urlObj["appid"] = WxPayConf_pub.APPID
urlObj["redirect_uri"] = redirectUrl
urlObj["response_type"] = "code"
urlObj["scope"] = "snsapi_base"
urlObj["state"] = "STATE#wechat_redirect"
bizString = self.formatBizQueryParaMap(urlObj, False)
return "https://open.weixin.qq.com/connect/oauth2/authorize?"+bizString
def createOauthUrlForOpenid(self):
"""生成可以獲得openid的url"""
urlObj = {}
urlObj["appid"] = WxPayConf_pub.APPID
urlObj["secret"] = WxPayConf_pub.APPSECRET
urlObj["code"] = self.code
urlObj["grant_type"] = "authorization_code"
bizString = self.formatBizQueryParaMap(urlObj, False)
return "https://api.weixin.qq.com/sns/oauth2/access_token?"+bizString
def getOpenid(self):
"""通過curl向微信提交code，以獲取openid"""
url = self.createOauthUrlForOpenid()
data = HttpClient().get(url)
self.openid = json.loads(data)["openid"]
return self.openid
def setPrepayId(self, prepayId):
"""設置prepay_id"""
self.prepay_id = prepayId
def setCode(self, code):
"""設置code"""
self.code = code
def getParameters(self):
"""設置jsapi的參數(shù)"""
jsApiObj = {}
jsApiObj["appId"] = WxPayConf_pub.APPID
timeStamp = int(time.time())
jsApiObj["timeStamp"] = "{0}".format(timeStamp)
jsApiObj["nonceStr"] = self.createNoncestr()
jsApiObj["package"] = "prepay_id={0}".format(self.prepay_id)
jsApiObj["signType"] = "MD5"
jsApiObj["paySign"] = self.getSign(jsApiObj)
self.parameters = json.dumps(jsApiObj)
return self.parameters
class Wxpay_client_pub(Common_util_pub):
"""請求型接口的基類"""
response = None #微信返回的響應
url = None #接口鏈接
curl_timeout = None #curl超時時間
def init(self):
self.parameters = {} #請求參數(shù)，類型為關聯(lián)數(shù)組
self.result = {} #返回參數(shù)，類型為關聯(lián)數(shù)組
def setParameter(self, parameter, parameterValue):
"""設置請求參數(shù)"""
self.parameters[self.trimString(parameter)] = self.trimString(parameterValue)
def createXml(self):
"""設置標配的請求參數(shù)，生成簽名，生成接口參數(shù)xml"""
self.parameters["appid"] = WxPayConf_pub.APPID #公眾賬號ID
self.parameters["mch_id"] = WxPayConf_pub.MCHID #商戶號
self.parameters["nonce_str"] = self.createNoncestr() #隨機字符串
self.parameters["sign"] = self.getSign(self.parameters) #簽名
return self.arrayToXml(self.parameters)
def postXml(self):
"""post請求xml"""
xml = self.createXml()
self.response = self.postXmlCurl(xml, self.url, self.curl_timeout)
return self.response
def postXmlSSL(self):
"""使用證書post請求xml"""
xml = self.createXml()
self.response = self.postXmlSSLCurl(xml, self.url, self.curl_timeout)
return self.response
def getResult(self):
"""獲取結果，默認不使用證書"""
self.postXml()
self.result = self.xmlToArray(self.response)
return self.result
class UnifiedOrder_pub(Wxpay_client_pub):
"""統(tǒng)一支付接口類"""
def init(self, timeout=WxPayConf_pub.CURL_TIMEOUT):
#設置接口鏈接
self.url = "https://api.mch.weixin.qq.com/pay/unifiedorder"
#設置curl超時時間
self.curl_timeout = timeout
super(UnifiedOrder_pub, self).init()
def createXml(self):
"""生成接口參數(shù)xml"""
#檢測必填參數(shù)
if any(self.parameters[key] is None for key in ("out_trade_no", "body", "total_fee", "notify_url", "trade_type")):
raise ValueError("missing parameter")
if self.parameters["trade_type"] == "JSAPI" and self.parameters["openid"] is None:
raise ValueError("JSAPI need openid parameters")
self.parameters["appid"] = WxPayConf_pub.APPID #公眾賬號ID
self.parameters["mch_id"] = WxPayConf_pub.MCHID #商戶號
self.parameters["spbill_create_ip"] = "127.0.0.1" #終端ip
self.parameters["nonce_str"] = self.createNoncestr() #隨機字符串
self.parameters["sign"] = self.getSign(self.parameters) #簽名
return self.arrayToXml(self.parameters)
def getPrepayId(self):
"""獲取prepay_id"""
self.postXml()
self.result = self.xmlToArray(self.response)
prepay_id = self.result["prepay_id"]
return prepay_id
class OrderQuery_pub(Wxpay_client_pub):
"""訂單查詢接口"""
def init(self, timeout=WxPayConf_pub.CURL_TIMEOUT):
#設置接口鏈接
self.url = "https://api.mch.weixin.qq.com/pay/orderquery"
#設置curl超時時間
self.curl_timeout = timeout
super(OrderQuery_pub, self).init()
def createXml(self):
"""生成接口參數(shù)xml"""
#檢測必填參數(shù)
if any(self.parameters[key] is None for key in ("out_trade_no", "transaction_id")):
raise ValueError("missing parameter")
self.parameters["appid"] = WxPayConf_pub.APPID #公眾賬號ID
self.parameters["mch_id"] = WxPayConf_pub.MCHID #商戶號
self.parameters["nonce_str"] = self.createNoncestr() #隨機字符串
self.parameters["sign"] = self.getSign(self.parameters) #簽名
return self.arrayToXml(self.parameters)
class Refund_pub(Wxpay_client_pub):
"""退款申請接口"""
def init(self, timeout=WxPayConf_pub.CURL_TIMEOUT):
#設置接口鏈接
self.url = "https://api.mch.weixin.qq.com/secapi/pay/refund"
#設置curl超時時間
self.curl_timeout = timeout
super(Refund_pub, self).init()
def createXml(self):
"""生成接口參數(shù)xml"""
if any(self.parameters[key] is None for key in ("out_trade_no", "out_refund_no", "total_fee", "refund_fee", "op_user_id")):
raise ValueError("missing parameter")
self.parameters["appid"] = WxPayConf_pub.APPID #公眾賬號ID
self.parameters["mch_id"] = WxPayConf_pub.MCHID #商戶號
self.parameters["nonce_str"] = self.createNoncestr() #隨機字符串
self.parameters["sign"] = self.getSign(self.parameters) #簽名
return self.arrayToXml(self.parameters)
def getResult(self):
""" 獲取結果，使用證書通信(需要雙向證書)"""
self.postXmlSSL()
self.result = self.xmlToArray(self.response)
return self.result
class RefundQuery_pub(Wxpay_client_pub):
"""退款查詢接口"""
def init(self, timeout=WxPayConf_pub.CURL_TIMEOUT):
#設置接口鏈接
self.url = "https://api.mch.weixin.qq.com/pay/refundquery"
#設置curl超時時間
self.curl_timeout = timeout
super(RefundQuery_pub, self).init()
def createXml(self):
"""生成接口參數(shù)xml"""
if any(self.parameters[key] is None for key in ("out_refund_no", "out_trade_no", "transaction_id", "refund_id")):
raise ValueError("missing parameter")
self.parameters["appid"] = WxPayConf_pub.APPID #公眾賬號ID
self.parameters["mch_id"] = WxPayConf_pub.MCHID #商戶號
self.parameters["nonce_str"] = self.createNoncestr() #隨機字符串
self.parameters["sign"] = self.getSign(self.parameters) #簽名
return self.arrayToXml(self.parameters)
def getResult(self):
""" 獲取結果，使用證書通信(需要雙向證書)"""
self.postXmlSSL()
self.result = self.xmlToArray(self.response)
return self.result
class DownloadBill_pub(Wxpay_client_pub):
"""對賬單接口"""
def init(self, timeout=WxPayConf_pub.CURL_TIMEOUT):
#設置接口鏈接
self.url = "https://api.mch.weixin.qq.com/pay/downloadbill"
#設置curl超時時間
self.curl_timeout = timeout
super(DownloadBill_pub, self).init()
def createXml(self):
"""生成接口參數(shù)xml"""
if any(self.parameters[key] is None for key in ("bill_date", )):
raise ValueError("missing parameter")
self.parameters["appid"] = WxPayConf_pub.APPID #公眾賬號ID
self.parameters["mch_id"] = WxPayConf_pub.MCHID #商戶號
self.parameters["nonce_str"] = self.createNoncestr() #隨機字符串
self.parameters["sign"] = self.getSign(self.parameters) #簽名
return self.arrayToXml(self.parameters)
def getResult(self):
"""獲取結果，默認不使用證書"""
self.postXml()
self.result = self.xmlToArray(self.response)
return self.result
class ShortUrl_pub(Wxpay_client_pub):
"""短鏈接轉換接口"""
def init(self, timeout=WxPayConf_pub.CURL_TIMEOUT):
#設置接口鏈接
self.url = "https://api.mch.weixin.qq.com/tools/shorturl"
#設置curl超時時間
self.curl_timeout = timeout
super(ShortUrl_pub, self).init()
def createXml(self):
"""生成接口參數(shù)xml"""
if any(self.parameters[key] is None for key in ("long_url", )):
raise ValueError("missing parameter")
self.parameters["appid"] = WxPayConf_pub.APPID #公眾賬號ID
self.parameters["mch_id"] = WxPayConf_pub.MCHID #商戶號
self.parameters["nonce_str"] = self.createNoncestr() #隨機字符串
self.parameters["sign"] = self.getSign(self.parameters) #簽名
return self.arrayToXml(self.parameters)
def getShortUrl(self):
"""獲取prepay_id"""
self.postXml()
prepay_id = self.result["short_url"]
return prepay_id
class Wxpay_server_pub(Common_util_pub):
"""響應型接口基類"""
SUCCESS, FAIL = "SUCCESS", "FAIL"
def init(self):
self.data = {} #接收到的數(shù)據(jù)，類型為關聯(lián)數(shù)組
self.returnParameters = {} #返回參數(shù)，類型為關聯(lián)數(shù)組
def saveData(self, xml):
"""將微信的請求xml轉換成關聯(lián)數(shù)組，以方便數(shù)據(jù)處理"""
self.data = self.xmlToArray(xml)
def checkSign(self):
"""校驗簽名"""
tmpData = dict(self.data) #make a copy to save sign
del tmpData['sign']
sign = self.getSign(tmpData) #本地簽名
if self.data['sign'] == sign:
return True
return False
def getData(self):
"""獲取微信的請求數(shù)據(jù)"""
return self.data
def setReturnParameter(self, parameter, parameterValue):
"""設置返回微信的xml數(shù)據(jù)"""
self.returnParameters[self.trimString(parameter)] = self.trimString(parameterValue)
def createXml(self):
"""生成接口參數(shù)xml"""
return self.arrayToXml(self.returnParameters)
def returnXml(self):
"""將xml數(shù)據(jù)返回微信"""
returnXml = self.createXml()
return returnXml
class Notify_pub(Wxpay_server_pub):
"""通用通知接口"""
class NativeCall_pub(Wxpay_server_pub):
"""請求商家獲取商品信息接口"""
def createXml(self):
"""生成接口參數(shù)xml"""
if self.returnParameters["return_code"] == self.SUCCESS:
self.returnParameters["appid"] = WxPayConf_pub.APPID #公眾賬號ID
self.returnParameters["mch_id"] = WxPayConf_pub.MCHID #商戶號
self.returnParameters["nonce_str"] = self.createNoncestr() #隨機字符串
self.returnParameters["sign"] = self.getSign(self.returnParameters) #簽名
return self.arrayToXml(self.returnParameters)
def getProductId(self):
"""獲取product_id"""
product_id = self.data["product_id"]
return product_id
class NativeLink_pub(Common_util_pub):
"""靜態(tài)鏈接二維碼"""
url = None #靜態(tài)鏈接
def init(self):
self.parameters = {} #靜態(tài)鏈接參數(shù)
def setParameter(self, parameter, parameterValue):
"""設置參數(shù)"""
self.parameters[self.trimString(parameter)] = self.trimString(parameterValue)
def createLink(self):
if any(self.parameters[key] is None for key in ("product_id", )):
raise ValueError("missing parameter")
self.parameters["appid"] = WxPayConf_pub.APPID #公眾賬號ID
self.parameters["mch_id"] = WxPayConf_pub.MCHID #商戶號
time_stamp = int(time.time())
self.parameters["time_stamp"] = "{0}".format(time_stamp) #時間戳
self.parameters["nonce_str"] = self.createNoncestr() #隨機字符串
self.parameters["sign"] = self.getSign(self.parameters) #簽名
bizString = self.formatBizQueryParaMap(self.parameters, false)
self.url = "weixin://wxpay/bizpayurl?"+bizString
def getUrl(self):
"""返回鏈接"""
self.createLink()
return self.url
def test():
c = HttpClient()
assert c.get("http://www.baidu.com")[:15] == "<!DOCTYPE html>"
c2 = HttpClient()
assert id(c) == id(c2)
if name == "main":
test()
聲明：本網(wǎng)頁內(nèi)容旨在傳播知識，若有侵權等問題請及時與本網(wǎng)聯(lián)系，我們將在第一時間刪除處理。TEL:177 7030 7066 E-MAIL:11247931@qq.com
