94 lines
3.5 KiB
Python
94 lines
3.5 KiB
Python
|
#!/usr/bin/env python3
|
|||
|
|
# -*- coding: utf-8 -*-
|
|||
|
|
"""
|
|||
|
|
模拟微信「商家转账到零钱」结果通知回调，请求本地/远程回调接口，
|
|||
|
|
用于验证：1）接口是否可达 2）wechat_callback_logs 表是否会写入一条记录。
|
|||
|
|
|
|||
|
|
说明：未使用真实签名与加密，服务端会验签失败并返回 500，
|
|||
|
|
但仍会写入 wechat_callback_logs 一条 handler_result=fail 的记录。
|
|||
|
|
运行前请确保 soul-api 已启动；运行后请查表 wechat_callback_logs 是否有新行。
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
import json
|
|||
|
|
import ssl
|
|||
|
|
import sys
|
|||
|
|
from datetime import datetime
|
|||
|
|
from urllib.request import Request, urlopen
|
|||
|
|
from urllib.error import URLError, HTTPError
|
|||
|
|
|
|||
|
|
# 默认请求地址（可改环境或命令行）
|
|||
|
|
DEFAULT_URL = "http://localhost:8080/api/payment/wechat/transfer/notify"
|
|||
|
|
|
|||
|
|
|
|||
|
|
def main():
|
|||
|
|
args = [a for a in sys.argv[1:] if a and not a.startswith("-")]
|
|||
|
|
insecure = "--insecure" in sys.argv or "-k" in sys.argv
|
|||
|
|
url = args[0] if args else DEFAULT_URL
|
|||
|
|
|
|||
|
|
if insecure and url.startswith("https://"):
|
|||
|
|
print("已启用 --insecure，跳过 SSL 证书校验（仅用于本地/测试）")
|
|||
|
|
ctx = ssl.create_default_context()
|
|||
|
|
ctx.check_hostname = False
|
|||
|
|
ctx.verify_mode = ssl.CERT_NONE
|
|||
|
|
else:
|
|||
|
|
ctx = None
|
|||
|
|
|
|||
|
|
# 模拟微信回调的请求体结构（真实场景中 resource.ciphertext 为 AEAD_AES_256_GCM 加密，这里用占位）
|
|||
|
|
body = {
|
|||
|
|
"id": "test-notify-id-" + datetime.now().strftime("%Y%m%d%H%M%S"),
|
|||
|
|
"create_time": datetime.now().strftime("%Y-%m-%dT%H:%M:%S+08:00"),
|
|||
|
|
"resource_type": "encrypt-resource",
|
|||
|
|
"event_type": "MCHTRANSFER.BILL.FINISHED",
|
|||
|
|
"summary": "模拟转账结果通知",
|
|||
|
|
"resource": {
|
|||
|
|
"original_type": "mch_payment",
|
|||
|
|
"algorithm": "AEAD_AES_256_GCM",
|
|||
|
|
"ciphertext": "fake-base64-ciphertext-for-test",
|
|||
|
|
"nonce": "fake-nonce",
|
|||
|
|
"associated_data": "mch_payment",
|
|||
|
|
},
|
|||
|
|
}
|
|||
|
|
body_bytes = json.dumps(body, ensure_ascii=False).encode("utf-8")
|
|||
|
|
|
|||
|
|
headers = {
|
|||
|
|
"Content-Type": "application/json",
|
|||
|
|
"Wechatpay-Timestamp": str(int(datetime.now().timestamp())),
|
|||
|
|
"Wechatpay-Nonce": "test-nonce-" + datetime.now().strftime("%H%M%S"),
|
|||
|
|
"Wechatpay-Signature": "fake-signature-for-test",
|
|||
|
|
"Wechatpay-Serial": "fake-serial-for-test",
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
req = Request(url, data=body_bytes, headers=headers, method="POST")
|
|||
|
|
|
|||
|
|
print(f"POST {url}")
|
|||
|
|
print(f"Body (摘要): event_type={body['event_type']}, resource_type={body['resource_type']}")
|
|||
|
|
print("-" * 50)
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
with urlopen(req, timeout=10, context=ctx) as resp:
|
|||
|
|
print(f"HTTP 状态: {resp.status}")
|
|||
|
|
raw = resp.read().decode("utf-8", errors="replace")
|
|||
|
|
try:
|
|||
|
|
parsed = json.loads(raw)
|
|||
|
|
print("响应 JSON:", json.dumps(parsed, ensure_ascii=False, indent=2))
|
|||
|
|
except Exception:
|
|||
|
|
print("响应 body:", raw[:500])
|
|||
|
|
except HTTPError as e:
|
|||
|
|
print(f"HTTP 状态: {e.code}")
|
|||
|
|
raw = e.read().decode("utf-8", errors="replace")
|
|||
|
|
try:
|
|||
|
|
parsed = json.loads(raw)
|
|||
|
|
print("响应 JSON:", json.dumps(parsed, ensure_ascii=False, indent=2))
|
|||
|
|
except Exception:
|
|||
|
|
print("响应 body:", raw[:500])
|
|||
|
|
except URLError as e:
|
|||
|
|
print(f"请求失败: {e.reason}")
|
|||
|
|
sys.exit(1)
|
|||
|
|
|
|||
|
|
print("-" * 50)
|
|||
|
|
print("请检查数据库表 wechat_callback_logs 是否有新记录（本次为模拟请求，预期会有一条 handler_result=fail 的记录）。")
|
|||
|
|
|
|||
|
|
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
main()
|