2025年4月5日 星期六 乙巳(蛇)年 正月初六 设为首页 加入收藏
rss
您当前的位置:首页 > 计算机 > 编程开发 > Python

python --支付宝

时间:08-16来源:作者:点击数:85
CDSY,CDSY.XYZ

python --支付宝

https://github.com/fzlee/alipay/blob/master/README.zh-hans.md#python-alipay-sdk

支付并签约

第三方

  • import os
  • from alipay import BaseAliPay
  • from alipay.utils import AliPayConfig
  • class MyBaseAlipay(BaseAliPay):
  • '''签约并支付,重写APP支付,初始化支付参数'''
  • def api_alipay_trade_app_pay(self, subject, out_trade_no, total_amount, notify_url=None, **kwargs):
  • biz_content = {
  • "subject": subject,
  • "out_trade_no": out_trade_no,
  • "total_amount": total_amount,
  • "product_code": "CYCLE_PAY_AUTH",
  • "agreement_sign_params": {
  • "personal_product_code": "CYCLE_PAY_AUTH_P",
  • "sign_scene": "INDUSTRY|DIGITAL_MEDIA",
  • "external_agreement_no": "test20190701",
  • "access_params": {"channel": "ALIPAYAPP"},
  • "period_rule_params": {
  • "period_type": "DAY",
  • "period": 30, "execute_time": "2021-09-29",
  • "single_amount": "1000.00",
  • }
  • },
  • }
  • biz_content.update(kwargs)
  • data = self.build_body("alipay.trade.app.pay", biz_content, notify_url=notify_url)
  • return self.sign_data(data)
  • def ali_pay() -> str:
  • BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
  • app_private_key_string = open(os.path.join(BASE_DIR, 'yingyong_siyao.txt')).read()
  • alipay_public_key_string = open(os.path.join(BASE_DIR, 'zhifubao_gongyao.txt')).read()
  • alipay = MyBaseAlipay(
  • appid="2021000116677890",
  • app_notify_url=None,
  • app_private_key_string=app_private_key_string,
  • alipay_public_key_string=alipay_public_key_string,
  • sign_type="RSA2",
  • debug=False,
  • verbose=True,
  • config=AliPayConfig(timeout=15)
  • )
  • order_string = alipay.api_alipay_trade_app_pay(
  • out_trade_no="20161112",
  • total_amount=0.01,
  • subject='蛐蛐竞技',
  • notify_url="https://example.com/notify" # 可选,不填则使用默认 notify url
  • )
  • return str(order_string)

自实现

  • from rest_framework.decorators import api_view
  • from rest_framework.response import Response
  • from datetime import datetime
  • from Crypto.PublicKey import RSA
  • from Crypto.Signature import PKCS1_v1_5
  • from Crypto.Hash import SHA256
  • from base64 import b64encode, b64decode
  • from urllib.parse import quote_plus
  • from urllib.parse import urlparse, parse_qs
  • from urllib.request import urlopen
  • from base64 import decodebytes, encodebytes
  • import json
  • class AliPay(object):
  • """
  • 支付宝支付接口
  • """
  • 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.import_key(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, *args, **kwargs):
  • import time
  • now_date = time.strftime('%Y-%m-%d')
  • biz_content = {
  • "total_amount": "0.01",
  • "subject": "蛐蛐竞技",
  • "out_trade_no": "123456",
  • "agreement_sign_params": {
  • "personal_product_code": "CYCLE_PAY_AUTH_P",
  • "sign_scene": "INDUSTRY|DIGITAL_MEDIA",
  • "access_params": {"channel": "ALIPAYAPP"},
  • "period_rule_params":
  • {"period_type": "DAY", "period": "30", "execute_time": "2021-09-28", "single_amount": 10.99}
  • }}
  • print(biz_content)
  • biz_content.update(kwargs)
  • # data = self.build_body("alipay.trade.page.pay", biz_content, self.return_url)
  • data = self.build_body("alipay.trade.app.pay", biz_content, self.return_url)
  • res = self.sign_data(data)
  • print(f'预请求参数地址为:{res}')
  • return res
  • 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
  • print(f'初始化参数:{data}', '\n')
  • 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 ordered_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)
  • @api_view(['GET', ])
  • def c(request):
  • alipay = AliPay(
  • appid="2021000116677890", # 设置签约的appid
  • app_notify_url="http://projectsedus.com/", # 异步支付通知url
  • app_private_key_path=u"yingyong_siyao.txt", # 设置应用私钥
  • alipay_public_key_path="zhifubao_gongyao.txt", # 支付宝的公钥,验证支付宝回传消息使用,不是你自己的公钥,
  • debug=True, # 默认False, # 设置是否是沙箱环境,True是沙箱环境
  • return_url="http://47.92.87.172:8000/" # 同步支付通知url
  • )
  • # 传递参数执行支付类里的direct_pay方法,返回签名后的支付参数,
  • url = alipay.direct_pay(
  • return_url="http://47.92.87.172:8000/" # 支付成功后,跳转url
  • )
  • re_url = "https://openapi.alipaydev.com/gateway.do?{data}".format(data=url)
  • return Response({'url':re_url})

退款

  • from alipay import BaseAliPay, AliPay
  • from alipay.utils import AliPayConfig
  • app_private_key_string = open(r"D:/yi/ddd/yingyong_siyao.txt").read()
  • alipay_public_key_string = open(r"D:/yi/ddd/zhifubao_gongyao.txt").read()
  • alipay = AliPay(
  • appid="2021002175672102",
  • app_notify_url=None, # 默认回调 url
  • app_private_key_string=app_private_key_string,
  • # 支付宝的公钥,验证支付宝回传消息使用,不是你自己的公钥,
  • alipay_public_key_string=alipay_public_key_string,
  • sign_type="RSA2", # RSA 或者 RSA2
  • debug=False, # 默认 False
  • verbose=True, # 输出调试数据
  • config=AliPayConfig(timeout=15) # 可选,请求超时时间
  • )
  • result = alipay.api_alipay_trade_refund(out_trade_no="OW1632640656", refund_amount="30")
  • print(result)

pc端自实现

为什么使用三方支付?

再没有三方支付平台之前,用户发起支付请求的时候,用户要去和银行签约(转账),特别的不方便,为了解决这些问题,就有了三方支付,三方平台去完成签约,给用户节省时间。

支付宝支付的流程

商户拿到支付宝的公钥、自己的私钥(私钥加密、公钥解密),用私钥请求支付宝,支付宝解密、验签、进行支付处理,支付宝将处理的返回值传给商户,当支付成功后,返还给商户订单号、金额、时间戳等消息,支付失败后同样给商户反馈结果。

配置流程

获取APPID

  • 支付宝开放平台:https://open.alipay.com/

登录支付宝开放平台–>点击控制台

在这里插入图片描述

点击沙箱(复制APPID)

在这里插入图片描述

在线生成密钥

点击文档,找到开发助手,点击在线加密。

在这里插入图片描述

获取私钥

在这里插入图片描述

获取公钥

点击应用公钥

在这里插入图片描述

生成公钥

在这里插入图片描述

现在已经拿到了需要的公钥。

Python项目中集成支付宝

构建支付类

  • 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 base64 import decodebytes, encodebytes
  • import json
  • class AliPay:
  • """
  • 支付宝支付接口(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",
  • }
  • 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"))
  • 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)
  • 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))
  • 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:
  • 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)

实例化类

  • def init_alipay():
  • # 初始化Alipay
  • alipay = AliPay(
  • appid="appid",
  • app_notify_url="回调地址",
  • return_url="回调地址",
  • app_private_key_path="私钥相对路径",
  • alipay_public_key_path="公钥相对路径",
  • debug=True # 支付环境
  • )
  • return alipay

API

  • async def get(self):
  • alipay = init_alipay()
  • # 传一个标题 订单号 订单价格
  • params = alipay.direct_pay("三方广告平台", order_no, money)
  • url = f"https://openapi.alipaydev.com/gateway.do?{params}"
  • return self.write(ret_json(url))
  • # 构建一个回调地址,用于支付成功后回调,在回调地址中可以获取订单号(out_trade_no)、金额(total_amount)、时间戳(timestamp),然后进行处理业务逻辑。
CDSY,CDSY.XYZ
方便获取更多学习、工作、生活信息请关注本站微信公众号城东书院 微信服务号城东书院 微信订阅号
推荐内容
相关内容
栏目更新
栏目热门
本栏推荐