网站导航
> 技术文档 > 别再为支付发愁！Python 搞定微信支付(V3接口)全流程，代码开源即用_python微信提现组建

别再为支付发愁！Python 搞定微信支付(V3接口)全流程，代码开源即用_python微信提现组建

网友: 技术文档 2025-09-03 00:22:57

引言

微信支付涉及多个环节，包括支付、退款、提现等，每个环节都需要进行相应的配置和处理。以下是一个完整的微信支付实现流程，涵盖了从准备工作到各个业务环节的具体实现。最近开发一款小程序，因为官方升级以前写的方法不太好用一怒之下重新写了一下V3版本的，给小伙伴们分享分享。

准备工作

在进行微信支付开发前，需要完成以下准备工作：

  • 注册微信小程序，获取APPID和APPsecret。注册微信小程序
  • 注册微信支付商户号，获取商户号（MCHID）。注册微信支付商户号
  • 商户号配置，登录商户号在产品中心开通jsapi和商家转账两个产品，其他的按自己的需求开通。

 

最重要的一个步骤商家转账功能设置白名单IP要不然用户提现到零钱功能无法实现。

点击商家转账->前往功能->安全能力->接口IP里面!!!!!。

  • 配置API密钥（V3KEY），用于签名生成和验证。看图。
  • 下载API证书，用于敏感接口的调用（如退款、提现）。配置比较复杂要看官方教程。
  • 获取微信支付公钥，用于验证微信支付回调的签名。安全工具有两种平台证书和支付公钥本文使用的是支付公钥看官方介绍，无比要注意，本文使用支付公钥，本文使用支付公钥，本文使用支付公钥。
  • 小程序和商户号必须正确关联。关联教程

API证书两个文件API key和API cert，一个Serial Number，还有公钥也有一个文件和ID：

所有的配置做完了以后我们会有一下配置（一共九个，少一个就会出问题）：

 # 小程序配置 MINIAPP_ID = \"\" # 替换为你的小程序AppID MINIAPP_SECRET = \"\" # 替换为你的小程序AppSecret # 微信支付配置 WECHAT_MCHID = \"\" # 商户号 WECHAT_API_V3_KEY = \"\" # APIv3密钥(32位) WECHAT_CERT_SERIAL_NO = \"\" # 商户证书序列号 WECHAT_PUBLIC_KEY_ID = \"\" # 证书路径配置 WECHAT_PRIVATE_KEY_PATH = str(BASE_DIR / \"certs/key\") # 商户私钥路径 WECHAT_PRIVATE_CERT_PATH = str(BASE_DIR / \"certs/cert\") WECHAT_PUBLIC_KEY_PATH = str(BASE_DIR / \"certs/public\")

到此位置基本准备工作做完了，可以往下走。

代码封装

v3版本有个很不错的python sdk wechatpayv3 用起来方便，支付和退款功能用了改sdk， 小伙伴们可以看看。wechatpayv3。废话不多说了直接上代码。

import jsonimport loggingimport osimport base64import timeimport uuidfrom random import choicesfrom string import ascii_letters, digitsfrom typing import Tuple, Optional, Dict, Anyfrom cryptography.hazmat.primitives.ciphers.aead import AESGCMfrom wechatpayv3 import WeChatPay, WeChatPayTypefrom config.config import settingslogger = logging.getLogger(__name__)class WechatPayService: \"\"\" 微信支付服务封装类（支持小程序支付、回调处理、退款、提现等） 文档参考： - 支付: https://pay.weixin.qq.com/wiki/doc/apiv3/apis/chapter3_5_1.shtml - 退款: https://pay.weixin.qq.com/wiki/doc/apiv3/apis/chapter3_5_9.shtml - 提现: https://pay.weixin.qq.com/wiki/doc/apiv3/apis/chapter4_3_1.shtml \"\"\" def __init__( self, mchid: str = settings.WECHAT_MCHID, appid: str = settings.MINIAPP_ID, private_key: Optional[str] = None, cert_serial_no: str = settings.WECHAT_CERT_SERIAL_NO, apiv3_key: str = settings.WECHAT_API_V3_KEY, notify_url: str = settings.WECHAT_NOTIFY_URL, cert_dir: str = settings.WECHAT_PRIVATE_CERT_PATH, public_key: Optional[str] = None, public_key_id: str = settings.WECHAT_PUBLIC_KEY_ID, timeout: Tuple[int, int] = (10, 30), ): \"\"\" 初始化微信支付服务 Args: mchid: 微信支付商户号 appid: 小程序appid private_key: 商户证书私钥内容（字符串），可选，默认从配置文件路径读取 cert_serial_no: 商户证书序列号 apiv3_key: APIv3密钥 notify_url: 支付结果回调地址 cert_dir: 平台证书缓存目录 public_key: 平台公钥内容（字符串），可选，默认从配置文件路径读取 public_key_id: 平台证书序列号 timeout: 请求超时时间(连接超时, 读取超时) \"\"\" self.mchid = mchid self.appid = appid self.notify_url = notify_url self.apiv3_key = apiv3_key # 处理密钥文件读取 try: private_key = private_key or self._read_key_file( f\"{settings.WECHAT_PRIVATE_KEY_PATH}/apiclient_key.pem\" ) public_key = public_key or self._read_key_file( f\"{settings.WECHAT_PUBLIC_KEY_PATH}/pub_key.pem\" ) except Exception as e: logger.error(f\"读取密钥文件失败: {str(e)}\") raise # 初始化微信支付SDK self.wxpay = WeChatPay( wechatpay_type=WeChatPayType.MINIPROG, mchid=mchid, private_key=private_key, cert_serial_no=cert_serial_no, apiv3_key=apiv3_key, appid=appid, notify_url=notify_url, cert_dir=cert_dir, public_key=public_key, public_key_id=public_key_id, timeout=timeout, ) def _read_key_file(self, file_path: str) -> str: \"\"\"安全读取密钥文件\"\"\" if not os.path.exists(file_path): raise FileNotFoundError(f\"密钥文件不存在: {file_path}\") with open(file_path, \"r\") as f: return f.read() def create_order( self, openid: str, amount: int, description: str, out_trade_no: Optional[str] = None, **kwargs, ) -> Tuple[bool, Dict]: \"\"\" 创建小程序支付订单 Args: openid: 用户openid amount: 支付金额（单位：分） description: 商品描述 out_trade_no: 商户订单号（可选，不传则自动生成） **kwargs: 其他可选参数（如attach附加数据等） Returns: Tuple[bool, Dict]: (success, data) success为True时，data包含调起支付所需参数； success为False时，data为错误信息 \"\"\" try: # 生成商户订单号（如果未提供） out_trade_no = out_trade_no or self._generate_out_trade_no() logger.info(f\"开始创建订单，订单号: {out_trade_no}\") code, message = self.wxpay.pay( description=description, out_trade_no=out_trade_no, amount={\"total\": amount}, pay_type=WeChatPayType.MINIPROG, payer={\"openid\": openid}, **kwargs, ) # 检查下单是否成功（HTTP状态码2XX表示成功） if 200 <= code  Dict[str, Any]: \"\"\" 解析下单API返回结果并生成小程序调起支付所需参数 Args: api_response: 微信支付API返回的原始消息 Returns: Dict[str, Any]: 小程序调起支付所需的参数字典 Raises: ValueError: 当解析失败时抛出 \"\"\" try: data = json.loads(api_response) prepay_id = data.get(\"prepay_id\") if not prepay_id: raise ValueError(\"响应中缺少prepay_id\") # 生成调起支付所需参数 timestamp = str(int(time.time())) nonce_str = uuid.uuid4().hex package = f\"prepay_id={prepay_id}\" # 生成签名 sign = self.wxpay.sign([self.appid, timestamp, nonce_str, package]) return { \"appId\": self.appid, \"timeStamp\": timestamp, \"nonceStr\": nonce_str, \"package\": package, \"signType\": \"RSA\", \"paySign\": sign, \"prepay_id\": prepay_id, \"out_trade_no\": data.get(\"out_trade_no\", \"\"), } except Exception as e: logger.exception(\"解析微信支付返回结果失败\") raise ValueError(f\"解析微信支付返回结果失败: {str(e)}\") def _generate_out_trade_no(self) -> str: \"\"\"生成商户订单号\"\"\" date_str = time.strftime(\"%Y%m%d%H%M%S\") rand_str = \"\".join(choices(ascii_letters + digits, k=6)) return f\"{self.mchid}{date_str}{rand_str}\" def verify_notification( self, headers: Dict, body: str ) -> Tuple[bool, Optional[Dict]]: \"\"\" 验证支付结果通知签名 Args: headers: 请求头（必须包含Wechatpay-开头的签名头） body: 请求体（原始字符串） Returns: Tuple[bool, Optional[Dict]]: (is_valid, resource) is_valid为True表示验证通过，resource为解密后的通知数据 \"\"\" try: result = self.wxpay.callback(headers, body) if result and result.get(\"resource\"): return True, result.get(\"resource\") return False, None except Exception as e: logger.exception(\"验证支付通知异常\") return False, None async def handle_callback( self, request_headers: Dict, request_body: bytes ) -> Tuple[bool, Optional[Dict]]: \"\"\" 统一处理微信支付各类回调（支付、退款、提现等） Args: request_headers: 请求头字典（必须包含Wechatpay-开头的签名头） request_body: 请求体原始字节数据 Returns: Tuple[bool, Optional[Dict]]: (is_valid, decrypted_data) is_valid表示回调是否验证通过，decrypted_data为解密后的数据 \"\"\" try: # 首先验证签名 is_valid, resource = self.verify_notification( request_headers, request_body.decode(\"utf-8\") ) if not is_valid or not resource: return False, None # 解密数据 decrypted_data = self._decrypt_callback_resource(resource) if not decrypted_data: return False, None # 解析JSON数据 callback_result = json.loads(decrypted_data) event_type = callback_result.get(\"event_type\", \"UNKNOWN\") logger.info(f\"收到微信支付回调: {event_type}\") return True, callback_result except Exception as e: logger.exception(\"处理回调数据时出错\") return False, None def _decrypt_callback_resource(self, resource: Dict) -> Optional[str]: \"\"\" 解密回调中的resource数据 Args: resource: 回调中的resource字典 Returns: Optional[str]: 解密后的JSON字符串，失败返回None \"\"\" try: nonce = resource[\"nonce\"] ciphertext = resource[\"ciphertext\"] associated_data = resource[\"associated_data\"] key_bytes = self.apiv3_key.encode(\"utf-8\") nonce_bytes = nonce.encode(\"utf-8\") ad_bytes = associated_data.encode(\"utf-8\") data = base64.b64decode(ciphertext) aesgcm = AESGCM(key_bytes) decrypted = aesgcm.decrypt(nonce_bytes, data, ad_bytes) return decrypted.decode(\"utf-8\") except Exception as e: logger.exception(\"解密回调数据失败\") return None def create_refund( self, out_trade_no: str, out_refund_no: str, amount: int, refund_amount: int, reason: Optional[str] = None, **kwargs, ) -> Tuple[bool, Dict]: \"\"\" 创建退款订单 Args: out_trade_no: 原支付订单号 out_refund_no: 退款单号 amount: 原订单金额（分） refund_amount: 退款金额（分） reason: 退款原因（可选） **kwargs: 其他可选参数 Returns: Tuple[bool, Dict]: (success, data) success为True时，data包含退款结果； success为False时，data为错误信息 \"\"\" try: logger.info(f\"开始创建退款，原订单号: {out_trade_no}, 退款单号: {out_refund_no}\") code, message = self.wxpay.refund( out_trade_no=out_trade_no, out_refund_no=out_refund_no, amount={\"total\": amount, \"refund\": refund_amount, \"currency\": \"CNY\"}, reason=reason, **kwargs, ) if 200 <= code  Tuple[bool, Dict]: \"\"\" 查询退款状态 Args: out_refund_no: 退款单号 Returns: Tuple[bool, Dict]: (success, data) success为True时，data包含退款状态； success为False时，data为错误信息 \"\"\" try: logger.info(f\"查询退款状态，退款单号: {out_refund_no}\") code, message = self.wxpay.query_refund(out_refund_no=out_refund_no) if 200 <= code < 300: result = json.loads(message) logger.info(f\"退款查询成功，状态: {result.get(\'status\')}\") return True, result error_msg = f\"退款查询失败[{code}]: {message}\" logger.error(error_msg) return False, {\"code\": code, \"message\": error_msg} except Exception as e: error_msg = f\"查询退款异常: {str(e)}\" logger.exception(error_msg) return False, {\"message\": error_msg}# 全局实例（单例模式）wechat_pay_service = WechatPayService( mchid=settings.WECHAT_MCHID, appid=settings.MINIAPP_ID, private_key=open(f\"{settings.WECHAT_PRIVATE_KEY_PATH}/apiclient_key.pem\").read(), cert_serial_no=settings.WECHAT_CERT_SERIAL_NO, apiv3_key=settings.WECHAT_API_V3_KEY, notify_url=settings.WECHAT_NOTIFY_URL, cert_dir=settings.WECHAT_PRIVATE_CERT_PATH, public_key=open(f\"{settings.WECHAT_PUBLIC_KEY_PATH}/pub_key.pem\").read(), public_key_id=settings.WECHAT_PUBLIC_KEY_ID,)

支付

支付环节主要包括生成预支付订单和调起支付：

  • 调用微信支付统一下单接口生成预支付订单，返回预支付交易会话标识（prepay_id）。
  • 前端根据返回的参数调起微信支付，用户完成支付后，微信会异步通知支付结果。

后端

from fastapi import HTTPException, status, APIRouter, dependencies, Securityfrom fastapi.encoders import jsonable_encoderfrom fastapi.responses import JSONResponsefrom .schema import OrderFormfrom application.models import Order, Userfrom core.WechatPayService import wechat_pay_servicefrom random import samplefrom string import digitsimport timefrom config.config import settingsdef create_order_number(): date_str = time.strftime(\"%Y%m%d%H%M%S\") rand_str = \'\'.join(sample(digits, 6)) return f\"{settings.WECHAT_MCHID}{date_str}{rand_str}\"order_router = APIRouter(prefix=\"/order\")@order_router.post(\"\", summary=\"创建订单\")async def create_order(order_form: OrderForm) -> JSONResponse: \"\"\" 创建订单 \"\"\" try: data = order_form.model_dump(exclude={\"amount\"}) data[\'amount\'] = order_form.amount data[\"out_trade_no\"] = create_order_number() order = await Order.create(**data) user = await User.filter(id=order_form.user_id).first() success, result = wechat_pay_service.create_order( openid=user.openid, amount=order.amount, out_trade_no=order.out_trade_no, description=order.description ) if not success: raise HTTPException( status_code=status.HTTP_406_NOT_ACCEPTABLE, detail=\"创建订单失败,原因是\"+result ) order.transaction_id = result.get(\"transaction_id\") await order.save() return JSONResponse({\"code\": 200, \"message\": \"创建订单成功\", \"data\":result}) except Exception as e: print(e) raise HTTPException( status_code=status.HTTP_406_NOT_ACCEPTABLE, detail=\"创建订单失败,原因是\"+str(e) )

前端：

 async onSubmit() { const { amount, description } = this.data if (!amount || !description) { wx.showToast({ title: \'请填写完整信息\', icon: \'none\' }) return } this.setData({ loading: true }) try { const user_id = wx.getStorageSync(\'user_id\') const res = await createOrder({ user_id, amount: parseInt(amount), // 转为分 description }) wx.requestPayment(res.data).then(resp=>{ console.log(resp) }) } catch (error) { wx.showToast({ title: error.errMsg || \'支付失败\', icon: \'none\' }) } finally { this.setData({ loading: false }) } },

    支付回调处理

    支付回调是微信支付完成后通知商户服务器的过程：

    • 验证回调的签名，确保请求来自微信支付。
    • 处理业务逻辑，如更新订单状态、发货等。
    • 返回处理结果给微信支付，避免重复通知。

    解码后的数据可以根据trade_state来判断并进行业务相关的处理。

    支付回调处理直接用wechat_pay_service.handle_callback,拿到的数据是

    { \"mchid\": \"\", \"appid\": \"\", \"out_trade_no\": \"172404557320250808040910658204\", \"transaction_id\": \"4200002822202508081886622870\", \"trade_type\": \"JSAPI\", \"trade_state\": \"SUCCESS\", \"trade_state_desc\": \"支付成功\", \"bank_type\": \"OTHERS\", \"attach\": \"\", \"success_time\": \"2025-08-08T04:09:21+08:00\", \"payer\": { \"openid\": \"\" }, \"amount\": { \"total\": 1, \"payer_total\": 1, \"currency\": \"CNY\", \"payer_currency\": \"CNY\" }}

      退款

      退款环节需要调用微信支付的退款接口：

      退款也是差不多的。后端代码

      • 使用API证书进行签名和加密，确保安全性。
      • 提交退款申请，包括订单号、退款金额等信息。
      • 微信支付处理退款后，会异步通知退款结果。
      @refund_router.post(\"/{order_id}\")async def create_refund(order_id: int): try: refund_number = create_refund_number() print(refund_number) order = await Order.get(pk=order_id) res = wechat_pay_service.wxpay.refund( transaction_id=order.transaction_id, out_refund_no=refund_number, out_trade_no=order.out_trade_no, funds_account=\"AVAILABLE\", amount={\'refund\':order.amount, \'total\':order.amount, \'currency\':\'CNY\'}, notify_url=settings.WECHAT_REFUND_URL, reason=\"用户申请退款\" ) print(\"res\", res) refund = await Refund.create( order_id=order_id, out_refund_no=refund_number, amount=order.amount, reason=\"用户申请退款\", status=\"created\" ) order.status = \"closed\" await order.save() return JSONResponse({\"code\": 200, \"message\": \"创建订单成功\", \"data\":jsonable_encoder(refund)}) except Exception as e: print(e) raise HTTPException( status_code=status.HTTP_406_NOT_ACCEPTABLE, detail=\"订单退款失败,原因是\"+str(e) )

        退款回调处理

        退款回调是微信支付处理完退款后通知商户服务器的过程：

        • 验证回调的签名，确保请求来自微信支付。
        • 处理业务逻辑，如更新退款状态、记录退款信息等。
        • 返回处理结果给微信支付，避免重复通知。

        直接调用handle_callback函数 返回的数据格式如下：

        { \"mchid\": \"\", \"out_trade_no\": \"\", \"transaction_id\": \"\", \"out_refund_no\": \"\", \"refund_id\": \"50303004312025080887471769928\", \"refund_status\": \"SUCCESS\", \"success_time\": \"2025-08-08T05:42:44+08:00\", \"amount\": { \"total\": 1, \"refund\": 1, \"payer_total\": 1, \"payer_refund\": 1 }, \"user_received_account\": \"支付用户零钱\"}

        通过refund_status来判断并进行业务相关的操作了。

        用户提现到零钱

        用户提现到零钱功能需要调用微信支付的提现接口：

        • 使用API证书进行签名和加密，确保安全性。
        • 提交提现申请，包括用户OpenID、提现金额等信息。
        • 微信支付处理提现后，会异步通知提现结果。

        这是比较麻烦的， 用wechatpayv3 也一直报错，只能自己写了。大伙也可以看看官方文档。

        下面按照官方的给的java sdk 写了python三个python类。试了各种大模型，写的贼好看就是不好用，各种奇葩的错误。最后还是自己写了一遍。然后给大模型，这个时候大模型厉害了，给我整理了一下就没有报错了。直接上代码：

        import randomimport stringimport timefrom datetime import datetime, timedeltaimport base64import jsonfrom typing import Dict, Optional, Anyfrom urllib.parse import quotefrom Crypto.PublicKey import RSAfrom Crypto.Signature import pkcs1_15from Crypto.Hash import SHA256from Crypto.Cipher import PKCS1_OAEPclass WXPayUtility: @staticmethod def to_json(obj: Any) -> str: \"\"\"将对象转换为 JSON 字符串\"\"\" return json.dumps(obj, ensure_ascii=False) @staticmethod def from_json(json_str: str) -> Dict: \"\"\"将 JSON 字符串解析为字典\"\"\" return json.loads(json_str) @staticmethod def read_key_string_from_path(key_path: str) -> str: \"\"\"从公私钥文件路径中读取文件内容\"\"\" with open(key_path, \'r\', encoding=\'utf-8\') as f: return f.read() @staticmethod def load_private_key_from_string(key_string: str) -> RSA.RsaKey: \"\"\"读取 PKCS#8 格式的私钥字符串并加载为私钥对象\"\"\" key_string = key_string.replace(\"-----BEGIN PRIVATE KEY-----\", \"\") \\ .replace(\"-----END PRIVATE KEY-----\", \"\") \\ .replace(\"\\n\", \"\") key_bytes = base64.b64decode(key_string) return RSA.import_key(key_bytes) @staticmethod def load_private_key_from_path(key_path: str) -> RSA.RsaKey: \"\"\"从 PKCS#8 格式的私钥文件中加载私钥\"\"\" return WXPayUtility.load_private_key_from_string( WXPayUtility.read_key_string_from_path(key_path) ) @staticmethod def load_public_key_from_string(key_string: str) -> RSA.RsaKey: \"\"\"读取 PKCS#8 格式的公钥字符串并加载为公钥对象\"\"\" key_string = key_string.replace(\"-----BEGIN PUBLIC KEY-----\", \"\") \\ .replace(\"-----END PUBLIC KEY-----\", \"\") \\ .replace(\"\\n\", \"\") key_bytes = base64.b64decode(key_string) return RSA.import_key(key_bytes) @staticmethod def load_public_key_from_path(key_path: str) -> RSA.RsaKey: \"\"\"从 PKCS#8 格式的公钥文件中加载公钥\"\"\" return WXPayUtility.load_public_key_from_string( WXPayUtility.read_key_string_from_path(key_path) ) @staticmethod def create_nonce(length: int = 32) -> str: \"\"\"创建指定长度的随机字符串，字符集为[0-9a-zA-Z]\"\"\" chars = string.ascii_letters + string.digits return \'\'.join(random.choice(chars) for _ in range(length)) @staticmethod def encrypt(public_key: RSA.RsaKey, plaintext: str) -> str: \"\"\"使用公钥按照 RSA_PKCS1_OAEP_PADDING 算法进行加密\"\"\" cipher = PKCS1_OAEP.new(public_key) encrypted = cipher.encrypt(plaintext.encode(\'utf-8\')) return base64.b64encode(encrypted).decode(\'utf-8\') @staticmethod def verify(message: str, signature: str, public_key: RSA.RsaKey) -> bool: \"\"\"使用公钥按照 SHA256withRSA 算法验证签名\"\"\" h = SHA256.new(message.encode(\'utf-8\')) try: pkcs1_15.new(public_key).verify(h, base64.b64decode(signature)) return True except (ValueError, TypeError): return False @staticmethod def build_authorization(mchid, certificate_serial_no, private_key, method, uri, body): nonce = WXPayUtility.create_nonce(32) timestamp = int(time.time()) # 构造签名串 message = f\"{method}\\n{uri}\\n{timestamp}\\n{nonce}\\n{body if body else \'\'}\\n\" print(\"\\n=== 签名调试信息 ===\") print(f\"Method: {method}\") print(f\"URI: {uri}\") print(f\"Timestamp: {timestamp}\") print(f\"Nonce: {nonce}\") print(f\"Body: {body}\") print(\"\\n签名串(带换行符):\") print(repr(message)) print(\"\\n签名串(实际格式):\") print(message) # 生成签名 signature = WXPayUtility.sign(message, private_key) print(\"\\n生成的签名:\", signature) auth = ( f\'WECHATPAY2-SHA256-RSA2048 mchid=\"{mchid}\",nonce_str=\"{nonce}\",\' f\'signature=\"{signature}\",timestamp=\"{timestamp}\",serial_no=\"{certificate_serial_no}\"\' ) print(\"\\nAuthorization头:\", auth) return auth @staticmethod def sign(message, private_key): \"\"\"添加签名调试信息\"\"\" print(\"\\n=== 签名过程 ===\") print(\"签名输入:\", message) try: h = SHA256.new(message.encode(\'utf-8\')) print(\"SHA256哈希:\", h.hexdigest()) signature = pkcs1_15.new(private_key).sign(h) b64_signature = base64.b64encode(signature).decode(\'utf-8\') print(\"Base64签名:\", b64_signature) return b64_signature except Exception as e: print(\"签名过程中出错:\", str(e)) raise @staticmethod def url_encode(content: str) -> str: \"\"\"对参数进行 URL 编码\"\"\" return quote(content, safe=\'\') @staticmethod def url_encode_params(params: Dict[str, Any]) -> str: \"\"\"对参数Map进行 URL 编码，生成 QueryString\"\"\" if not params: return \"\" return \"&\".join( f\"{k}={WXPayUtility.url_encode(str(v))}\" for k, v in params.items() ) @staticmethod def validate_response( wechatpay_public_key_id: str, wechatpay_public_key: RSA.RsaKey, headers: Dict[str, str], body: str ) -> None: \"\"\"验证微信支付APIv3应答签名\"\"\" timestamp = headers.get(\"Wechatpay-Timestamp\") try: response_time = datetime.fromtimestamp(int(timestamp)) # 拒绝过期请求（5分钟） if abs(datetime.now() - response_time) > timedelta(minutes=5): raise ValueError(  f\"Validate http response,timestamp[{timestamp}] of httpResponse is expires, \"  f\"request-id[{headers.get(\'Request-ID\')}]\" ) except (ValueError, TypeError) as e: raise ValueError( f\"Validate http response,timestamp[{timestamp}] of httpResponse is invalid, \" f\"request-id[{headers.get(\'Request-ID\')}]\" ) from e message = f\"{timestamp}\\n{headers.get(\'Wechatpay-Nonce\')}\\n{body if body else \'\'}\\n\" serial_number = headers.get(\"Wechatpay-Serial\") if serial_number != wechatpay_public_key_id: raise ValueError( f\"Invalid Wechatpay-Serial, Local: {wechatpay_public_key_id}, Remote: {serial_number}\" ) signature = headers.get(\"Wechatpay-Signature\") if not signature: raise ValueError(\"Missing Wechatpay-Signature in headers\") if not WXPayUtility.verify(message, signature, wechatpay_public_key): raise ValueError( f\"Validate response failed,the WechatPay signature is incorrect.\\n\" f\"Request-ID[{headers.get(\'Request-ID\')}]\\tresponseHeader[{headers}]\\t\" f\"responseBody[{body[:1024] if body else \'\'}]\" )class ApiException(Exception): \"\"\"微信支付API错误异常\"\"\" def __init__(self, status_code: int, body: str, headers: Dict[str, str]): self.status_code = status_code self.body = body self.headers = headers self.error_code = None self.error_message = None if body: try: data = json.loads(body) self.error_code = data.get(\"code\") self.error_message = data.get(\"message\") except json.JSONDecodeError: pass super().__init__( f\"微信支付API访问失败，StatusCode: [{status_code}], Body: [{body}], Headers: [{headers}]\" ) def get_status_code(self) -> int: \"\"\"获取 HTTP 应答状态码\"\"\" return self.status_code def get_body(self) -> str: \"\"\"获取 HTTP 应答包体内容\"\"\" return self.body def get_headers(self) -> Dict[str, str]: \"\"\"获取 HTTP 应答 Header\"\"\" return self.headers def get_error_code(self) -> Optional[str]: \"\"\"获取 错误码 （错误应答中的 code 字段）\"\"\" return self.error_code def get_error_message(self) -> Optional[str]: \"\"\"获取 错误消息 （错误应答中的 message 字段）\"\"\" return self.error_message
        import jsonimport loggingfrom typing import List, Optional, Dictfrom dataclasses import dataclass, fieldimport requestsfrom pathlib import Pathfrom core.WXPayUtility import WXPayUtility, ApiExceptionimport timefrom config.config import settingslogger = logging.getLogger(__name__)@dataclassclass TransferSceneReportInfo: \"\"\"转账场景上报信息\"\"\" info_type: str info_content: str def to_dict(self) -> Dict: return {\"info_type\": self.info_type, \"info_content\": self.info_content}@dataclassclass TransferBillsRequest: \"\"\"转账请求参数\"\"\" appid: str out_bill_no: str transfer_scene_id: str openid: str user_name: Optional[str] = None transfer_amount: int = 0 transfer_remark: str = \"\" notify_url: str = \"\" user_recv_perception: str = \"\" transfer_scene_report_infos: List[TransferSceneReportInfo] = field(default_factory=list) def to_json(self) -> str: data = { \"appid\": self.appid, \"out_bill_no\": self.out_bill_no, \"transfer_scene_id\": self.transfer_scene_id, \"openid\": self.openid, \"transfer_amount\": self.transfer_amount, \"transfer_remark\": self.transfer_remark, \"notify_url\": self.notify_url, \"user_recv_perception\": self.user_recv_perception, \"transfer_scene_report_infos\": [info.to_dict() for info in self.transfer_scene_report_infos] } if self.user_name is not None: data[\"user_name\"] = self.user_name return json.dumps(data, separators=(\',\', \':\'))@dataclassclass TransferBillResponse: \"\"\"转账响应结果\"\"\" out_bill_no: str transfer_bill_no: str create_time: str state: str fail_reason: Optional[str] = None package_info: Optional[str] = None mch_id: Optional[str] = None app_id: Optional[str] = None @classmethod def from_json(cls, json_str: str): data = json.loads(json_str) return cls( out_bill_no=data.get(\"out_bill_no\"), transfer_bill_no=data.get(\"transfer_bill_no\"), create_time=data.get(\"create_time\"), state=data.get(\"state\"), fail_reason=data.get(\"fail_reason\"), package_info=data.get(\"package_info\") )class WeChatPayTransfer: \"\"\"微信支付转账服务\"\"\" API_HOST = \"https://api.mch.weixin.qq.com\" TRANSFER_PATH = \"/v3/fund-app/mch-transfer/transfer-bills\" def __init__(self, mchid: str, private_key_path: str, cert_serial_no: str,  wechatpay_serial: str, wechatpay_public_key_path: str): \"\"\" 初始化转账服务 :param mchid: 商户号 :param private_key_path: 商户私钥路径 :param cert_serial_no: 商户证书序列号 :param wechatpay_serial: 微信支付平台证书序列号 :param wechatpay_public_key_path: 微信支付平台公钥路径 \"\"\" self.mchid = mchid self.cert_serial_no = cert_serial_no self.wechatpay_serial = wechatpay_serial # 加载密钥 self.private_key = WXPayUtility.load_private_key_from_path(private_key_path) self.wechatpay_public_key = WXPayUtility.load_public_key_from_path(wechatpay_public_key_path) self.http_client = requests.Session() logger.info(\"微信支付转账服务初始化完成\") def transfer(self, request: TransferBillsRequest) -> TransferBillResponse: \"\"\" 发起商家转账 :param request: 转账请求参数 :return: 转账响应结果 :raises: ApiException 当API返回错误时 \"\"\" url = self.API_HOST + self.TRANSFER_PATH body = request.to_json() # 构建签名头 auth = self._build_authorization(body) headers = { \"Accept\": \"application/json\", \"Content-Type\": \"application/json\", \"Authorization\": auth, \"Wechatpay-Serial\": self.wechatpay_serial } try: logger.info(f\"发起微信转账请求: {request.out_bill_no}\") response = self.http_client.post(url, headers=headers, data=body, timeout=30) response.raise_for_status() # 验证响应签名 self._validate_response(response) return TransferBillResponse.from_json(response.text) except requests.exceptions.RequestException as e: logger.error(f\"微信转账请求失败: {str(e)}\") if e.response is not None: raise ApiException(e.response.status_code, e.response.text, dict(e.response.headers)) raise RuntimeError(f\"请求失败: {str(e)}\") def _build_authorization(self, body: str) -> str: \"\"\"构建Authorization头\"\"\" method = \"POST\" uri = self.TRANSFER_PATH nonce = WXPayUtility.create_nonce(32) timestamp = str(int(time.time())) # 构造签名串 message = f\"{method}\\n{uri}\\n{timestamp}\\n{nonce}\\n{body}\\n\" # 使用商户私钥签名 signature = WXPayUtility.sign(message, self.private_key) # 构建Authorization头 return ( f\'WECHATPAY2-SHA256-RSA2048 \' f\'mchid=\"{self.mchid}\",\' f\'nonce_str=\"{nonce}\",\' f\'signature=\"{signature}\",\' f\'timestamp=\"{timestamp}\",\' f\'serial_no=\"{self.cert_serial_no}\"\' ) def _validate_response(self, response: requests.Response): \"\"\"验证响应签名\"\"\" WXPayUtility.validate_response( wechatpay_public_key_id=self.wechatpay_serial, wechatpay_public_key=self.wechatpay_public_key, headers=dict(response.headers), body=response.text )wxpay_transfer_service = WeChatPayTransfer( mchid=settings.WECHAT_MCHID, private_key_path=str(Path(settings.WECHAT_PRIVATE_KEY_PATH) / \"apiclient_key.pem\"), cert_serial_no=settings.WECHAT_CERT_SERIAL_NO, wechatpay_serial=settings.WECHAT_PUBLIC_KEY_ID, wechatpay_public_key_path=str(Path(settings.WECHAT_PUBLIC_KEY_PATH) / \"pub_key.pem\"))

        用法直接上代码：

        @withdraw_router.post(\"\", summary=\"提现申请\")async def create_withdraw(withdraw_form: WithdrawForm) -> JSONResponse: try: user = await User.filter(id=withdraw_form.user_id).first() data = withdraw_form.model_dump() data[\'out_batch_no\'] = create_withdraw_number() withdraw = await Withdraw.create(**data) request_data = TransferBillsRequest( appid=settings.MINIAPP_ID, out_bill_no=data[\'out_batch_no\'], # 生成唯一订单号 transfer_scene_id=\"1000\", # 转账场景ID openid=user.openid, transfer_amount=withdraw_form.amount, transfer_remark=withdraw_form.description, notify_url=settings.WECHAT_WITHDRAW_URL, transfer_scene_report_infos=[ TransferSceneReportInfo(\"活动名称\", \"用户奖励\"), TransferSceneReportInfo(\"奖励说明\", \"老用户奖励\") ] ) result = wxpay_transfer_service.transfer(request_data) result_dict = jsonable_encoder( result) result_dict[\'appid\'] = settings.MINIAPP_ID result_dict[\'mid\'] = settings.WECHAT_MCHID return JSONResponse({\"code\": 200, \"msg\": \"success\", \"data\": result_dict}) except Exception as e: print(\"提现出错！：\", e) raise HTTPException(status_code=status.HTTP_406_NOT_ACCEPTABLE, detail=\"提现失败,请稍后再试\")

        商户转账到零钱现在的版本贼严格，                TransferSceneReportInfo(\"活动名称\", \"用户奖励\"),
                        TransferSceneReportInfo(\"奖励说明\", \"老用户奖励\") 这里要写的明明白白，而且官方提供了固定的格式和名称，大家要注意。看看官方文档转账场景报备信息参数

        注意：现在用户提现时必须用户自己主动发起， 调用该接口时        result = wxpay_transfer_service.transfer(request_data)，result数据结构如下：

        { \'out_bill_no\': \'172404557320250808061850350468\', \'transfer_bill_no\': \'\', \'create_time\': \'2025-08-08T06: 18: 51+08: 00\', \'state\': \'WAIT_USER_CONFIRM\', \'fail_reason\': None, \'package_info\': \'\', \'mch_id\': None, \'app_id\': None}

        这里的    \'package_info\': \'\', 
            \'mch_id\': None, 
            \'app_id\': None这三个数据必须传给前端wx.requestMerchantTransfer函数，自己去看看。前端代码如下：

         // 提交提现申请 async onSubmit() { const { amount, description } = this.data if (!amount || !description) { wx.showToast({ title: \'请填写完整信息\', icon: \'none\' }) return } const amountInCent = Math.round(parseFloat(amount) * 100) if (amountInCent { if (wx.canIUse(\'requestMerchantTransfer\')) { wx.requestMerchantTransfer({ mchId: res.data.mch_id, appId: res.data.app_id, package: res.data.package_info, success: (res) => {  // res.err_msg将在页面展示成功后返回应用时返回ok，并不代表付款成功  console.log(\'success:\', res); }, fail: (res) => {  console.log(\'fail:\', res); }, }); } else { wx.showModal({ content: \'你的微信版本过低，请更新至最新版本。\', showCancel: false, }); } }) wx.showToast({ title: \'提现申请已提交\' }) this.setData({ amount: \'\', description: \'\' }) this.loadWithdraws() } catch (error) { wx.showToast({ title: error.message || \'提现失败\', icon: \'none\' }) } finally { this.setData({ loading: false }) } },

          用户提现回调处理

          提现回调是微信支付处理完提现后通知商户服务器的过程：

          • 验证回调的签名，确保请求来自微信支付。
          • 处理业务逻辑，如更新提现状态、记录提现信息等。
          • 返回处理结果给微信支付，避免重复通知。

          回调返回的数数据通过handle_callback解码以后的数据如下：

          { \"mch_id\": \"\", \"out_bill_no\": \"\", \"transfer_bill_no\": \"1330007968556092508080033039221820\", \"transfer_amount\": 20, \"state\": \"SUCCESS\", \"openid\": \"\", \"create_time\": \"2025-08-08T06:03:57+08:00\", \"update_time\": \"2025-08-08T06:04:01+08:00\", \"mchid\": \"\"}

          通过state字段来判断并进行相关的业务操作。

          到目前位置小程序基本支付，退款，用户提现功能已经实现了，欢迎大家讨论，不妥之处欢迎大佬指点指点，如果对你有帮助可以点赞收藏，谢谢。

          网络标签：

          相关问题