payment-platform/logs/_lsy-order-refunds-probe.py

# -*- coding: utf-8 -*-
"""只读探针：订单抽屉那条新查询，在**真库**上到底会不会返回行。

它跑的是与新端点**同一条 SQL**（同一个 WHERE / ORDER BY），但**量具不是线上那支** ——
真正的端点是 MyBatis 发出去的、还要过租户拦截器与 @PreAuthorize。本脚本只能证明
「按 out_trade_no 能查到这几行、字段名与形态长这样」，证明不了端点本身。
（这条区分的来历见 memory `probe-instrument-is-not-the-production-instrument`。）

用法: python logs/_lsy-order-refunds-probe.py <out_trade_no> [...]
只读：只发 SELECT。控制台 GBK ⇒ stdout 只打 ASCII，正文写 UTF-8 文件。
"""
import io
import os
import re
import sys

import psycopg2

ROOT = r"D:\project2\payment-platform"
YML = os.path.join(ROOT, "java", "src", "main", "resources", "application.yml")
OUT = os.path.join(ROOT, "logs", "_lsy-order-refunds-probe.txt")

buf = []


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


def db():
    y = io.open(YML, encoding="utf-8", errors="replace").read()
    url = re.search(r"(jdbc:postgresql://[^\s\"']+)", y).group(1)
    tail = y[y.index(url):]
    user = re.search(r"username:\s*(\S+)", tail).group(1)
    pwd = re.search(r"password:\s*(\S+)", tail).group(1)
    m = re.search(r"//([^:/]+):(\d+)/([^?\s]+)", url)
    return psycopg2.connect(host=m.group(1), port=int(m.group(2)), dbname=m.group(3),
                            user=user, password=pwd, connect_timeout=8)


orders = [a for a in sys.argv[1:] if not a.startswith("--")]
if not orders:
    raise SystemExit("用法: python logs/_lsy-order-refunds-probe.py <out_trade_no> [...]")

conn = db()
cur = conn.cursor()

# 后端实体字段 → 全局 SNAKE_CASE 之后前端要读的键（这是本探针要复核的那件事）
COLS = ("out_refund_no", "refund_fee", "status", "refund_no", "finish_time")

for o in orders:
    say("=" * 72)
    say("out_trade_no = %s" % o)
    cur.execute("""SELECT status, total_amount, refunded_amount
                   FROM pay_lsy_order WHERE out_trade_no = %s""", (o,))
    row = cur.fetchone()
    say("  pay_lsy_order: %s" % (row,))
    if not row:
        say("  ! 库里没有这笔订单")
        continue

    # ↓↓↓ 与新端点 refundsOf() 同一条查询 ↓↓↓
    cur.execute("""SELECT out_refund_no, refund_fee, status, refund_no, finish_time,
                          refund_trace_no, created_time
                   FROM pay_lsy_refund WHERE out_trade_no = %s
                   ORDER BY created_time DESC""", (o,))
    rows = cur.fetchall()
    say("  新端点那条查询返回 %d 行" % len(rows))
    for r in rows:
        say("    " + " | ".join("%s=%s" % (c, v) for c, v in zip(
            ("out_refund_no", "refund_fee", "status", "refund_no", "finish_time",
             "refund_trace_no", "created_time"), r)))
    # ★ 下面这几行量的是**表里的列**，不是**响应里的键** —— 两回事，别把标题写成后者。
    #   列存在 ⇒ 该字段在实体上有映射（`LsyRefundEntity` 逐字段读过；`status` 来自父类
    #   `PaymentBaseEntity`），SNAKE_CASE 之后键名与列名一致，故键也在。但这条推理链的
    #   最后一跳**本脚本没跑**（它不发 HTTP、不过 Jackson），真正的端到端读数得由接口给。
    cur.execute("""SELECT column_name FROM information_schema.columns
                   WHERE table_name = 'pay_lsy_refund'""")
    have = set(x[0] for x in cur.fetchall())
    for c in COLS:
        say("  表里的列 `%s` %s（响应键名同一套 SNAKE_CASE；端到端未由本脚本验证）"
            % (c, "存在" if c in have else "**不存在**"))

    # 回调轨迹那条：IN (订单 trace ∪ 每笔退款的 trace)。只数行，证明退款回调行确实存在。
    cur.execute("""SELECT pay_trace_no FROM pay_lsy_order WHERE out_trade_no = %s""", (o,))
    pt = cur.fetchone()[0]
    cur.execute("""SELECT refund_trace_no FROM pay_lsy_refund WHERE out_trade_no = %s""", (o,))
    traces = [pt] + [t[0] for t in cur.fetchall() if t[0]]
    say("  回调轨迹的 trace 集合 = %d 个" % len(traces))
    cur.execute("""SELECT to_char(created_time,'HH24:MI:SS.MS'), trade_type, verify_status,
                          process_status, pay_trace_no
                   FROM pay_lsy_notify_log WHERE pay_trace_no = ANY(%s)
                   ORDER BY created_time DESC""", (traces,))
    logs = cur.fetchall()
    say("  回调轨迹命中 %d 行（改前只按订单 trace 查，命中 %d 行）"
        % (len(logs), sum(1 for l in logs if l[4] == pt)))
    for l in logs:
        say("    %s" % (l,))

conn.close()
io.open(OUT, "w", encoding="utf-8").write("\n".join(buf) + "\n")
say("")
say("正文已写入 %s" % os.path.relpath(OUT, ROOT))