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

# -*- coding: utf-8 -*-
"""用刚下的那笔主扫单，走一遍**对账那条腿**的查询。

为什么单列一发：
  `LsyQueryService.queryOrder` 有两条腿 —— 有 trade_no 用 trade_no，没有则用
  pay_trace_no + pay_time。而 /v3/qrpay 的应答**根本不含 trade_no**（文档的返回表没有这个字段，
  实测也没有），所以「主扫单被对账捞起时」走的一定是 **pay_trace_no + pay_time 那条分支**。
  这条分支此前从没被真上游读过一次。

  pay_trace_no / pay_time **直接从上一发的原始应答字节里读**（logs/_lsy-qrpay-resp.bin），
  不手抄 —— 手抄的常数是失真源。

预期：一笔记账中（没人扫过码）的聚合码单 ⇒ result_code 应为 PAY_IN_PROCESS，
      而不是「订单不存在」。若回「订单不存在」⇒ 下单与查询两条腿对同一笔单的键不一致。

★ 凭据同前：密钥只从 application.yml 读、只打长度。★ 只读请求，不创建不修改。
"""
import json
import os
import re
import sys
import uuid
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_]+):(.*)\}$")
SRC_BIN = "logs/_lsy-qrpay-resp.bin"
OUT = "logs/_lsy-query-probe.txt"

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 = resolve(cfg["base-url"])[0]
inst_no = resolve(cfg["inst-no"])[0]
key = resolve(cfg["key"])[0]

if not os.path.exists(SRC_BIN):
    raise SystemExit("缺少 %s —— 先跑 _lsy-qrpay-probe.py" % SRC_BIN)
prev = json.loads(open(SRC_BIN, "rb").read().decode("utf-8"))

trace = prev.get("pay_trace_no")
ptime = prev.get("pay_time")
if not trace or not ptime:
    raise SystemExit("上一发应答里没有 pay_trace_no/pay_time，无法查：%s" % sorted(prev))

say("── 被查的那笔单（读自上一发的原始应答，非手抄）──")
say("  pay_trace_no = %s" % trace)
say("  pay_time     = %s" % ptime)
say("  上一发是否给了 trade_no : %s  （qrpay 返回表里本就没有这个字段）" % ("trade_no" in prev))
say()

payload = {
    "inst_no": inst_no,
    "mch_no": prev.get("mch_no") or "Y52211090004",
    "query_trace_no": uuid.uuid4().hex,     # 上游要求每次查询唯一
    # ↓ 走 LsyQueryService 的**无 trade_no 分支**
    "pay_trace_no": trace,
    "pay_time": ptime,
}

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 打码）──")
shown = dict(payload)
shown["sign"] = "<len=%d>" % len(sign)
say(json.dumps(shown, ensure_ascii=False, indent=2))
say()

url = base_url.rstrip("/") + "/v3/queryorder"
body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
say("── 出站 ──")
say("  POST %s" % url)
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("logs/_lsy-query-resp.bin", "wb").write(raw)
resp = json.loads(raw.decode("utf-8"))

say("── 上游应答 ──")
say("  http_code = %s, bytes = %d" % (code, len(raw)))
say(json.dumps(resp, ensure_ascii=False, indent=2))
say()

# 入站验签：null 不参与（已两次实证的读法）
usign = resp.get("sign")
if usign:
    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" % len(nonnull))
    say("  重算 == 上游 sign : %s" % (mine == usign))
    say()

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