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

# -*- coding: utf-8 -*-
"""按 trade_no 查一次 /v3/queryorder —— 判 H5 对账走哪条分支。

背景（实测，非推断）：
  `LsyQueryService.queryOrder` 的逻辑是「order 上有 trade_no 就用 trade_no，
  否则用 pay_trace_no + pay_time」。
  主扫（/v3/qrpay）的应答**没有 trade_no** ⇒ 主扫单走第二条分支（已实证）。
  但 H5（/v3/h5pay）的 PAY_SUCCESS 应答**带 trade_no**（logs/_lsy-h5pay-resp.bin 里实测有）
  ⇒ H5 单落库时 trade_no 非空 ⇒ 对账时走的是**第一条分支**。
  而代码注释里写的是「qrpay 与 h5pay 那两个路径传的 patch 本来就没有 trade_no」——
  与实测不符（h5 有）。本发就是要判：**按 trade_no 到底查不查得到这笔单**。

★ 只读请求。trade_no 从上一发的原始应答字节里读，不手抄。
★ 凭据同前：密钥只从 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-h5pay-resp.bin"
OUT = "logs/_lsy-query-by-tradeno-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-h5pay-probe.py" % SRC_BIN)
prev = json.loads(open(SRC_BIN, "rb").read().decode("utf-8"))

trade_no = prev.get("trade_no")
if not trade_no:
    raise SystemExit("H5 那一发的应答里没有 trade_no ⇒ 本探针要判的前提不成立：%s" % sorted(prev))

say("── 被查的那笔单（读自上一发的原始应答，非手抄）──")
say("  trade_no     = %s" % trade_no)
say("  pay_trace_no = %s" % prev.get("pay_trace_no"))
say("  result_code  = %s  （下单受理，尚未被扫/未支付）" % prev.get("result_code"))
say()

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

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-by-tradeno-resp.bin", "wb").write(raw)
resp = json.loads(raw.decode("utf-8"))

say("── 上游应答 ──")
say("  http_code = %s, bytes = %d" % (code, len(raw)))
for k in sorted(resp):
    v = resp[k]
    say("  %-16s = %s" % (k, "null" if v is None else repr(v)))
say()

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))
else:
    say("── 应答里没有 sign 字段 ──")
say()

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