微信小程序代码实现（登录、授权和支付）
整体流程看上一篇博客，或者去微信公众平台查看文档
只列出核心代码，详细代码见码云michaelben
登录
// //小程序端
// app.js
App({
onLaunch: function () {
var that=this
// // 登录
wx.login({
success: res => {
console.log("code",res.code)
wx.request({
url: that.globalData.URL+"login/",
data:{
"code":res.code
},
header:{
"content-type":"application/json"
},
method:"POST",
success:function(e){
console.log(e)
wx.setStorageSync("token", e.data.data.token)
}
})
//发送 res.code 到后台换取 openId, sessionKey, unionId
}
})
// onShow:function(e){
// // console.log("show",e)
// },
// onHide:function(){
// console.log("hide")
// },
// onError:function(e){
// console.log("error", e)
// },
globalData: {
userInfo: null,
URL:"http://127.0.0.1:8000/"
}
})
## 后端
# url.py
path('login/', views.Login.as_view()),
# views.py
from rest_framework.views import APIView
from rest_framework.response import Response
from app01.wx import Wx_login
import time, hashlib
from django.core.cache import cache
from app01 import models
class Login(APIView):
def post(self,request):
param = request.data
if param.get("code"):
# print(param.get('code'))
data=Wx_login.login(param.get("code"))
# print(data)
if data:
# 1 session_key+时间戳存到一个key.md5
md5=hashlib.md5()
md5.update(data.get("session_key").encode("utf8"))
md5.update(str(time.time()).encode("utf8"))
key=md5.hexdigest()
#2 session_key与openid做绑定赋值给val
val=data.get("session_key")+'&'+data.get("openid")
#3key->val存到redis,
cache.set(key,val)
#4把openid存到数据库
user_data=models.Wxuser.objects.filter(openid=data.get("openid")).first()
if not user_data:
models.Wxuser.objects.create(openid=data.get("openid"))
#5把key返回给小程序
return Response({"code": 200, "msg": "suc","data":{"token":key}})
else:
return Response({"code": 202, "msg": "code无效"})
else:
return Response({"code":201,"msg":"缺少参数"})
授权（只列出录音，其他非userinfo需要权限的功能见授权）
// test.wxml
<button bind:tap="record">录音</button>
//test.js
const app=getApp()
record:function(){
wx.getSetting({
success(res) {
console.log("res", res.authSetting['scope.record'])
if (!res.authSetting['scope.record']) {
wx.authorize({
scope: 'scope.record',
success() {
// 用户已经同意小程序使用录音功能，后续调用 wx.startRecord 接口不会弹窗询问
wx.startRecord()
},
fail(){
console.log("你没有授权")
}
})
}else{
wx.startRecord()
}
}
})
}
授权（用户信息）
前端
// test.wxml
<button open-type="getUserInfo" bindgetuserinfo="info1">获取用户信息</button>
// test.js
info1:function(res){
console.log(res,"按钮")
// wx.getUserInfo({
// success: function (res) {
// console.log(res, "用户信息")
// }
// })
var that=this
wx.checkSession({
success() {
//session_key 未过期，并且在本生命周期一直有效
wx.request({
url: app.globalData.URL + "userinfo/",
data: {
encryptedData: res.detail.encryptedData,
iv: res.detail.iv,
token:wx.getStorageSync("token")
},
header: {
"content-type": "application/json"
},
method: "POST",
success: function (e) {
console.log(e)
}
})
},
fail() {
// session_key 已经失效，需要重新执行登录流程
// wx.login() //重新登录
}
})
},
后端
#view.py
class UserInfo(APIView):
def post(self,request):
param = request.data
if param.get('token') and param.get("encryptedData") and param.get("iv"):
cache_data=cache.get(param.get('token'))
print(cache_data)
if cache_data:
# 获取session_key
session_key,openid=cache_data.split("&")
#数据解密
user_info=WXBizDataCrypt.get_info(session_key, param.get("encryptedData"),param.get("iv"))
print(user_info)
#存入数据库
user_data={
'name': user_info['nickName'],
'avatar': user_info['avatarUrl'],
'language': user_info['language'],
'province': user_info['province'],
'city': user_info['city'],
'country': user_info['country'],
}
models.Wxuser.objects.filter(openid=openid).update(**user_data)
data=models.Wxuser.objects.filter(openid=openid).first()
data=Wxuser_ser.Wxuser_ser(data,many=False).data
return Response({"code": 200, "msg": "suc", "data": data})
else:
return Response({"code": 202, "msg": "token无效"})
else:
return Response({"code": 201, "msg": "缺少参数"})
#WXBiz.py
import base64
import json
from Crypto.Cipher import AES
from app01.wx import settings
class WXBizDataCrypt:
def __init__(self, appId, sessionKey):
self.appId = appId
self.sessionKey = sessionKey
def decrypt(self, encryptedData, iv):
# base64 decode
sessionKey = base64.b64decode(self.sessionKey)
encryptedData = base64.b64decode(encryptedData)
iv = base64.b64decode(iv)
cipher = AES.new(sessionKey, AES.MODE_CBC, iv)
decrypted = json.loads(self._unpad(cipher.decrypt(encryptedData)))
if decrypted['watermark']['appid'] != self.appId:
raise Exception('Invalid Buffer')
return decrypted
def _unpad(self, s):
return s[:-ord(s[len(s)-1:])]
@classmethod
def get_info(cls,sessionKey,encryptedData,iv):
return cls(settings.AppId, sessionKey).decrypt(encryptedData, iv)
支付(详细见码云和官方文档)
后端
class Pay(APIView):
def post(self,request):
param = request.data
if param.get("token"):
cache_data = cache.get(param.get("token"))
if cache_data:
# 获取客户端ip,如果是负载均衡，就用HTTP_X_FORWARDED_FOR，如果不是就用下面的
if request.META.get('HTTP_X_FORWARDED_FOR'):
self.ip = request.META['HTTP_X_FORWARDED_FOR']
else:
self.ip = request.META['REMOTE_ADDR']
session_key,self.openid=cache_data.split("&")
data=self.get_pay_data()
return Response({"code": 200, "msg": "suc","data":data})
else:
return Response({"code": 202, "msg": "token无效"})
def get_nonce_str(self,num=30):
# strs = ""
# for i in range(30):
# strs += str(random.randint(0,9))
all_str = "0123456789abcdefghijklmnopqrstuvwxyz"
strs = "".join(random.sample(all_str,num))
return strs
def get_out_trade_no(self):
import time
strs = str(int(time.time()))+self.get_nonce_str(5)
return strs
def get_sign(self):
data_dic = {
"nonce_str": self.nonce_str,
"out_trade_no": self.out_trade_no,
"spbill_create_ip": self.ip,
"notify_url": self.notify_url,
"openid": self.openid,
"body": self.body,
"trade_type": "JSAPI",
"sign_type": "MD5",
"appid": self.appid,
"total_fee": self.total_fee,
"mch_id": self.mch_id
}
str_a = "&".join([ f"{i}={data_dic[i]}" for i in sorted(data_dic)])
str_b = f"{str_a}&key={settings.pay_apikey}"
md5 = hashlib.md5()
md5.update(str_b.encode("utf8"))
return md5.hexdigest().upper()
def xml_to_dic(self,xml_data):
import xml.etree.ElementTree as ET
xml_data = ET.fromstring(xml_data)
dic = {}
for child in xml_data:
dic[child.tag] =child.text
return dic
def get_prepay_data(self):
url = "https://api.mch.weixin.qq.com/pay/unifiedorder"
response = requests.post(url=url,data=self.body_data.encode("utf8"),headers={"content-type":"application/xml"})
xml_data = response.content
dic_data = self.xml_to_dic(xml_data)
return dic_data
def get_second_sign(self):
self.second_nonceStr = self.get_nonce_str()
self.timeStamp = str(int(time.time()))
data_dic = {
"appId":settings.AppId,
"timeStamp":self.timeStamp,
"nonceStr":self.second_nonceStr,
"package":f"prepay_id={self.prepay_data.get('prepay_id')}",
"signType":"MD5"
}
print(data_dic)
str_a = "&".join([f"{i}={data_dic[i]}" for i in sorted(data_dic)])
str_b = f"{str_a}&key={settings.pay_apikey}"
md5 = hashlib.md5()
md5.update(str_b.encode("utf8"))
return md5.hexdigest().upper()
def get_pay_data(self):
self.appid = settings.AppId
self.mch_id = settings.pay_mchid
self.nonce_str = self.get_nonce_str()
self.sign_type = "MD5"
self.body = "py11最难一届"
self.out_trade_no = self.get_out_trade_no()
self.total_fee = 1
self.spbill_create_ip = self.ip
self.notify_url = "http://www.weixin.qq.com/wxpay/pay.php"
self.trade_type = "JSAPI"
self.sign = self.get_sign()
self.body_data = f"""
<xml>
<appid>{self.appid}</appid>
<mch_id>{self.mch_id}</mch_id>
<nonce_str>{self.nonce_str}</nonce_str>
<sign>{self.sign}</sign>
<body>{self.body}</body>
<out_trade_no>{self.out_trade_no}</out_trade_no>
<total_fee>1</total_fee>
<sign_type>MD5</sign_type>
<spbill_create_ip>{ self.spbill_create_ip}</spbill_create_ip>
<notify_url>{self.notify_url}</notify_url>
<openid>{self.openid}</openid>
<trade_type>JSAPI</trade_type>
</xml>"""
self.prepay_data=self.get_prepay_data()
second_sign=self.get_second_sign()
data = {
"timeStamp":self.timeStamp,
"nonceStr":self.second_nonceStr,
"package":f"prepay_id={self.prepay_data.get('prepay_id')}",
"paySign":second_sign
}
return data