Python使用微信SDK实现的微信支付功能示例
作者：yueguanghaidao 发布时间：2022-09-22 00:19:41
标签：Python,微信SDK,微信支付
本文实例讲述了Python使用微信SDK实现的微信支付功能。分享给大家供大家参考，具体如下：
最近一段时间一直在搞微信平台开发，v3.37版本微信支付接口变化贼大，所以就看着php的demo移植为Python版，为了保持一致，所以接口方法基本都没有变，这样的好处就是不用写demo了，看着微信官方的demo照葫芦画瓢就可以了。
代码放到github下载地址：https://github.com/Skycrab/wzhifuSDK
还可以点击此处本站下载。
我主要测试了JsApi调用方式，其它的调用方式并没有测试，如果你发现了bug，请多多pull request，我将不甚感激。
方便观看，代码贴于此。
#coding:utf-8
"""
Created on 2014-11-24
@author: http://blog.csdn.net/yueguanghaidao
* 微信支付帮助库
* ====================================================
* 接口分三种类型：
* 【请求型接口】--Wxpay_client_
* 统一支付接口类--UnifiedOrder
* 订单查询接口--OrderQuery
* 退款申请接口--Refund
* 退款查询接口--RefundQuery
* 对账单接口--DownloadBill
* 短链接转换接口--ShortUrl
* 【响应型接口】--Wxpay_server_
* 通用通知接口--Notify
* Native支付——请求商家获取商品信息接口--NativeCall
* 【其他】
* 静态链接二维码--NativeLink
* JSAPI支付--JsApi
* =====================================================
* 【CommonUtil】常用工具：
* trimString()，设置参数时需要用到的字符处理函数
* createNoncestr()，产生随机字符串，不长于32位
* formatBizQueryParaMap(),格式化参数，签名过程需要用到
* 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):
"""配置账号信息"""
#=======【基本信息设置】=====================================
#微信公众号身份的唯一标识。审核通过后，在微信发送的邮件中查看
APPID = "wx8888888888888888"
#JSAPI接口中获取openid，审核后在公众平台开启开发模式后可查看
APPSECRET = "48888888888888888888888888888887"
#受理商ID，身份标识
MCHID = "18888887"
#商户支付密钥Key。审核通过后，在微信发送的邮件中查看
KEY = "48888888888888888888888888888886"
#=======【异步通知url设置】===================================
#异步通知url，商户根据实际开发过程设定
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发送请求"""
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发送请求"""
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):
"""产生随机字符串，不长于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):
"""格式化参数，签名过程需要使用"""
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):
"""生成签名"""
#签名步骤一：按字典序排序参数,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网页端调起支付接口"""
code = None #code码，用以获取openid
openid = None #用户的openid
parameters = None #jsapi参数，格式为json
prepay_id = None #使用统一支付接口得到的预支付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的参数"""
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 = {} #请求参数，类型为关联数组
self.result = {} #返回参数，类型为关联数组
def setParameter(self, parameter, parameterValue):
"""设置请求参数"""
self.parameters[self.trimString(parameter)] = self.trimString(parameterValue)
def createXml(self):
"""设置标配的请求参数，生成签名，生成接口参数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):
"""统一支付接口类"""
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):
"""生成接口参数xml"""
#检测必填参数
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):
"""生成接口参数xml"""
#检测必填参数
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):
"""生成接口参数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):
"""生成接口参数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):
"""生成接口参数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):
"""生成接口参数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 = {} #接收到的数据，类型为关联数组
self.returnParameters = {} #返回参数，类型为关联数组
def saveData(self, xml):
"""将微信的请求xml转换成关联数组，以方便数据处理"""
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):
"""获取微信的请求数据"""
return self.data
def setReturnParameter(self, parameter, parameterValue):
"""设置返回微信的xml数据"""
self.returnParameters[self.trimString(parameter)] = self.trimString(parameterValue)
def createXml(self):
"""生成接口参数xml"""
return self.arrayToXml(self.returnParameters)
def returnXml(self):
"""将xml数据返回微信"""
returnXml = self.createXml()
return returnXml
class Notify_pub(Wxpay_server_pub):
"""通用通知接口"""
class NativeCall_pub(Wxpay_server_pub):
"""请求商家获取商品信息接口"""
def createXml(self):
"""生成接口参数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):
"""静态链接二维码"""
url = None #静态链接
def __init__(self):
self.parameters = {} #静态链接参数
def setParameter(self, parameter, parameterValue):
"""设置参数"""
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()
希望本文所述对大家Python程序设计有所帮助。
0
投稿猜你喜欢
函数内部的函数只能在函数内部调用，不能在函数外部调用，通过接下来的学习你将会知道为什么会出现这种情况。一、名称空间名称空间（name spa- 本文实例讲述了php利用cookies实现购物车的方法。分享给大家供大家参考。具体分析如下：php购物车是在电子商务网站会用到的,一种像超市
- 实战场景Python 工程师在日常的工作中，经常会碰到解析和处理PDF文件的情况，实战中需求主要分为如下情况：提取 PDF 中的文字将 PD
- 如下所示：update-alternatives --config python一条简单的命令，如下图所示：来源：https://blog.
- 本文实例讲述了Python利用matplotlib绘制约数个数统计图。分享给大家供大家参考，具体如下：利用Python计算1000以内自然数
- 在CSS布局中，ul，li的运用是非常常见的。配合DIV与CSS，创建无表格布局。 其实Dreamweaver中也有自带的CSS布局的范例，
- 一、利用xpath进行（全程使用）driver.find_element_by_xpath()二、代码部分与图片内容打开淘宝网站，点击登录，
- 1. Python模块和包：一切从基础开始Python模块是一个Python文件，包含一些相关的函数、类或变量的定义，可以通过 i
- logging模块介绍Python的logging模块提供了通用的日志系统，熟练使用logging模块可以方便开发者开发第三方模块或者是自己
- 实现效果：实现代码import numpy as npfrom skimage import img_as_floatimport matp
- 在使用数据库的时候,难免要在使用过程中进行删除的操作,如果是使用int类型的字段,令其自增长,这是个最简单的办法,但是后果会有些不是你想要的
- KNN核心算法函数，具体内容如下#! /usr/bin/env python3# -*- coding: utf-8 -*-# fileNa
- 这篇文章主要介绍了Python如何使用字符打印照片,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可
- 本文实例为大家分享了js浏览器倒计时跳转页面效果，供大家参考，具体内容如下效果图：<!DOCTYPE html><html
- Windows下ORACLE完全卸载：使用OUI可以卸载数据库，但卸载后注册表和文件系统内仍会有部分残留。这些残留不仅占用磁盘空间，而且影响
- oracle命令行删除用户： connect / as sysdba; shutdown abort; startup; drop user
- 我经常需要用Python与solr进行异步请求工作。这里有段代码阻塞在Solr http请求上， 直到第一个完成才会执行第二个请
- 本文实例讲述了python使用 __init__初始化操作。分享给大家供大家参考，具体如下：# -*- coding:utf-8 -*-#
- 本文实例讲述了Python2随机数列生成器。分享给大家供大家参考，具体如下：#filename:randNumber.pyimport ra
- //图片上传<SCRIPT RUNAT=SERVER LANGUAGE=VBSCRIPT> Function GetUpload
