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

# -*- coding: utf-8 -*-
"""判决探针：拿一笔**已知可查**的单，改用 `pay_trace_no + pay_time` 分支再查一次。

要分的两个假设（此前无法分辨）：
  H1 `queryorder` 按 `pay_trace_no+pay_time` 能查到单，只是**未扫的聚合码单**查不到
     ⇒ 「聚合码未扫 = 不在查询域」成立，属上游语义问题。
  H2 `pay_trace_no+pay_time` 这条分支**根本不通**（至少在本测试环境/本商户上）
     ⇒ 原结论是用错键造出来的，对账腿对**所有**走这条分支的单都瞎，性质完全不同。

量具：用 H5 那笔单（`logs/_lsy-h5pay-resp.bin`）—— 它的 trade_no 分支**实测查得到**
（`logs/_lsy-query-by-tradeno-resp.bin`，614 B / PAY_IN_PROCESS）。同一笔单换个键：
查得到 ⇒ H1；还是「订单信息不存在」⇒ H2。

★ 只读查询。键值一律从原始应答字节里读，不手抄。
★ 密钥只从 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"
CTRL_BIN = "logs/_lsy-query-by-tradeno-resp.bin"
OUT = "logs/_lsy-query-by-traceno-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]

prev = json.loads(open(SRC_BIN, "rb").read().decode("utf-8"))
ctrl = json.loads(open(CTRL_BIN, "rb").read().decode("utf-8"))

trace = prev.get("pay_trace_no")
ptime = prev.get("pay_time")
if not (trace and ptime):
    raise SystemExit("H5 应答里缺 pay_trace_no/pay_time ⇒ 前提不成立: %s" % sorted(prev))

say("── 先摆正对照：同一笔单，trade_no 分支的既有读数（读自原始应答字节）──")
say("  trade_no     = %s" % prev.get("trade_no"))
say("  pay_trace_no = %s" % trace)
say("  pay_time     = %s" % ptime)
say("  正对照读数   = return_code=%s / return_msg=%s / result_code=%s"
    % (ctrl.get("return_code"), ctrl.get("return_msg"), ctrl.get("result_code")))
say("  正对照必须是「查得到」—— 否则本探针的对照组不成立")
if ctrl.get("result_code") is None:
    raise SystemExit("正对照本身就没查到 ⇒ 换个对照组，别用这笔")
say()

payload = {
    "inst_no": inst_no,
    "mch_no": prev.get("mch_no"),
    "query_trace_no": uuid.uuid4().hex,
    # ↓ 走 LsyQueryService 的**无 trade_no 分支**（故意不发 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("  注意：**没有** trade_no 这个键 ⇒ 走的正是主扫单走的那条分支")
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-traceno-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()

say("── 判决 ──")
if resp.get("result_code") is not None:
    say("  查到了 ⇒ H1：pay_trace_no+pay_time 分支**通**，")
    say("          问题只出在「未扫的聚合码单」这个状态上。")
else:
    say("  没查到 ⇒ H2：这条分支**不通**（本测试环境/本商户），")
    say("          原结论「聚合码未扫查不到」是用错键造出来的，对账腿对走这条分支的单**全瞎**。")
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))
say()

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