payment-platform/logs/_lsy-h5pay-probe.py

# -*- coding: utf-8 -*-
"""乐收赢测试环境 **H5 下单** 探针：向上游发一次真实的 /v3/h5pay。

为什么单列一发：被扫（/v3/pay）与主扫（/v3/qrpay）都已真跑过，这是**第三条**
也是最后一条未碰过的出站腿。字段集不同（必须显式 pay_type，不接受 000 AUTO）。

★ 与生产代码的关系（别把本探针当生产量具）：
  报文逐字段抄自 `LsyOrderService.createH5Order`（2026-09-28 读法，第 316-336 行）：
    mch_no / [shop_no] / pay_type / pay_trace_no / pay_time / total_amount
    / [order_body] / [attach] / [notify_url]  + inst_no + sign（client 自动补）
  本探针是**另一支量具**：证明「这个字段集被上游接受」，不是「Java 客户端发的字节被接受」。

★ 不传 notify_url（同上：传了会让上游测试环境回调我们，是有副作用的动作）。
★ 凭据（GC3/GC5）：密钥只从 application.yml 就地读，只打印长度与来源。
"""
import json
import os
import re
import sys
import time
import urllib.request
import urllib.error
from hashlib import md5

YML = "java/src/main/resources/application.yml"
VAR = re.compile(r"^\$\{([A-Za-z0-9_]+):(.*)\}$")
OUT = "logs/_lsy-h5pay-probe.txt"
RESP_BIN = "logs/_lsy-h5pay-resp.bin"

PAY_TYPE = sys.argv[1] if len(sys.argv) > 1 else "300"   # 300 支付宝 / 400 微信

buf = []


def say(s=""):
    buf.append(s)
    print(s.encode("ascii", "replace").decode())


def read_block(path, top):
    lines = open(path, encoding="utf-8", errors="replace").read().split("\n")
    start = None
    for i, ln in enumerate(lines):
        if re.match(r"^%s:\s*$" % re.escape(top), ln):
            start = i
            break
    if start is None:
        raise SystemExit("找不到顶层段 %s:" % top)
    out = {}
    for j in range(start + 1, len(lines)):
        ln = lines[j]
        if ln.strip() == "" or ln.lstrip().startswith("#"):
            continue
        if not ln.startswith((" ", "\t")):
            break
        m = re.match(r"^\s+([A-Za-z0-9_.\-]+):\s*(.*)$", ln)
        if m:
            v = re.split(r"\s+#", m.group(2), maxsplit=1)[0].strip().strip("'\"")
            out[m.group(1)] = v
    return out


def resolve(raw):
    m = VAR.match(raw)
    if not m:
        return raw, "application.yml 字面量"
    var, default = m.group(1), m.group(2)
    if os.environ.get(var):
        return os.environ[var], "环境变量 %s" % var
    return default, "application.yml 默认值（键 %s）" % var


cfg = read_block(YML, "leshouying")
base_url, base_src = resolve(cfg["base-url"])
inst_no, inst_src = resolve(cfg["inst-no"])
key, key_src = resolve(cfg["key"])

say("── 取值来源（不含密钥值）──")
say("  base-url = %s   [%s]" % (base_url, base_src))
say("  inst-no  = %s             [%s]" % (inst_no, inst_src))
say("  key      = <len=%d>        [%s]" % (len(key), key_src))
say()

now = time.strftime("%Y%m%d%H%M%S")
trace = "probe" + now + "h5"
payload = {
    "inst_no": inst_no,
    "mch_no": "Y52211090004",
    "pay_type": PAY_TYPE,            # ★ H5 必须显式指定，不接受 000
    "pay_trace_no": trace,
    "pay_time": now,
    "total_amount": "1",
    "order_body": "联调探针-H5",
    # shop_no / attach / notify_url 生产侧「非 null 才发」，此处留空不传
}

A = "&".join("%s=%s" % (k, payload[k]) for k in sorted(payload))
sign = md5((A + "&key=" + key).encode("utf-8")).hexdigest().lower()
payload["sign"] = sign

say("── 将发送（不含密钥；sign 打码）──")
say("  A = %s" % A)
say()
body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
shown = json.loads(body.decode("utf-8"))
shown["sign"] = "<len=%d>" % len(sign)
say("  body = %s" % json.dumps(shown, ensure_ascii=False, indent=2))
say()
say("  字段清单（= 生产 createH5Order 发的集合）：%s" % ", ".join(sorted(payload)))
say()

url = base_url.rstrip("/") + "/v3/h5pay"
say("── 出站 ──")
say("  POST %s" % url)
say("  Content-Type: application/json;charset=utf-8")
say("  bytes = %d" % len(body))
say()

req = urllib.request.Request(
    url, data=body,
    headers={"Content-Type": "application/json;charset=utf-8"},
    method="POST",
)
try:
    with urllib.request.urlopen(req, timeout=30) as r:
        code, raw = r.status, r.read()
except urllib.error.HTTPError as e:
    code, raw = e.code, e.read()
except Exception as e:
    say("!! 请求未完成: %r" % (e,))
    open(OUT, "w", encoding="utf-8").write("\n".join(buf))
    sys.exit(1)

open(RESP_BIN, "wb").write(raw)
say("── 上游应答 ──")
say("  http_code = %s" % code)
say("  bytes     = %d" % len(raw))
say("  原始字节已落 %s" % RESP_BIN)
say()

try:
    resp = json.loads(raw.decode("utf-8"))
    say("  解码 = utf-8 成功")
except Exception as e:
    say("  utf-8 解码失败: %r" % (e,))
    open(OUT, "w", encoding="utf-8").write("\n".join(buf))
    sys.exit(1)

for k in sorted(resp):
    v = resp[k]
    if v is None:
        say("  %-16s = null" % k)
    else:
        say("  %-16s = %r" % (k, v))
say()

rkey = resp.get("sign")
if rkey:
    nonnull = {k: v for k, v in resp.items() if k != "sign" and v is not None}
    A2 = "&".join("%s=%s" % (k, nonnull[k]) for k in sorted(nonnull))
    mine = md5((A2 + "&key=" + key).encode("utf-8")).hexdigest().lower()
    say("── 入站验签（null 不参与）──")
    say("  参与签名的字段数 = %d / 应答字段数 %d" % (len(nonnull), len(resp) - 1))
    say("  A = %s" % A2)
    say("  重算 == 上游 sign : %s" % (mine == rkey))
else:
    say("── 应答里没有 sign 字段 ──")
say()

say("── 判读线索（脚本不下结论）──")
say("  return_code = %s" % resp.get("return_code"))
say("  return_msg  = %s" % resp.get("return_msg"))
say("  result_code = %s" % resp.get("result_code"))
say()

open(OUT, "w", encoding="utf-8").write("\n".join(buf))