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

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

为什么要打这一发（前面已经验过签名了）：
  - 之前那条只读 /v3/queryorder 证明了「签名规则 + 信封」对；
  - 但**下单**才走完整的业务校验（mch_no 是否有效、shop_no 要不要、pay_trace_no 唯一性、
    total_amount 形态、time_expire 取值）——这些都是「签名对」证明不了的。
  - 2026-09-28 用户已与上游确认：文档里的示例参数可用于测试。

★ 与生产代码的关系（重要，别把本探针当生产量具）：
  报文**逐字段抄自** `LsyOrderService.createQrOrder`（主扫那条腿）在 2026-09-28 的读法：
    mch_no / [shop_no] / pay_trace_no / pay_time / total_amount / time_expire
    / [order_body] / [attach] / [notify_url]    + inst_no + sign（由 client 自动补）
  本脚本会把**实际发出去的 body** 原样打出来，好让人把它与上面那份字段清单逐格对
  —— 转写件必须能被核对，否则它只是一段漂亮的猜测。
  但本探针仍是**另一支量具**：它证明的是「这个字段集被上游接受」，不是「Java 客户端发出的字节被接受」。

★ 不传 notify_url：上游文档说「不传取机构默认配置回调」。传了就等于让上游的测试环境
  去回调我们的地址，那是一个**有副作用**的动作，本次不需要。

★ 凭据（GC3/GC5）：机构密钥只从 application.yml 就地读，只打印长度与来源，绝不落文件、不上命令行。
★ 出站 body 里的 sign 打码（它是这一笔的鉴权件）——字段形状仍然完整可见。
"""
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-qrpay-probe.txt"
RESP_BIN = "logs/_lsy-qrpay-resp.bin"

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"])
expire_min = resolve(cfg.get("default-qr-expire-minutes", "15"))[0]

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("  default-qr-expire-minutes= %s            [取自 application.yml]" % expire_min)
say()

# ── 组报文：字段集逐格抄自 LsyOrderService.createQrOrder ─────────────────
now = time.strftime("%Y%m%d%H%M%S")
trace = "probe" + now + "qr"
payload = {
    "inst_no": inst_no,                      # client 的 signedPayload 自动补
    "mch_no": "Y52211090004",                # 文档测试商户号
    "pay_trace_no": trace,                   # 上游要求商户系统内唯一；本探针每跑一次都不同
    "pay_time": now,                         # yyyyMMddHHmmss
    "total_amount": "1",                     # 1 分
    "time_expire": str(expire_min),
    "order_body": "联调探针-主扫",
    # 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("  字段清单（= 生产 createQrOrder 发的集合）：%s" % ", ".join(sorted(payload)))
say()

url = base_url.rstrip("/") + "/v3/qrpay"
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)

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("  code_url    = %s" % resp.get("code_url"))
say()

# ── 入站验签：用上一条已经实证过的「null 不参与」读法 ─────────────────
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))
    say()

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