unipay/unipay/providers/paypal.py

# providers/paypal.py
import aiohttp
import asyncio
import typing as t
from abc import ABC, abstractmethod
import os
from ..core import Gateway, GatewayError
from ..utils import safe_json_dumps

class PayPalGateway(Gateway):
	def __init__(self, client_id: str, client_secret: str, sandbox: bool = True):
		self.client_id = client_id
		self.client_secret = client_secret
		self.sandbox = sandbox
		self.base_url = "https://api-m.sandbox.paypal.com" if sandbox else "https://api-m.paypal.com"
		self._token: t.Optional[str] = None
		self._token_expiry: t.Optional[float] = None

	async def _get_access_token(self) -> str:
		if self._token and self._token_expiry and self._token_expiry > asyncio.get_event_loop().time():
			return self._token

		url = f"{self.base_url}/v1/oauth2/token"
		auth = aiohttp.BasicAuth(self.client_id, self.client_secret)
		data = {"grant_type": "client_credentials"}
		async with aiohttp.ClientSession() as session:
			async with session.post(url, data=data, auth=auth) as resp:
				resp.raise_for_status()
				r = await resp.json()
				self._token = r["access_token"]
				self._token_expiry = asyncio.get_event_loop().time() + r.get("expires_in", 300) - 10
				return self._token

	async def create_payment(self, payload: t.Dict[str, t.Any]) -> str:
		"""
		创建 PayPal 订单
		payload 必须包含：
		- amount: 数值
		- currency: 币种
		- description / subject
		- return_url
		- notify_url
		- out_trade_no: 商户订单号
		"""
		access_token = await self._get_access_token()
		url = f"{self.base_url}/v2/checkout/orders"

		# PayPal 订单 body
		body = {
			"intent": "CAPTURE",
			"purchase_units": [
				{
					"amount": {
						"currency_code": payload["currency"],
						"value": f"{payload['amount']:.2f}"
					},
					"custom_id": payload["out_trade_no"],  # 保留商户交易号
					"description": payload.get("description") or payload.get("subject", "")
				}
			],
			"application_context": {
				"return_url": payload["return_url"],
				"cancel_url": payload.get("cancel_url", payload["return_url"])
			}
		}

		async with aiohttp.ClientSession() as session:
			async with session.post(url, json=body, headers={"Authorization": f"Bearer {access_token}"}) as resp:
				resp.raise_for_status()
				data = await resp.json()
				# 获取 PayPal 端交易号
				transaction_id = data["id"]
				# 获取跳转 URL
				approval_url = next((link["href"] for link in data.get("links", []) if link.get("rel") == "approve"), None)
				return approval_url
				# return {"provider": "paypal", "data": data, "out_trade_no": payload["out_trade_no"], "transaction_id": transaction_id}

	async def refund(self, payload: t.Dict[str, t.Any]) -> t.Dict[str, t.Any]:
		"""
		payload:
		- transaction_id: PayPal 端交易号
		- amount
		- out_refund_no
		"""
		access_token = await self._get_access_token()
		url = f"{self.base_url}/v2/payments/captures/{payload['transaction_id']}/refund"
		body = {"amount": {"value": f"{payload['amount']:.2f}", "currency_code": payload.get("currency", "USD")}}

		async with aiohttp.ClientSession() as session:
			async with session.post(url, json=body, headers={"Authorization": f"Bearer {access_token}"}) as resp:
				resp.raise_for_status()
				r = await resp.json()
				return {
					"out_trade_no": payload.get("out_trade_no"),
					"refund_id": r.get("id"),
					"status": r.get("status")
				}

	async def query(self, payload: t.Dict[str, t.Any]) -> t.Dict[str, t.Any]:
		"""
		查询订单状态
		payload:
		- transaction_id: PayPal 端交易号
		"""
		access_token = await self._get_access_token()
		url = f"{self.base_url}/v2/checkout/orders/{payload['transaction_id']}"
		async with aiohttp.ClientSession() as session:
			async with session.get(url, headers={"Authorization": f"Bearer {access_token}"}) as resp:
				resp.raise_for_status()
				r = await resp.json()
				out_trade_no = None
				if "purchase_units" in r and len(r["purchase_units"]) > 0:
					out_trade_no = r["purchase_units"][0].get("custom_id")
				return {
					"out_trade_no": out_trade_no,
					"transaction_id": r.get("id"),
					"status": r.get("status"),
					"data": r
				}

	async def handle_notify(self, headers: t.Dict[str,str], body: str) -> t.Dict[str, t.Any]:
		"""
		PayPal webhook 处理
		返回 dict 包含：
		- out_trade_no: 商户交易号
		- transaction_id: PayPal 端交易号
		- status: COMPLETED/REFUNDED 等
		- attach: webhook 原始数据
		"""
		import json
		event = json.loads(body)
		out_trade_no = None
		transaction_id = None
		if event.get("resource"):
			transaction_id = event["resource"].get("id")
			# 先取 custom_id（创建订单时传入）
			pu = event["resource"].get("purchase_units", [])
			if pu and len(pu) > 0:
				out_trade_no = pu[0].get("custom_id")
			# fallback: invoice_id
			if not out_trade_no and event["resource"].get("invoice_id"):
				out_trade_no = event["resource"]["invoice_id"]

		return {
			"provider": "paypal",
			"out_trade_no": out_trade_no,
			"transaction_id": transaction_id,
			"status": event.get("event_type"),
			"attach": event
		}