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()
|