支付宝接入
python-alipay-sdk
pycryptodome
一、电脑网站支付
1.1 获取支付宝密钥
沙箱网址
1.APPID
2.应用私钥
3.支付宝公钥
1.2 存放密钥
- 在与 settings.py 的同级目录下创建
pem文件夹 - pem 文件夹下创建
app_private_key.pem和alipay_public_key.pem- app_private_key.pem ：存放应用私钥（选择非Java语言）
- alipay_public_key.pem ：存放支付宝公钥
- 注意存放格式！！！
支付宝公钥
-----BEGIN PUBLIC KEY-----
MIIBIjANBgkxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
-----END PUBLIC KEY-----
应用私钥
-----BEGIN RSA PRIVATE KEY-----
MIIEogIBAAKCAQEAgZXESCF0wkPXCKjLxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
-----END RSA PRIVATE KEY-----
1.3 配置 settings
settings.py
import os
# 应用私钥（选择非Java语言）pem 是文件夹名称
APP_PRIVATE_KEY_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'pem', 'app_private_key.pem')
# 支付宝公钥
ALIPAY_PUBLIC_KEY_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'pem', 'alipay_public_key.pem')
# 沙箱应用的 APPID --> https://open.alipay.com/develop/sandbox/app
APP_ID = '20xxxx9'
# 加密方式
SIGN = 'RSA2'
# 是否是支付宝测试环境(沙箱环境)，如果采用真是支付宝环境，配置False
DEBUG = True
# 支付网关
GATEWAY = 'https://openapi.alipaydev.com/gateway.do' if DEBUG else 'https://openapi.alipay.com/gateway.do'
BASE_URL = 'http://127.0.0.1:8000' # dajngo 服务器
LUFFY_URL = 'http://127.0.0.1:3000' # react 前端地址
NOTIFY_URL = BASE_URL + "/order/success/" # 付款后支付宝向 http://127.0.0.1:8000/order/success/ 发送post请求
RETURN_URL = LUFFY_URL + "/#/pay/success" # 付款后支付宝向 http://127.0.0.1:3000/#/pay/success 发送get请求，请求前端的支付成功页面
理论知识（必看）
1.用户选中订单点击支付，向Dajngo后端发送请求，Django后端接口根据商品信息生成能够唤醒支付界面的 pay_url，前端接收并打开pay_url
2.用户登录支付宝，并成功支付后
3.第三方服务器【支付宝】向我们配置的return_url发送一个GET请求
4.同时向notify_url发送一个异步回调POST请求
注意
- RETURN_URL 和 NOTIFY_URL 必须能够使用公网访问
- 公网访问：其它电脑能访问你设置的 url
- 借助内网穿透工具即可natapp内网穿透应用
auth.py
# 1、放行的网址
if request.path_info in ['/login/', '/register/', '/order/success/']:
return
注意
如果你的Django项目设置了拦截的话，请放行notify_url地址，因为支付宝发送过来的请求不带认证令牌！
aliPayConfig.py
from datetime import datetime
from Crypto.PublicKey import RSA
from Crypto.Signature import PKCS1_v1_5
from Crypto.Hash import SHA256
from urllib.parse import quote_plus
from urllib.parse import urlparse, parse_qs
from base64 import decodebytes, encodebytes
import json
class AliPay(object):
"""
支付宝支付接口(PC端支付接口)
"""
def __init__(self, appid, app_notify_url, app_private_key_path,
alipay_public_key_path, return_url, debug=False):
self.appid = appid
self.app_notify_url = app_notify_url
self.app_private_key_path = app_private_key_path
self.app_private_key = None
self.return_url = return_url
with open(self.app_private_key_path) as fp:
self.app_private_key = RSA.importKey(fp.read())
self.alipay_public_key_path = alipay_public_key_path
with open(self.alipay_public_key_path) as fp:
self.alipay_public_key = RSA.importKey(fp.read())
if debug is True:
self.__gateway = "https://openapi.alipaydev.com/gateway.do"
else:
self.__gateway = "https://openapi.alipay.com/gateway.do"
def direct_pay(self, subject, out_trade_no, total_amount, return_url=None, **kwargs):
biz_content = {
"subject": subject,
"out_trade_no": out_trade_no,
"total_amount": total_amount,
"product_code": "FAST_INSTANT_TRADE_PAY", # 默认不需要修改
# "qr_pay_mode":4
}
biz_content.update(kwargs)
data = self.build_body("alipay.trade.page.pay", biz_content, self.return_url)
return self.sign_data(data)
def build_body(self, method, biz_content, return_url=None):
data = {
"app_id": self.appid,
"method": method,
"charset": "utf-8",
"sign_type": "RSA2",
"timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"version": "1.0",
"biz_content": biz_content
}
if return_url is not None:
data["notify_url"] = self.app_notify_url
data["return_url"] = self.return_url
return data
def sign_data(self, data):
data.pop("sign", None)
# 排序后的字符串
unsigned_items = self.ordered_data(data)
unsigned_string = "&".join("{0}={1}".format(k, v) for k, v in unsigned_items)
sign = self.sign(unsigned_string.encode("utf-8"))
# ordered_items = self.ordered_data(data)
quoted_string = "&".join("{0}={1}".format(k, quote_plus(v)) for k, v in unsigned_items)
# 获得最终的订单信息字符串
signed_string = quoted_string + "&sign=" + quote_plus(sign)
return signed_string
def ordered_data(self, data):
complex_keys = []
for key, value in data.items():
if isinstance(value, dict):
complex_keys.append(key)
# 将字典类型的数据dump出来
for key in complex_keys:
data[key] = json.dumps(data[key], separators=(',', ':'))
return sorted([(k, v) for k, v in data.items()])
def sign(self, unsigned_string):
# 开始计算签名
key = self.app_private_key
signer = PKCS1_v1_5.new(key)
signature = signer.sign(SHA256.new(unsigned_string))
# base64 编码，转换为unicode表示并移除回车
sign = encodebytes(signature).decode("utf8").replace("\n", "")
return sign
def _verify(self, raw_content, signature):
# 开始计算签名
key = self.alipay_public_key
signer = PKCS1_v1_5.new(key)
digest = SHA256.new()
digest.update(raw_content.encode("utf8"))
if signer.verify(digest, decodebytes(signature.encode("utf8"))):
return True
return False
def verify(self, data, signature):
if "sign_type" in data:
sign_type = data.pop("sign_type")
# 排序后的字符串
unsigned_items = self.ordered_data(data)
message = "&".join(u"{}={}".format(k, v) for k, v in unsigned_items)
return self._verify(message, signature)
alipayTools.py
from AIChatProject import settings
from chatApp.aliPayConfig import AliPay # 注意是从aliPayConfig.py中引用的，而不是python-alipay-sdk的基础AliPay包
def ali():
curr_alipay = AliPay(
appid=settings.APP_ID, # 应用ID
app_notify_url=settings.NOTIFY_URL, # 异步POST回调地址
return_url=settings.RETURN_URL, # 同步GET回调地址
app_private_key_path=settings.APP_PRIVATE_KEY_PATH, # 应用私钥存放路径
alipay_public_key_path=settings.ALIPAY_PUBLIC_KEY_PATH, # 支付宝公钥存放路径
debug=settings.DEBUG # true是沙箱， false是真实环境
)
return curr_alipay
models.py
class UserInfo(models.Model):
id = models.AutoField(primary_key=True, verbose_name='用户编号')
user_account = models.CharField(verbose_name='账号', max_length=32, unique=True)
user_password = models.CharField(verbose_name='密码', max_length=255)
user_account_balance = models.DecimalField(verbose_name='账户余额', max_digits=10, decimal_places=2, default=10000)
auto_type = (
(1, '本地用户'),
(2, '第三方用户'),
)
user_type = models.SmallIntegerField(verbose_name='用户类型', choices=auto_type, default=1)
class Order(models.Model):
"""订单模型"""
status_choices = (
(0, '未支付'),
(1, '已支付'),
(2, '已取消'),
(3, '超时取消'),
)
pay_choices = (
(1, '支付宝'),
(2, '微信支付'),
)
subject = models.CharField(max_length=150, verbose_name="订单标题")
total_amount = models.DecimalField(max_digits=10, decimal_places=2, verbose_name="订单总价", default=0)
out_trade_no = models.CharField(max_length=64, verbose_name="订单号", unique=True)
trade_no = models.CharField(max_length=64, null=True, verbose_name="流水号")
order_status = models.SmallIntegerField(choices=status_choices, default=0, verbose_name="订单状态")
pay_type = models.SmallIntegerField(choices=pay_choices, default=1, verbose_name="支付方式")
pay_time = models.DateTimeField(null=True, verbose_name="支付时间")
# DO_NOTHING 表示不采取任何操作，当关联的对象被删除时，不执行任何处理动作。
user = models.ForeignKey(UserInfo, related_name='chatapp_userinfo', on_delete=models.DO_NOTHING, db_constraint=False, verbose_name="下单用户")
created_time = models.DateTimeField(auto_now_add=True, verbose_name='创建时间')
views.py
class OrderSerializer(serializers.ModelSerializer):
class Meta:
model = Order
fields = '__all__'
# 一、理论知识第1步
class AliPayView(APIView):
def post(self, request):
# 1.实例化支付宝
curr_alipay = ali()
# 2.获取用户购买的订单
data = request.data
user = request.userInfo
# 3.建立支付宝唤醒pay_url
# 3.1 必要参数（subject, out_trade_no, total_amount）
subject = data.get('subject', '')
out_trade_no = str(uuid.uuid4())
total_amount = data.get('total_amount', '')
# 3.2 调用实例的方法生成请求参数
query_params = curr_alipay.direct_pay(subject, out_trade_no, total_amount)
# 3.3 参数拼接
pay_url = settings.GATEWAY + '?' + query_params
# 4.创建数据库订单
user_order = {
'subject': subject,
'total_amount': total_amount,
'out_trade_no': out_trade_no,
'user': user.id
}
serializer = OrderSerializer(data=user_order)
if serializer.is_valid():
# 5.数据库存储订单
serializer.save()
return Response(R(data=pay_url).to_dict())
else:
return Response(R(error='订单存档失败！').to_dict())
# 二、理论知识第4步：支付宝服务器发送POST请求
# notify_url 异步请求异步通知接口
class PayNotifyView(APIView):
def post(self, request):
curr_alipay = ali()
try:
# 1.获取支付传递多来的参数
result_data = request.data.dict()
# 2.将 sign 取出
signature = result_data.pop('sign')
out_trade_no = result_data.get('out_trade_no')
trade_no = result_data.get('trade_no')
# 3.验签
result = curr_alipay.verify(result_data, signature)
# 确认是否支付成功
if result and result_data["trade_status"] in ("TRADE_SUCCESS", "TRADE_FINISHED"):
# 4.完成订单更新：订单状态、流水号、支付时间
result_order = Order.objects.filter(out_trade_no=out_trade_no).first()
# 4.1 获取订单对应的用户
u_id = result_order.user.id
order_user = UserInfo.objects.filter(id=u_id).first()
print('验签成功，当前用户是：', u_id, order_user.user_account)
# 4.2 校验订单是否重复提交！
if result_order.order_status == 1:
return Response(R(data={'message': '该订单已经完成',
'free_account': order_user.user_account_balance}).to_dict())
# 4.3 修改订单状态
Order.objects.filter(out_trade_no=out_trade_no, order_status=0).update(order_status=1, trade_no=trade_no)
# 5.更新用户的余额
new_free_tokens = order_user.user_account_balance + result_order.total_amount * Constant.PRICE_ONE
print(f'old token {order_user.user_account_balance} new token {new_free_tokens}')
UserInfo.objects.filter(id=u_id).update(user_account_balance=new_free_tokens)
# 完成日志记录
return Response(R(data={'message': '充值成功', 'free_account': new_free_tokens}).to_dict())
else:
# logger.error('%s订单支付失败' % out_trade_no)
print('%s订单支付失败，校验未通过或trade_status不对' % out_trade_no)
except Exception as e:
print('报错--->', e)
pass
print('%s订单支付失败' % out_trade_no)
return Response(R(error='%s订单支付失败' % out_trade_no).to_dict())
![[Lesson 01] TiDB数据库架构概述](https://img-blog.csdnimg.cn/caedd74ee358494ea25ed06e0651c004.png)
