payment-platform/logs/_lsy-lifecycle.py

# -*- coding: utf-8 -*-
"""乐收赢全链路联调台子：收单 / 查单 / 退款 / 退款查询 / 关单 / 撤销。

设计要点
  * 每一条出站都把**原始应答字节**落盘（logs/_lsy-lc/<tag>.bin），报告里只写结论。
  * 每一条出站都做**入站验签**（null 不参与）—— 验签绝不跳过。
  * 单条失败不中断：整轮跑完，末尾给汇总表。
  * 控制台是 GBK ⇒ stdout 只打 ASCII，正文进 UTF-8 文件。
  * 密钥只从 application.yml 读，只打长度。

用法: python logs/_lsy-lifecycle.py [section ...]
      section ∈ create, query, refund, close, cancel, supply, all（默认 all）
"""
import io
import json
import os
import re
import sys
import time
import uuid
import urllib.error
import urllib.request
from hashlib import md5

YML = "java/src/main/resources/application.yml"
VAR = re.compile(r"^\$\{([A-Za-z0-9_]+):(.*)\}$")
OUTDIR = "logs/_lsy-lc"
OUT = "logs/_lsy-lifecycle.txt"

buf = []
rows = []


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


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


# ---------------------------------------------------------------- 配置
def read_block(path, top):
    lines = io.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:
            out[m.group(1)] = re.split(r"\s+#", m.group(2), maxsplit=1)[0].strip().strip("'\"")
    return out


def resolve(raw):
    m = VAR.match(raw)
    if not m:
        return raw
    var, default = m.group(1), m.group(2)
    return os.environ.get(var) or default


cfg = read_block(YML, "leshouying")
BASE = resolve(cfg["base-url"]).rstrip("/")
INST = resolve(cfg["inst-no"])
KEY = resolve(cfg["key"])
MCH = "Y52211090004"

if not os.path.isdir(OUTDIR):
    os.makedirs(OUTDIR)


# ---------------------------------------------------------------- 出站
def call(path, params, tag, expect_sign=True):
    """签名 → POST → 落盘 → 入站验签。返回 (http_code, dict|None, raw)。"""
    p = dict(params)
    p.setdefault("inst_no", INST)
    p.setdefault("mch_no", MCH)
    A = "&".join("%s=%s" % (k, p[k]) for k in sorted(p) if p[k] is not None)
    p["sign"] = md5((A + "&key=" + KEY).encode("utf-8")).hexdigest().lower()
    body = json.dumps(p, ensure_ascii=False).encode("utf-8")
    req = urllib.request.Request(BASE + path, 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:  # noqa: BLE001
        say("  !! 请求未完成 %s: %r" % (tag, e))
        row(tag, path, "TRANSPORT_ERR", repr(e)[:80])
        return None, None, b""

    io.open("%s/%s.bin" % (OUTDIR, tag), "wb").write(raw)
    try:
        resp = json.loads(raw.decode("utf-8"))
    except Exception:  # noqa: BLE001
        say("  !! 应答不是 JSON（%d B）" % len(raw))
        row(tag, path, "NON_JSON", "%d B" % len(raw))
        return code, None, raw

    # 入站验签（null 不参与）
    usign = resp.get("sign")
    vs = "无 sign"
    if usign:
        nn = {k: v for k, v in resp.items() if k != "sign" and v is not None}
        A2 = "&".join("%s=%s" % (k, nn[k]) for k in sorted(nn))
        vs = "验签OK(%d)" % len(nn) if md5((A2 + "&key=" + KEY).encode("utf-8")).hexdigest().lower() == usign else "验签FAIL"
    elif expect_sign:
        vs = "无sign(异常)"

    rcode = resp.get("return_code")
    rescode = resp.get("result_code")
    say("  POST %-22s -> http %s  %4d B  return_code=%s result_code=%s  %s"
        % (path, code, len(raw), rcode, rescode, vs))
    say("       return_msg = %s" % resp.get("return_msg"))
    row(tag, path, "%s/%s" % (rcode, rescode), "%s; %s" % (str(resp.get("return_msg"))[:40], vs))
    return code, resp, raw


def stamp():
    return time.strftime("%Y%m%d%H%M%S")


def uid(prefix):
    return "%s%s%s" % (prefix, time.strftime("%H%M%S"), uuid.uuid4().hex[:6])


STATE = {}          # 跨用例共享的句柄


# ---------------------------------------------------------------- E 补录
def sec_supply():
    say("=" * 78)
    say("【补录】/v3/createSupply —— 造一笔「已支付成功」的单，解锁退款腿")
    say("=" * 78)
    trace = uid("sup")
    now = stamp()
    settle = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(time.time() - 60))
    _, resp, _ = call("/v3/createSupply", {
        "channel_type": "37",          # 37=建行（本商户名就是「建行小店…」）
        "pay_mode": "2",               # 2=扫码支付
        "type": "300",                 # 300=支付宝
        "function_flag": "1",          # 1=补单交易成功订单
        "pay_time": now,
        "settle_time": settle,
        "pay_trace_no": trace,
        "total_fee": "1",
        "order_body": "联调补录",
    }, "supply-1")
    if resp:
        say("       data(乐收赢订单号) = %s" % resp.get("data"))
        STATE["sup_trace"] = trace
        STATE["sup_time"] = now
        STATE["sup_no"] = resp.get("data")
    say()


# ---------------------------------------------------------------- A 收单
def sec_create():
    say("=" * 78)
    say("【收单】三条腿 + 参数变体")
    say("=" * 78)

    # A1 聚合码：最小参数集（= 生产 createQrOrder 发的集合）
    t1, p1 = uid("qr"), stamp()
    _, r1, _ = call("/v3/qrpay", {
        "pay_trace_no": t1, "pay_time": p1, "total_amount": "1",
        "time_expire": "15", "order_body": "联调-主扫-最小",
    }, "create-qrpay-min")
    if r1:
        STATE["qr"] = {"trace": t1, "time": p1, "code_url": r1.get("code_url"), "resp": r1}
    say()

    # A2 聚合码：全参数
    t2, p2 = uid("qf"), stamp()
    _, r2, _ = call("/v3/qrpay", {
        "pay_trace_no": t2, "pay_time": p2, "total_amount": "2",
        "time_expire": "30", "order_body": "联调-主扫-全参",
        "attach": "A2", "notify_url": "https://example.com/notify",
        "front_url": "https://example.com/front", "auto_payment": "0",
        "undiscount_amount": "1",
    }, "create-qrpay-full")
    if r2:
        STATE["qr2"] = {"trace": t2, "time": p2, "code_url": r2.get("code_url")}
    say()

    # A3 同号重下（幂等/拒绝）
    if STATE.get("qr"):
        call("/v3/qrpay", {
            "pay_trace_no": STATE["qr"]["trace"], "pay_time": STATE["qr"]["time"],
            "total_amount": "1", "order_body": "联调-重复单号",
        }, "create-qrpay-dup")
    say()

    # A4 H5
    t4, p4 = uid("h5"), stamp()
    _, r4, _ = call("/v3/h5pay", {
        "pay_trace_no": t4, "pay_time": p4, "total_amount": "1",
        "pay_type": "300", "order_body": "联调-H5",
    }, "create-h5pay")
    if r4:
        STATE["h5"] = {"trace": t4, "time": p4, "trade_no": r4.get("trade_no"),
                       "code_url": r4.get("code_url")}
    say()

    # A5 被扫（无真付款码，预期失败；仍记一条腿的读数）
    t5, p5 = uid("bc"), stamp()
    call("/v3/pay", {
        "pay_trace_no": t5, "pay_time": p5, "total_amount": "1",
        "auth_code": "284812345678901234", "pay_type": "300", "order_body": "联调-被扫",
    }, "create-pay-barcode")
    say()

    # A6 被扫：未知条码类型（错误路径）
    t6, p6 = uid("bx"), stamp()
    call("/v3/pay", {
        "pay_trace_no": t6, "pay_time": p6, "total_amount": "1",
        "auth_code": "0" * 18, "pay_type": "000",
    }, "create-pay-badcode")
    say()


# ---------------------------------------------------------------- B 查单
def sec_query():
    say("=" * 78)
    say("【查单】/v3/queryorder —— 五条键路")
    say("=" * 78)

    h5 = STATE.get("h5")
    if h5 and h5.get("trade_no"):
        call("/v3/queryorder", {"query_trace_no": uuid.uuid4().hex,
                                "trade_no": h5["trade_no"]}, "query-h5-bytradeno")
        call("/v3/queryorder", {"query_trace_no": uuid.uuid4().hex,
                                "pay_trace_no": h5["trace"], "pay_time": h5["time"]},
             "query-h5-bytrace")
        call("/v3/queryorder", {"query_trace_no": uuid.uuid4().hex,
                                "pay_trace_no": h5["trace"]}, "query-h5-traceONLY")
        call("/v3/queryorder", {"query_trace_no": uuid.uuid4().hex,
                                "trade_no": h5["trade_no"], "pay_trace_no": h5["trace"],
                                "pay_time": h5["time"]}, "query-h5-both-keys")
    say()

    qr = STATE.get("qr")
    if qr:
        call("/v3/queryorder", {"query_trace_no": uuid.uuid4().hex,
                                "pay_trace_no": qr["trace"], "pay_time": qr["time"]},
             "query-qrpay-unscanned")
    say()

    sup = STATE.get("sup_trace")
    if sup:
        call("/v3/queryorder", {"query_trace_no": uuid.uuid4().hex,
                                "pay_trace_no": sup, "pay_time": STATE["sup_time"]},
             "query-supply-bytrace")
        if STATE.get("sup_no"):
            call("/v3/queryorder", {"query_trace_no": uuid.uuid4().hex,
                                    "trade_no": STATE["sup_no"]}, "query-supply-bytradeno")
    say()

    call("/v3/queryorder", {"query_trace_no": uuid.uuid4().hex,
                            "pay_trace_no": "nosuchorder" + uuid.uuid4().hex[:8],
                            "pay_time": stamp()}, "query-nonexistent")
    say()

    # 键缺失：两个键都不发（文档说二选一必传）
    call("/v3/queryorder", {"query_trace_no": uuid.uuid4().hex}, "query-nokey")
    say()


# ---------------------------------------------------------------- D 退款
def sec_refund():
    say("=" * 78)
    say("【退款】/v3/refundorder + /v3/queryrefund")
    say("=" * 78)

    # D1 对**未支付**的聚合码单退款（错误路径，看上游怎么拒）
    qr = STATE.get("qr")
    if qr:
        call("/v3/refundorder", {
            "refund_trace_no": uid("rf"), "refund_fee": "1",
            "pay_time": stamp(), "pay_trace_no": qr["trace"],
            "pay_trace_time": qr["time"],
        }, "refund-unpaid-qrpay")
    say()

    # D2 对**补录成功**的单退款（正路）
    sup = STATE.get("sup_trace")
    if sup:
        rt = uid("rf")
        _, r, _ = call("/v3/refundorder", {
            "refund_trace_no": rt, "refund_fee": "1", "pay_time": stamp(),
            "pay_trace_no": sup, "pay_trace_time": STATE["sup_time"],
        }, "refund-supply-bytrace")
        STATE["refund"] = {"trace": rt, "resp": r}
        if r and not r.get("refund_no") and r.get("refund_trace_no"):
            STATE["refund"]["trace"] = r.get("refund_trace_no")
    say()

    # D3 退款查询
    if STATE.get("refund"):
        rf = STATE["refund"]
        if rf["resp"] and rf["resp"].get("refund_no"):
            call("/v3/queryrefund", {"query_trace_no": uuid.uuid4().hex,
                                     "refund_no": rf["resp"]["refund_no"]}, "queryrefund-byno")
        call("/v3/queryrefund", {"query_trace_no": uuid.uuid4().hex,
                                 "refund_trace_no": rf["trace"],
                                 "refund_time": stamp()}, "queryrefund-bytrace")
    say()

    # D4 超额退款
    if sup:
        call("/v3/refundorder", {
            "refund_trace_no": uid("rx"), "refund_fee": "999999",
            "pay_time": stamp(), "pay_trace_no": sup,
            "pay_trace_time": STATE["sup_time"],
        }, "refund-overamount")
    say()

    # D5 退款不存在的单
    call("/v3/refundorder", {
        "refund_trace_no": uid("rn"), "refund_fee": "1", "pay_time": stamp(),
        "pay_trace_no": "nosuch" + uuid.uuid4().hex[:8], "pay_trace_time": stamp(),
    }, "refund-nonexistent")
    say()


# ---------------------------------------------------------------- C 关单
def sec_close():
    say("=" * 78)
    say("【关单】/v3/qrclose（聚合码，按 code_url）+ 幂等 + 错误路径")
    say("=" * 78)

    qr = STATE.get("qr")
    if qr and qr.get("code_url"):
        call("/v3/qrclose", {"code_url": qr["code_url"]}, "close-qr-1st")
        call("/v3/qrclose", {"code_url": qr["code_url"]}, "close-qr-2nd")
        call("/v3/queryorder", {"query_trace_no": uuid.uuid4().hex,
                                "pay_trace_no": qr["trace"], "pay_time": qr["time"]},
             "query-qrpay-afterclose")
    say()

    call("/v3/qrclose", {"code_url": "https://test.gdyfsk.com/yfpay/s/s-99999"},
         "close-qr-nonexistent")
    call("/v3/qrclose", {}, "close-qr-nocodeurl")
    say()

    # H5 / 聚合码的通用关单腿（我方**没接** /v3/closeorder，这里顺手量一下它认不认）
    h5 = STATE.get("h5")
    if h5 and h5.get("trade_no"):
        call("/v3/closeorder", {
            "close_trace_no": uid("cl"), "close_time": stamp(),
            "trade_no": h5["trade_no"],
        }, "closeorder-h5-bytradeno")
    if STATE.get("qr"):
        call("/v3/closeorder", {
            "close_trace_no": uid("cl"), "close_time": stamp(),
            "pay_trace_no": STATE["qr"]["trace"], "pay_time": STATE["qr"]["time"],
        }, "closeorder-qr-bytrace")
    say()


# ---------------------------------------------------------------- F 撤销
def sec_cancel():
    say("=" * 78)
    say("【撤销】/v3/cancelorder（被扫）")
    say("=" * 78)
    call("/v3/cancelorder", {
        "cancel_trace_no": uid("cx"), "cancel_time": stamp(),
        "pay_trace_no": "nosuch" + uuid.uuid4().hex[:8], "pay_time": stamp(),
    }, "cancel-nonexistent")
    say()


SECTIONS = {
    "supply": sec_supply, "create": sec_create, "query": sec_query,
    "refund": sec_refund, "close": sec_close, "cancel": sec_cancel,
}
ORDER = ["supply", "create", "query", "close", "refund", "cancel"]


def main():
    want = sys.argv[1:] or ["all"]
    todo = ORDER if "all" in want else [s for s in ORDER if s in want]
    say("乐收赢全链路联调 —— 测试环境 %s / inst_no=%s / mch_no=%s / key=<len=%d>"
        % (BASE, INST, MCH, len(KEY)))
    say("时间 %s" % time.strftime("%Y-%m-%d %H:%M:%S"))
    say()
    for s in todo:
        try:
            SECTIONS[s]()
        except Exception as e:  # noqa: BLE001
            say("!! 段落 %s 抛异常: %r" % (s, e))
            say()

    say("=" * 78)
    say("汇总（%d 条）" % len(rows))
    say("=" * 78)
    say("%-26s %-22s %-22s %s" % ("tag", "path", "return/result", "note"))
    for t, p, v, d in rows:
        say("%-26s %-22s %-22s %s" % (t, p, v, d))

    io.open(OUT, "w", encoding="utf-8").write("\n".join(buf) + "\n")
    print("report -> %s" % OUT)


main()