payment-platform/logs/_lsy-refund-live.py

# -*- coding: utf-8 -*-
"""退款实弹联调：拿**已有的一笔成功订单**走我方开放 API 退款，再查退款单。

与 logs/_lsy-lifecycle.py 的 refund 段的区别：那个直连上游、绕过我方全链路，库里不留痕；
这个走 `/payment/openapi/pay/refund`，验签 + 租户闸 + 退款单落库 + 上游退款全都要过。

设计要点（与 _lsy-barcode-live.py 同规矩）
  * 租户**从被退的那笔订单反查**，不按「第一个可用商户」挑 —— 挑错租户会红在
    「订单不存在」上，那是夹具错不是被测对象错。
  * key/secret 只打长度；单号只打长度与形态，不打印整串。
  * 每条出站的**原始应答字节**落盘 logs/_lsy-rf/<tag>.bin，报告里只写结论。
  * 控制台是 GBK ⇒ stdout 只打 ASCII，正文进 UTF-8 文件。
  * 直连库复核，不信接口自述。

用法: python logs/_lsy-refund-live.py <out_trade_no> [退款分，默认全额]
"""
import io
import json
import os
import re
import sys
import time
import uuid
import urllib.error
import urllib.request
from hashlib import sha256
from urllib.parse import quote

import psycopg2

ROOT = r"D:\project2\payment-platform"
YML = os.path.join(ROOT, "java", "src", "main", "resources", "application.yml")
OUTDIR = os.path.join(ROOT, "logs", "_lsy-rf")
OUT = os.path.join(ROOT, "logs", "_lsy-refund-live.txt")
BASE = os.environ.get("LSY_BASE", "http://localhost:8001/api/v1")

buf = []
rows = []


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


def row(tag, verdict, detail=""):
    rows.append((tag, verdict, detail))


def save(tag, raw):
    os.makedirs(OUTDIR, exist_ok=True)
    p = os.path.join(OUTDIR, tag + ".bin")
    with open(p, "wb") as f:
        f.write(raw)
    return p


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)


# 与 _lsy-barcode-live.py 逐字同源（文档 §3.3）
def build_signature(secret, body):
    parts = []
    for key in sorted(body):
        v = body[key]
        if v is None or v == "" or v == {} or v == [] or key == "sign":
            continue
        if isinstance(v, (dict, list)):
            text = json.dumps(v, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
        else:
            text = str(v)
        parts.append("%s=%s" % (key, quote(text, safe="")))
    sign_str = "&".join(parts)
    return sign_str, __import__("hmac").new(secret.encode("utf-8"),
                                            sign_str.encode("utf-8"),
                                            sha256).hexdigest()


def call(path, body, key, secret, tag):
    raw = json.dumps(body, ensure_ascii=False).encode("utf-8")
    req = urllib.request.Request(BASE + path, data=raw, method="POST")
    req.add_header("Content-Type", "application/json")
    req.add_header("Authorization", "ApiKey " + key)
    _, sig = build_signature(secret, body)
    req.add_header("Signature", sig)
    try:
        with urllib.request.urlopen(req, timeout=60) as r:
            status, data = r.status, r.read()
    except urllib.error.HTTPError as e:
        status, data = e.code, e.read()
    p = save(tag, data)
    try:
        parsed = json.loads(data.decode("utf-8"))
    except Exception:
        parsed = data.decode("utf-8", "replace")
    return status, parsed, p


# ================================================================ 主流程
_pos = [a for a in sys.argv[1:] if not a.startswith("--")]
if not _pos:
    raise SystemExit("用法: python logs/_lsy-refund-live.py <out_trade_no> [退款分]")
ORDER = _pos[0]
WANT_FEE = int(_pos[1]) if len(_pos) > 1 else None

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

cur.execute("""SELECT tenant_id, mch_no, status, total_amount, receipt_amount, refunded_amount,
                      trade_no, pay_trace_no
               FROM pay_lsy_order WHERE out_trade_no = %s""", (ORDER,))
o = cur.fetchone()
if not o:
    raise SystemExit("库里没有订单 %s" % ORDER)
TENANT, MCH, OSTATUS, TOTAL, RECEIPT, REFUNDED = o[0], o[1], o[2], o[3], o[4], o[5]
say("被退订单 %s" % ORDER)
say("  tenant_id=%s  mch_no=%s  status=%s" % (TENANT, MCH, OSTATUS))
say("  total=%s  已退=%s  可退=%s" % (TOTAL, REFUNDED, (TOTAL or 0) - (REFUNDED or 0)))
if OSTATUS not in ("SUCCESS", "PARTIAL_REFUNDED"):
    raise SystemExit("订单状态 %s 不可退，中止" % OSTATUS)

FEE = WANT_FEE if WANT_FEE is not None else (TOTAL - REFUNDED)
if FEE <= 0 or FEE > (TOTAL - REFUNDED or 0):
    raise SystemExit("退款金额 %s 不在可退区间内，中止" % FEE)

cur.execute("SELECT api_key, api_secret FROM sys_tenant_api_key "
            "WHERE tenant_id = %s ORDER BY id LIMIT 1", (TENANT,))
k = cur.fetchone()
if not k:
    raise SystemExit("租户 %s 下一把 api_key 都没有" % TENANT)
API_KEY, API_SECRET = k
say("  用该租户的 key：api_key=<len %d>  api_secret=<len %d>   （值不打印）"
    % (len(API_KEY), len(API_SECRET)))

oRefundNo = ("RF%s%s" % (time.strftime("%H%M%S"), uuid.uuid4().hex[:6]))[:32]
say("  out_refund_no = prefix=RF len=%d  退款 %s 分" % (len(oRefundNo), FEE))

# ---- 1. 退款
say("")
say("=" * 70)
say("  第 1 步  POST /payment/openapi/pay/refund")
say("=" * 70)
body = {"out_trade_no": ORDER, "out_refund_no": oRefundNo, "refund_fee": FEE}
st, parsed, p = call("/payment/openapi/pay/refund", body, API_KEY, API_SECRET, "refund")
say("HTTP %s   原始应答 %d 字节 → %s" % (st, os.path.getsize(p), os.path.relpath(p, ROOT)))
say(json.dumps(parsed, ensure_ascii=False, indent=2))
data = parsed.get("data") if isinstance(parsed, dict) else None
row("退款", "HTTP %s" % st, "status=%s" % ((data or {}).get("status")))

# ---- 2. 退款查询
say("")
say("=" * 70)
say("  第 2 步  POST /payment/openapi/pay/refund/query")
say("=" * 70)
st2, parsed2, p2 = call("/payment/openapi/pay/refund/query", {"out_refund_no": oRefundNo},
                        API_KEY, API_SECRET, "refundquery")
say("HTTP %s   原始应答 %d 字节 → %s" % (st2, os.path.getsize(p2), os.path.relpath(p2, ROOT)))
say(json.dumps(parsed2, ensure_ascii=False, indent=2))
row("退款查询", "HTTP %s" % st2,
    "local_status=%s" % (((parsed2.get("data") or {}) if isinstance(parsed2, dict) else {}).get("local_status")))

# ---- 3. 查单（看订单侧是否被回写）
say("")
say("=" * 70)
say("  第 3 步  POST /payment/openapi/pay/order/query")
say("=" * 70)
st3, parsed3, p3 = call("/payment/openapi/pay/order/query", {"out_trade_no": ORDER},
                        API_KEY, API_SECRET, "orderquery")
say("HTTP %s   原始应答 %d 字节 → %s" % (st3, os.path.getsize(p3), os.path.relpath(p3, ROOT)))
say(json.dumps(parsed3, ensure_ascii=False, indent=2))
row("查单", "HTTP %s" % st3,
    "local_status=%s" % (((parsed3.get("data") or {}) if isinstance(parsed3, dict) else {}).get("local_status")))

# ---- 4. 直连库复核
say("")
say("=" * 70)
say("  第 4 步  直连库复核（不信接口自述）")
say("=" * 70)
conn.rollback()  # 丢掉只读事务快照，读最新
# 列名逐个取自 LsyRefundEntity（含 PaymentBaseEntity 继承来的 status）——此处原写 refund_reason，
# 那一列**本表没有**（它在汇安融那张表上），一跑就 UndefinedColumn。别凭记忆改这里，去读实体。
cur.execute("""SELECT out_refund_no, refund_no, status, refund_fee, refund_trace_no,
                      upstream_result_code, upstream_result_msg, created_time, updated_time
               FROM pay_lsy_refund WHERE out_refund_no = %s""", (oRefundNo,))
r = cur.fetchone()
if r:
    cols = ["out_refund_no", "refund_no", "status", "refund_fee", "refund_trace_no",
            "upstream_result_code", "upstream_result_msg", "created_time", "updated_time"]
    for c, v in zip(cols, r):
        say("  %-22s = %s" % (c, v))
    row("退款单落库", "status=%s" % r[2], "fee=%s refund_no=%s" % (r[3], r[1]))
else:
    say("  ！pay_lsy_refund 里没有 out_refund_no=%s 这一行" % oRefundNo)
    row("退款单落库", "缺行", "")

cur.execute("""SELECT status, total_amount, refunded_amount
               FROM pay_lsy_order WHERE out_trade_no = %s""", (ORDER,))
o2 = cur.fetchone()
say("")
say("  订单侧：status=%s  total=%s  refunded_amount=%s" % o2)
row("订单回写", "status=%s" % o2[0], "refunded=%s" % o2[2])

# 退款回调（trade_type=0）—— 上送上游的 notify_url 指向公网域名，那台与本机共用同一个库，
# 所以这里**可能**看得到它写的行；看到即证「退款回调确实到达并被处理」。
cur.execute("""SELECT id, to_char(created_time,'HH24:MI:SS.MS'), trade_type, refund_no,
                      verify_status, process_status, tenant_id, error_msg
               FROM pay_lsy_notify_log
               WHERE created_time > now() - interval '15 minutes'
               ORDER BY id""")
nl = cur.fetchall()
say("")
say("近 15 分钟 pay_lsy_notify_log 行数 = %d" % len(nl))
for n in nl:
    say("  %s" % (n,))
row("回调日志", "%d 行" % len(nl), "")

conn.close()

say("")
say("=" * 70)
say("  汇总")
say("=" * 70)
for t, v, d in rows:
    say("  %-12s | %-22s | %s" % (t, v, d))

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