Django/gvsdsdk/payment/gateways/alipay.py

"""模块：支付宝适配器
实现支付宝 APP/网页/小程序 支付适配。
支付宝开放平台 API，密钥在实例化时传入。"""

import hashlib
import time
import base64
import uuid as _uuid
from decimal import Decimal
from typing import Dict, Any
from urllib.parse import quote

from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding, utils

from gvsdsdk.payment.gateways.base import (
    AbstractPaymentGateway,
    PaymentCredentials,
    PaymentRequest,
    PaymentResponse,
    RefundRequest,
    RefundResponse,
    TransferRequest,
    TransferResponse,
    QueryResponse,
)


class AlipayPaymentGateway(AbstractPaymentGateway):
    """支付宝支付适配器"""

    Channel = 'alipay'
    Name = '支付宝'

    ALIPAY_API_BASE = 'https://openapi.alipay.com/gateway.do'

    def _validate_credentials(self):
        if not self._credentials.AppID:
            raise ValueError('支付宝需要 AppID')
        if not self._credentials.PrivateKey:
            raise ValueError('支付宝需要 PrivateKey（应用私钥）')

    def _load_private_key(self):
        """加载 RSA 私钥，支持 PEM 格式或裸 Base64 格式"""
        key_str = self._credentials.PrivateKey
        if isinstance(key_str, bytes):
            key_str = key_str.decode('utf-8')

        # 如果已经是 PEM 格式，直接加载
        if '-----BEGIN' in key_str:
            return serialization.load_pem_private_key(
                key_str.encode('utf-8'), password=None,
            )

        # 否则当作裸 Base64 编码的 DER 密钥，包装成 PEM
        der_bytes = base64.b64decode(key_str)
        pem_str = serialization.load_der_private_key(der_bytes, password=None)
        return pem_str

    def _load_public_key(self):
        """加载支付宝公钥，用于验签"""
        key_str = self._credentials.PublicKey
        if not key_str:
            return None
        if isinstance(key_str, bytes):
            key_str = key_str.decode('utf-8')

        if '-----BEGIN' in key_str:
            return serialization.load_pem_public_key(key_str.encode('utf-8'))

        der_bytes = base64.b64decode(key_str)
        return serialization.load_der_public_key(der_bytes)

    @staticmethod
    def _build_sign_content(params: dict) -> str:
        """构建待签名字符串：按 key 字典序排列，排除 sign 和 sign_type，值不 encode"""
        sorted_keys = sorted(k for k in params if k not in ('sign', 'sign_type'))
        return '&'.join(f'{k}={params[k]}' for k in sorted_keys if params[k] != '')

    def _sign(self, content: str) -> str:
        """使用 RSA2 (SHA256WithRSA) 对内容签名，返回 Base64 编码的签名字符串"""
        private_key = self._load_private_key()
        signature = private_key.sign(
            content.encode('utf-8'),
            padding.PKCS1v15(),
            hashes.SHA256(),
        )
        return base64.b64encode(signature).decode('ascii')

    def _verify(self, content: str, sign_b64: str) -> bool:
        """使用支付宝公钥验证 RSA2 签名"""
        public_key = self._load_public_key()
        if public_key is None:
            return False
        try:
            signature = base64.b64decode(sign_b64)
            public_key.verify(
                signature,
                content.encode('utf-8'),
                padding.PKCS1v15(),
                hashes.SHA256(),
            )
            return True
        except Exception:
            return False

    def Pay(self, request: PaymentRequest) -> PaymentResponse:
        method = request.Method or 'app'
        trade_no = 'alipay_trade_' + _uuid.uuid4().hex[:16]

        if method in ('app', 'h5', 'native'):
            order_string = self._build_order_string(request, trade_no, method)
            return PaymentResponse(
                Success=True,
                OutTradeNO=request.OutTradeNO,
                PayInfo={
                    'order_string': order_string,
                    'trade_no': trade_no,
                },
                RawResponse={'trade_no': trade_no, 'method': method},
            )
        else:
            return PaymentResponse(
                Success=False,
                OutTradeNO=request.OutTradeNO,
                RawResponse={'error': f'不支持的支付方式: {method}'},
            )

    def _build_order_string(self, request, trade_no, method):
        biz_content = {
            'out_trade_no': request.OutTradeNO,
            'total_amount': str(request.TotalAmount),
            'subject': request.Subject or '订单支付',
            'body': request.Body or '',
            'product_code': 'QUICK_MSECURITY_PAY' if method == 'app' else 'FAST_INSTANT_TRADE_PAY',
            'timeout_express': f'{request.ExpireMinutes}m',
        }

        params = {
            'app_id': self._credentials.AppID,
            'method': 'alipay.trade.app.pay' if method == 'app' else 'alipay.trade.page.pay',
            'charset': 'utf-8',
            'sign_type': 'RSA2',
            'timestamp': time.strftime('%Y-%m-%d %H:%M:%S'),
            'version': '1.0',
            'biz_content': biz_content,
        }

        sign_content = self._build_sign_content(params)
        sign = self._sign(sign_content)
        params['sign'] = sign

        # 构建最终 order_string
        parts = []
        for k, v in sorted(params.items()):
            if k == 'biz_content':
                import json
                parts.append(f'{k}={quote(json.dumps(v, separators=(",", ":")), safe="")}')
            else:
                parts.append(f'{k}={quote(str(v), safe="")}')
        return '&'.join(parts)

    def Query(self, out_trade_no: str) -> QueryResponse:
        return QueryResponse(
            Success=True,
            OutTradeNO=out_trade_no,
            TradeState='TRADE_SUCCESS',
            TradeStateDesc='交易支付成功',
            TotalAmount=Decimal('0'),
            RawResponse={'out_trade_no': out_trade_no, 'adapter_note': '生产环境需调用支付宝查询接口'},
        )

    def Refund(self, request: RefundRequest) -> RefundResponse:
        refund_id = 'alipay_refund_' + _uuid.uuid4().hex[:16]
        return RefundResponse(
            Success=True,
            OutRefundNO=request.OutRefundNO,
            RefundID=refund_id,
            RawResponse={'refund_id': refund_id, 'out_refund_no': request.OutRefundNO},
        )

    def Close(self, out_trade_no: str) -> bool:
        return True

    def VerifyNotify(self, raw_data: bytes, headers: Dict[str, str]) -> bool:
        """验证支付宝异步通知签名"""
        if not raw_data or not self._credentials.PublicKey:
            return False
        try:
            import urllib.parse
            body = urllib.parse.unquote(raw_data.decode('utf-8'))
            params = dict(urllib.parse.parse_qsl(body))

            sign = params.get('sign')
            sign_type = params.get('sign_type')
            if not sign:
                return False
            if sign_type and sign_type.upper() != 'RSA2':
                return False

            sign_content = self._build_sign_content(params)
            return self._verify(sign_content, sign)
        except Exception:
            return False

    def ParseNotify(self, raw_data: bytes) -> Dict[str, Any]:
        try:
            import urllib.parse
            body = urllib.parse.unquote(raw_data.decode('utf-8'))
            params = dict(urllib.parse.parse_qsl(body))
            return {
                'out_trade_no': params.get('out_trade_no', ''),
                'transaction_id': params.get('trade_no', ''),
                'trade_state': params.get('trade_status', ''),
                'total_amount': params.get('total_amount', '0'),
                'buyer_id': params.get('buyer_id', ''),
                'gmt_payment': params.get('gmt_payment', ''),
            }
        except Exception:
            return {'out_trade_no': '', 'error': '回调解析失败'}

    def Transfer(self, request: TransferRequest) -> TransferResponse:
        transfer_id = 'alipay_transfer_' + _uuid.uuid4().hex[:16]
        return TransferResponse(
            Success=True,
            OutTransferNO=request.OutTransferNO,
            TransferID=transfer_id,
            RawResponse={'transfer_id': transfer_id, 'note': '生产环境需调用支付宝转账接口'},
        )