AliPayHelper

ali_pay_helper.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import json
from base64 import encodebytes
from datetime import datetime
from typing import Dict, Any
from urllib.parse import quote_plus

from Crypto.Hash import SHA256
from Crypto.PublicKey import RSA
from Crypto.Signature import PKCS1_v1_5

from ali_pay.config import ALIPAY_GATEWAY, ALIPAY_PRIVATE_KEY, ALIPAY_PUBLIC_KEY, ALIPAY_APP_ID, ALIPAY_SIGN_TYPE

ALIPAY_PRIVATE_KEY = RSA.importKey(f'-----BEGIN RSA PRIVATE KEY-----\n{ALIPAY_PRIVATE_KEY}\n-----END RSA PRIVATE KEY-----')
ALIPAY_PUBLIC_KEY  = RSA.importKey(f'-----BEGIN RSA PRIVATE KEY-----\n{ALIPAY_PUBLIC_KEY }\n-----END RSA PRIVATE KEY-----')

def sign(s: str):
    return encodebytes(
        PKCS1_v1_5.new(ALIPAY_PRIVATE_KEY).sign(SHA256.new(s.encode()))
    ).decode().replace("\n", "")

def sign_data(data: Dict[str, Any]):
    items = sorted(data.items())
    return "&".join(
        "{}={}".format(k, quote_plus(v)) for k, v in (
            items + [('sign', sign("&".join("{}={}".format(k, v) for k, v in items)))]
        )
    )

class AliPayHelper:
    params: Dict[str, Any] = {
        'app_id': ALIPAY_APP_ID,
        'sign_type': ALIPAY_SIGN_TYPE,
        'charset': 'UTF8',
        "version": "1.0",
        'format': 'json'
    }
    gateway: str = ALIPAY_GATEWAY

    @property
    def default_query_params(self):
        return {
            **self.params,
            'timestamp': datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        }

    def page_pay(
        self, subject, out_trade_no, total_amount,
        return_url=None, notify_url=None, **kwargs
    ):
        biz_content = {
            'subject': subject,
            'out_trade_no': out_trade_no,
            'total_amount': total_amount,
            'product_code': 'FAST_INSTANT_TRADE_PAY'
        }
        params = {
            **self.default_query_params,
            'method': 'alipay.trade.page.pay',
            'biz_content': json.dumps(biz_content)
        }
        params.update(**{
            'return_url': return_url,
            'notify_url': notify_url
        })

        keys = ("notify_url", "return_url")
        for key in keys:
            if key in params and not params.get(key, None):
                params.pop(key, None)

        return self.gateway + '?' + sign_data(params)

config.py

# -*- coding: utf-8 -*-
from starlette.config import Config

config = Config(".env")
ALIPAY_PRIVATE_KEY = config(
    "ALIPAY_PRIVATE_KEY", cast=str
)
ALIPAY_PUBLIC_KEY = config(
    "ALIPAY_PUBLIC_KEY", cast=str
)
ALIPAY_GATEWAY = config(
    "ALIPAY_GATEWAY", cast=str
)
ALIPAY_APP_ID = config(
    "ALIPAY_APP_ID", cast=str
)
ALIPAY_SIGN_TYPE = config(
    "ALIPAY_SIGN_TYPE", cast=str
)