628 lines
23 KiB
Python
628 lines
23 KiB
Python
"""微信支付工具模块
|
|
|
|
支持微信支付 V3 API
|
|
文档参考: https://pay.weixin.qq.com/docs/v3/apis/index.html
|
|
|
|
使用方式:
|
|
from app.core.wechat_pay import wechat_pay, WechatPayClient
|
|
|
|
# 统一下单
|
|
result = wechat_pay.unified_order(out_trade_no, total_amount, description, notify_url)
|
|
|
|
# 申请退款
|
|
result = wechat_pay.refund(transaction_id, refund_amount, total_amount)
|
|
"""
|
|
|
|
import hashlib
|
|
import time
|
|
import random
|
|
import string
|
|
import json
|
|
import base64
|
|
import xml.etree.ElementTree as ET
|
|
from typing import Optional, Dict, Any
|
|
from datetime import datetime
|
|
from urllib.parse import urljoin
|
|
import ssl
|
|
import urllib.request
|
|
import urllib.error
|
|
|
|
from app.core.config import settings
|
|
|
|
|
|
class WechatPayClient:
|
|
"""微信支付客户端"""
|
|
|
|
def __init__(self):
|
|
self.mchid = settings.WXPAY_MCHID
|
|
self.appid = settings.WXPAY_APPID
|
|
self.api_key = settings.WXPAY_API_KEY
|
|
self.cert_path = settings.WXPAY_CERT_PATH
|
|
self.private_key_path = settings.WXPAY_PRIVATE_KEY_PATH
|
|
self.sandbox = settings.WXPAY_SANDBOX
|
|
|
|
# API 版本
|
|
self.api_base = "https://api.mchpay.qpic.cn/v3" if self.sandbox else "https://api.mchpay.qpic.cn/v3"
|
|
|
|
def _generate_nonce_str(self, length: int = 32) -> str:
|
|
"""生成随机字符串"""
|
|
return ''.join(random.choices(string.ascii_letters + string.digits, k=length))
|
|
|
|
def _generate_sign(self, data: Dict[str, Any]) -> str:
|
|
"""生成签名 (V2 API)"""
|
|
# V2 签名算法
|
|
sorted_keys = sorted(data.keys())
|
|
sign_str = "&".join([f"{k}={data[k]}" for k in sorted_keys if data[k]])
|
|
sign_str += f"&key={self.api_key}"
|
|
return hashlib.md5(sign_str.encode('utf-8')).hexdigest().upper()
|
|
|
|
def _generate_sign_v3(self, method: str, url: str, timestamp: str, nonce: str, body: str = "") -> str:
|
|
"""生成签名 V3 (SHA256 with RSA)"""
|
|
try:
|
|
from cryptography.hazmat.primitives import hashes, serialization
|
|
from cryptography.hazmat.primitives.asymmetric import padding
|
|
from cryptography.hazmat.backends import default_backend
|
|
from cryptography.x509 import load_pem_x509_certificate
|
|
|
|
with open(self.private_key_path, 'r') as f:
|
|
private_key = f.read()
|
|
|
|
# 构建签名串
|
|
sign_str = f"{method}\n{url}\n{timestamp}\n{nonce}\n{body}\n"
|
|
|
|
# 使用私钥签名
|
|
from cryptography.hazmat.primitives import hashes
|
|
from cryptography.hazmat.primitives.asymmetric import padding
|
|
|
|
key = serialization.load_pem_private_key(
|
|
private_key.encode(),
|
|
password=None,
|
|
backend=default_backend()
|
|
)
|
|
|
|
signature = key.sign(
|
|
sign_str.encode('utf-8'),
|
|
padding.PKCS1v15(),
|
|
hashes.SHA256()
|
|
)
|
|
|
|
return base64.b64encode(signature).decode('utf-8')
|
|
except ImportError:
|
|
raise ImportError("请安装 cryptography: pip install cryptography")
|
|
|
|
def _make_request(self, url: str, method: str = "GET", data: Dict = None, need_cert: bool = False) -> Dict:
|
|
"""发送HTTP请求"""
|
|
headers = {
|
|
'Content-Type': 'application/json',
|
|
'User-Agent': 'SQXP/1.0'
|
|
}
|
|
|
|
body = json.dumps(data) if data else ""
|
|
|
|
try:
|
|
if need_cert and self.cert_path and self.private_key_path:
|
|
import ssl
|
|
context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
|
|
context.load_cert_chain(self.cert_path, self.private_key_path)
|
|
req = urllib.request.Request(url, data=body.encode(), headers=headers, method=method)
|
|
with urllib.request.urlopen(req, context=context) as response:
|
|
return json.loads(response.read().decode())
|
|
else:
|
|
req = urllib.request.Request(url, data=body.encode(), headers=headers, method=method)
|
|
with urllib.request.urlopen(req) as response:
|
|
return json.loads(response.read().decode())
|
|
except urllib.error.HTTPError as e:
|
|
error_body = e.read().decode() if e.fp else ""
|
|
try:
|
|
return json.loads(error_body)
|
|
except:
|
|
return {"error": error_body, "code": e.code}
|
|
except Exception as e:
|
|
return {"error": str(e), "code": -1}
|
|
|
|
def unified_order(
|
|
self,
|
|
out_trade_no: str,
|
|
total_amount: int,
|
|
description: str,
|
|
notify_url: str,
|
|
openid: str = None
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
统一下单接口 (V3)
|
|
|
|
Args:
|
|
out_trade_no: 商户订单号
|
|
total_amount: 订单金额(分)
|
|
description: 商品描述
|
|
notify_url: 回调通知地址
|
|
openid: 用户openid (JSAPI支付必需)
|
|
|
|
Returns:
|
|
返回支付参数，用于调起微信支付
|
|
"""
|
|
timestamp = str(int(time.time()))
|
|
nonce_str = self._generate_nonce_str()
|
|
|
|
url = f"{self.api_base}/transactions/jsapi"
|
|
|
|
data = {
|
|
"appid": self.appid,
|
|
"mchid": self.mchid,
|
|
"description": description,
|
|
"out_trade_no": out_trade_no,
|
|
"time_expire": datetime.now().strftime("%Y-%m-%dT%H:%M:%S+08:00"),
|
|
"attach": "sqxp_order",
|
|
"notify_url": notify_url,
|
|
"amount": {
|
|
"total": total_amount,
|
|
"currency": "CNY"
|
|
},
|
|
"payer": {
|
|
"openid": openid
|
|
} if openid else None
|
|
}
|
|
|
|
# 移除None值
|
|
data = {k: v for k, v in data.items() if v is not None}
|
|
|
|
body = json.dumps(data)
|
|
sign = self._generate_sign_v3("POST", f"/v3/transactions/jsapi", timestamp, nonce_str, body)
|
|
|
|
headers = {
|
|
"Authorization": f'WECHATPAY2-SHA256-RSA2048 mchid="{self.mchid}",nonce_str="{nonce_str}",signature="{sign}",timestamp="{timestamp}",serial_no="{settings.WXPAY_SERIAL_NO}"',
|
|
"Content-Type": "application/json",
|
|
"Accept": "application/json"
|
|
}
|
|
|
|
try:
|
|
import ssl
|
|
context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
|
|
context.check_hostname = False
|
|
context.verify_mode = ssl.CERT_NONE
|
|
|
|
if self.cert_path and self.private_key_path:
|
|
context.load_cert_chain(self.cert_path, self.private_key_path)
|
|
|
|
req = urllib.request.Request(url, data=body.encode(), headers=headers, method="POST")
|
|
with urllib.request.urlopen(req, context=context, timeout=30) as response:
|
|
result = json.loads(response.read().decode())
|
|
|
|
if "prepay_id" in result:
|
|
# 返回调起支付的参数
|
|
pay_sign = self._generate_pay_sign(result["prepay_id"])
|
|
return {
|
|
"success": True,
|
|
"prepay_id": result["prepay_id"],
|
|
"pay_sign": pay_sign,
|
|
"timestamp": timestamp,
|
|
"nonce_str": nonce_str
|
|
}
|
|
else:
|
|
return {"success": False, "error": result.get("message", "下单失败")}
|
|
except Exception as e:
|
|
return {"success": False, "error": str(e)}
|
|
|
|
def _generate_pay_sign(self, prepay_id: str) -> str:
|
|
"""生成支付签名 (小程序调起支付用)"""
|
|
timestamp = str(int(time.time()))
|
|
nonce_str = self._generate_nonce_str()
|
|
|
|
sign_str = f"{self.appid}\n{timestamp}\n{nonce_str}\nprepay_id={prepay_id}\n"
|
|
return hashlib.sha256(sign_str.encode('utf-8')).hexdigest()
|
|
|
|
def query_order(self, out_trade_no: str = None, transaction_id: str = None) -> Dict:
|
|
"""查询订单"""
|
|
if not out_trade_no and not transaction_id:
|
|
return {"success": False, "error": "缺少订单号"}
|
|
|
|
timestamp = str(int(time.time()))
|
|
nonce_str = self._generate_nonce_str()
|
|
|
|
if out_trade_no:
|
|
url = f"{self.api_base}/v3/pay/transactions/out-trade-no/{out_trade_no}?mchid={self.mchid}"
|
|
else:
|
|
url = f"{self.api_base}/v3/pay/transactions/id/{transaction_id}?mchid={self.mchid}"
|
|
|
|
headers = {
|
|
"Authorization": f'WECHATPAY2-SHA256-RSA2048 mchid="{self.mchid}",nonce_str="{nonce_str}",signature="",timestamp="{timestamp}",serial_no=""',
|
|
"Accept": "application/json"
|
|
}
|
|
|
|
try:
|
|
import ssl
|
|
context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
|
|
context.check_hostname = False
|
|
context.verify_mode = ssl.CERT_NONE
|
|
|
|
req = urllib.request.Request(url, headers=headers, method="GET")
|
|
with urllib.request.urlopen(req, context=context, timeout=30) as response:
|
|
result = json.loads(response.read().decode())
|
|
return {"success": True, "data": result}
|
|
except urllib.error.HTTPError as e:
|
|
return {"success": False, "error": f"HTTP {e.code}: {e.read().decode()}"}
|
|
except Exception as e:
|
|
return {"success": False, "error": str(e)}
|
|
|
|
def refund(
|
|
self,
|
|
out_trade_no: str,
|
|
total_amount: int,
|
|
refund_amount: int,
|
|
reason: str = None
|
|
) -> Dict:
|
|
"""
|
|
申请退款
|
|
|
|
Args:
|
|
out_trade_no: 商户订单号
|
|
total_amount: 订单总金额(分)
|
|
refund_amount: 退款金额(分)
|
|
reason: 退款原因
|
|
|
|
Returns:
|
|
退款结果
|
|
"""
|
|
if refund_amount > total_amount:
|
|
return {"success": False, "error": "退款金额不能大于订单金额"}
|
|
|
|
timestamp = str(int(time.time()))
|
|
nonce_str = self._generate_nonce_str()
|
|
out_refund_no = f"REF{datetime.now().strftime('%Y%m%d%H%M%S')}{random.randint(1000, 9999)}"
|
|
|
|
url = f"{self.api_base}/v3/refund/domestic/refunds"
|
|
|
|
data = {
|
|
"out_trade_no": out_trade_no,
|
|
"out_refund_no": out_refund_no,
|
|
"reason": reason or "用户申请退款",
|
|
"notify_url": settings.WXPAY_REFUND_NOTIFY_URL,
|
|
"amount": {
|
|
"refund": refund_amount,
|
|
"total": total_amount,
|
|
"currency": "CNY"
|
|
}
|
|
}
|
|
|
|
body = json.dumps(data)
|
|
sign = self._generate_sign_v3("POST", "/v3/refund/domestic/refunds", timestamp, nonce_str, body)
|
|
|
|
headers = {
|
|
"Authorization": f'WECHATPAY2-SHA256-RSA2048 mchid="{self.mchid}",nonce_str="{nonce_str}",signature="{sign}",timestamp="{timestamp}",serial_no="{settings.WXPAY_SERIAL_NO}"',
|
|
"Content-Type": "application/json",
|
|
"Accept": "application/json"
|
|
}
|
|
|
|
try:
|
|
import ssl
|
|
context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
|
|
context.check_hostname = False
|
|
context.verify_mode = ssl.CERT_NONE
|
|
|
|
if self.cert_path and self.private_key_path:
|
|
context.load_cert_chain(self.cert_path, self.private_key_path)
|
|
|
|
req = urllib.request.Request(url, data=body.encode(), headers=headers, method="POST")
|
|
with urllib.request.urlopen(req, context=context, timeout=30) as response:
|
|
result = json.loads(response.read().decode())
|
|
return {
|
|
"success": True,
|
|
"refund_id": result.get("refund_id"),
|
|
"out_refund_no": out_refund_no,
|
|
"refund_status": result.get("status")
|
|
}
|
|
except urllib.error.HTTPError as e:
|
|
error_body = e.read().decode()
|
|
return {"success": False, "error": f"HTTP {e.code}: {error_body}"}
|
|
except Exception as e:
|
|
return {"success": False, "error": str(e)}
|
|
|
|
def close_order(self, out_trade_no: str) -> Dict:
|
|
"""关闭订单"""
|
|
timestamp = str(int(time.time()))
|
|
nonce_str = self._generate_nonce_str()
|
|
|
|
url = f"{self.api_base}/v3/transactions/out-trade-no/{out_trade_no}/close"
|
|
|
|
data = {"mchid": self.mchid}
|
|
body = json.dumps(data)
|
|
sign = self._generate_sign_v3("POST", f"/v3/transactions/out-trade-no/{out_trade_no}/close", timestamp, nonce_str, body)
|
|
|
|
headers = {
|
|
"Authorization": f'WECHATPAY2-SHA256-RSA2048 mchid="{self.mchid}",nonce_str="{nonce_str}",signature="{sign}",timestamp="{timestamp}",serial_no="{settings.WXPAY_SERIAL_NO}"',
|
|
"Content-Type": "application/json"
|
|
}
|
|
|
|
try:
|
|
import ssl
|
|
context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
|
|
context.check_hostname = False
|
|
context.verify_mode = ssl.CERT_NONE
|
|
|
|
if self.cert_path and self.private_key_path:
|
|
context.load_cert_chain(self.cert_path, self.private_key_path)
|
|
|
|
req = urllib.request.Request(url, data=body.encode(), headers=headers, method="POST")
|
|
with urllib.request.urlopen(req, context=context, timeout=30) as response:
|
|
return {"success": True}
|
|
except urllib.error.HTTPError as e:
|
|
return {"success": False, "error": f"HTTP {e.code}: {e.read().decode()}"}
|
|
except Exception as e:
|
|
return {"success": False, "error": str(e)}
|
|
|
|
|
|
def get_wechat_pay_client() -> WechatPayClient:
|
|
"""获取微信支付客户端实例"""
|
|
return WechatPayClient()
|
|
|
|
|
|
# 全局实例 (延迟初始化)
|
|
_wechat_pay_client: Optional[WechatPayClient] = None
|
|
|
|
|
|
def wechat_pay() -> WechatPayClient:
|
|
"""获取微信支付客户端"""
|
|
global _wechat_pay_client
|
|
if _wechat_pay_client is None:
|
|
_wechat_pay_client = WechatPayClient()
|
|
return _wechat_pay_client
|
|
|
|
|
|
class WechatPayV2Client:
|
|
"""微信支付 V2 客户端 (兼容旧版)"""
|
|
|
|
def __init__(self):
|
|
self.mchid = settings.WXPAY_MCHID
|
|
self.appid = settings.WXPAY_APPID
|
|
self.api_key = settings.WXPAY_API_KEY
|
|
self.appid_mini = settings.WXPAY_MINI_APPID
|
|
self.notify_url = settings.WXPAY_NOTIFY_URL
|
|
|
|
if settings.WXPAY_SANDBOX:
|
|
self.pay_url = "https://sandbox.mch.weixin.qq.com/pay/unifiedorder"
|
|
else:
|
|
self.pay_url = "https://api.mch.weixin.qq.com/pay/unifiedorder"
|
|
|
|
def _generate_sign(self, params: Dict) -> str:
|
|
"""生成签名"""
|
|
sorted_keys = sorted(params.keys())
|
|
sign_str = "&".join([f"{k}={params[k]}" for k in sorted_keys if params[k]])
|
|
sign_str += f"&key={self.api_key}"
|
|
return hashlib.md5(sign_str.encode('utf-8')).hexdigest().upper()
|
|
|
|
def _dict_to_xml(self, params: Dict) -> str:
|
|
"""字典转XML"""
|
|
xml_str = "<xml>"
|
|
for k, v in params.items():
|
|
xml_str += f"<{k}><![CDATA[{v}]]></{k}>"
|
|
xml_str += "</xml>"
|
|
return xml_str
|
|
|
|
def _xml_to_dict(self, xml_str: str) -> Dict:
|
|
"""XML转字典"""
|
|
try:
|
|
root = ET.fromstring(xml_str)
|
|
return {child.tag: child.text for child in root}
|
|
except:
|
|
return {}
|
|
|
|
def build_miniprogram_pay_params(self, prepay_id: str) -> Dict[str, str]:
|
|
"""小程序调起支付的二次签名参数（与公众号 JSAPI 规则一致）"""
|
|
appid = self.appid_mini or self.appid
|
|
time_stamp = str(int(time.time()))
|
|
nonce_str = ''.join(random.choices(string.ascii_letters + string.digits, k=32))
|
|
package = f"prepay_id={prepay_id}"
|
|
sign_type = "MD5"
|
|
pay_fields = {
|
|
"appId": appid,
|
|
"nonceStr": nonce_str,
|
|
"package": package,
|
|
"signType": sign_type,
|
|
"timeStamp": time_stamp,
|
|
}
|
|
sign_str = "&".join([f"{k}={pay_fields[k]}" for k in sorted(pay_fields.keys())])
|
|
sign_str += f"&key={self.api_key}"
|
|
pay_sign = hashlib.md5(sign_str.encode("utf-8")).hexdigest().upper()
|
|
return {
|
|
"appId": appid,
|
|
"timeStamp": time_stamp,
|
|
"nonceStr": nonce_str,
|
|
"package": package,
|
|
"signType": sign_type,
|
|
"paySign": pay_sign,
|
|
}
|
|
|
|
def unified_order(
|
|
self,
|
|
out_trade_no: str,
|
|
total_fee: int,
|
|
body: str,
|
|
openid: str = None,
|
|
trade_type: str = "JSAPI"
|
|
) -> Dict:
|
|
"""
|
|
统一下单 (V2 API)
|
|
|
|
Args:
|
|
out_trade_no: 商户订单号
|
|
total_fee: 订单金额(分)
|
|
body: 商品描述
|
|
openid: 用户openid
|
|
trade_type: 交易类型 JSAPI/NATIVE/APP
|
|
|
|
Returns:
|
|
返回支付参数
|
|
"""
|
|
params = {
|
|
"appid": self.appid,
|
|
"mch_id": self.mchid,
|
|
"nonce_str": ''.join(random.choices(string.ascii_letters + string.digits, k=32)),
|
|
"body": body,
|
|
"out_trade_no": out_trade_no,
|
|
"total_fee": total_fee,
|
|
"spbill_create_ip": "127.0.0.1",
|
|
"notify_url": self.notify_url,
|
|
"trade_type": trade_type
|
|
}
|
|
|
|
if openid:
|
|
params["openid"] = openid
|
|
|
|
# 小程序 JSAPI 必须使用小程序 AppID 与商户号绑定关系一致
|
|
if trade_type == "JSAPI" and self.appid_mini:
|
|
params["appid"] = self.appid_mini
|
|
|
|
# 生成签名
|
|
params["sign"] = self._generate_sign(params)
|
|
|
|
# 发送请求
|
|
try:
|
|
import ssl
|
|
context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
|
|
context.check_hostname = False
|
|
context.verify_mode = ssl.CERT_NONE
|
|
|
|
req = urllib.request.Request(
|
|
self.pay_url,
|
|
data=self._dict_to_xml(params).encode('utf-8'),
|
|
headers={"Content-Type": "text/xml"}
|
|
)
|
|
|
|
with urllib.request.urlopen(req, context=context, timeout=30) as response:
|
|
result = self._xml_to_dict(response.read().decode())
|
|
|
|
if result.get("return_code") == "SUCCESS" and result.get("result_code") == "SUCCESS":
|
|
if trade_type == "JSAPI":
|
|
prepay = result.get("prepay_id")
|
|
pay_extra = self.build_miniprogram_pay_params(prepay) if prepay else {}
|
|
return {
|
|
"success": True,
|
|
"prepay_id": prepay,
|
|
"package": pay_extra.get("package"),
|
|
"pay_params": pay_extra,
|
|
}
|
|
elif trade_type == "NATIVE":
|
|
return {
|
|
"success": True,
|
|
"code_url": result.get("code_url")
|
|
}
|
|
elif trade_type == "APP":
|
|
return {
|
|
"success": True,
|
|
"prepay_id": result.get("prepay_id")
|
|
}
|
|
else:
|
|
return {
|
|
"success": False,
|
|
"error": result.get("err_code_des", result.get("return_msg", "下单失败"))
|
|
}
|
|
except Exception as e:
|
|
return {"success": False, "error": str(e)}
|
|
|
|
def query_order(self, out_trade_no: str = None, transaction_id: str = None) -> Dict:
|
|
"""查询订单"""
|
|
if not out_trade_no and not transaction_id:
|
|
return {"success": False, "error": "缺少订单号"}
|
|
|
|
params = {
|
|
"appid": self.appid,
|
|
"mch_id": self.mchid,
|
|
"nonce_str": ''.join(random.choices(string.ascii_letters + string.digits, k=32))
|
|
}
|
|
|
|
if out_trade_no:
|
|
params["out_trade_no"] = out_trade_no
|
|
else:
|
|
params["transaction_id"] = transaction_id
|
|
|
|
params["sign"] = self._generate_sign(params)
|
|
|
|
query_url = "https://api.mch.weixin.qq.com/pay/orderquery"
|
|
|
|
try:
|
|
import ssl
|
|
context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
|
|
context.check_hostname = False
|
|
context.verify_mode = ssl.CERT_NONE
|
|
|
|
req = urllib.request.Request(
|
|
query_url,
|
|
data=self._dict_to_xml(params).encode('utf-8'),
|
|
headers={"Content-Type": "text/xml"}
|
|
)
|
|
|
|
with urllib.request.urlopen(req, context=context, timeout=30) as response:
|
|
result = self._xml_to_dict(response.read().decode())
|
|
return {"success": True, "data": result}
|
|
except Exception as e:
|
|
return {"success": False, "error": str(e)}
|
|
|
|
def refund(
|
|
self,
|
|
out_trade_no: str,
|
|
total_fee: int,
|
|
refund_fee: int,
|
|
refund_desc: str = None
|
|
) -> Dict:
|
|
"""申请退款 (V2)"""
|
|
out_refund_no = f"REF{datetime.now().strftime('%Y%m%d%H%M%S')}{random.randint(1000, 9999)}"
|
|
|
|
params = {
|
|
"appid": self.appid,
|
|
"mch_id": self.mchid,
|
|
"nonce_str": ''.join(random.choices(string.ascii_letters + string.digits, k=32)),
|
|
"out_trade_no": out_trade_no,
|
|
"out_refund_no": out_refund_no,
|
|
"total_fee": total_fee,
|
|
"refund_fee": refund_fee,
|
|
"refund_desc": refund_desc or "用户申请退款"
|
|
}
|
|
|
|
params["sign"] = self._generate_sign(params)
|
|
|
|
refund_url = "https://api.mch.weixin.qq.com/secapi/pay/refund"
|
|
|
|
try:
|
|
import ssl
|
|
context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
|
|
context.check_hostname = False
|
|
context.verify_mode = ssl.CERT_NONE
|
|
|
|
# 需要双向证书
|
|
if settings.WXPAY_SSL_CERT and settings.WXPAY_SSL_KEY:
|
|
cert_file = settings.WXPAY_SSL_CERT
|
|
key_file = settings.WXPAY_SSL_KEY
|
|
req = urllib.request.Request(
|
|
refund_url,
|
|
data=self._dict_to_xml(params).encode('utf-8'),
|
|
headers={"Content-Type": "text/xml"}
|
|
)
|
|
with urllib.request.urlopen(req, context=context, timeout=30) as response:
|
|
result = self._xml_to_dict(response.read().decode())
|
|
if result.get("return_code") == "SUCCESS":
|
|
return {
|
|
"success": True,
|
|
"out_refund_no": out_refund_no,
|
|
"refund_id": result.get("refund_id")
|
|
}
|
|
return {"success": False, "error": result.get("return_msg", "退款失败")}
|
|
else:
|
|
return {"success": False, "error": "缺少SSL证书配置"}
|
|
except Exception as e:
|
|
return {"success": False, "error": str(e)}
|
|
|
|
def parse_notify(self, xml_data: str) -> Dict:
|
|
"""解析支付回调"""
|
|
result = self._xml_to_dict(xml_data)
|
|
if result.get("return_code") == "SUCCESS":
|
|
# 验证签名
|
|
sign = result.pop("sign", None)
|
|
if sign and sign == self._generate_sign(result):
|
|
return {"success": True, "data": result}
|
|
return {"success": False, "error": "签名验证失败"}
|
|
|
|
|
|
def get_wechat_pay_v2() -> WechatPayV2Client:
|
|
"""获取微信支付V2客户端"""
|
|
return WechatPayV2Client()
|