yeuimu
5f4e1ffa1f
- api/kuaidi100: 上门取件API客户端(bOrder/price/cancel/detail,MD5签名封装) - api/sds: 变体克重(children接口,24h缓存)/面单PDF上传/挂包裹/carriageNoStatus筛选 - flows/sds/waybill_autogen: 每10分钟遍历运单待生成订单(factoryIds+carriageNoStatus+orderStatusCode) 自动下极兔(returnType=20+tempid出面单),面单回调转PDF回传SDS; 幂等跳过已有运单;面单超时不自动取消(记人工跟进);schedule 默认 active:false - scripts: 手动E2E测试脚本 + 回调接收服务(验签MD5(param+salt)+JSONL落盘,配frpc 18090穿透) - 依赖 +pillow(面单PNG转PDF)
408 lines
16 KiB
Python
408 lines
16 KiB
Python
"""快递100「上门取件API(线上支付)」端到端测试脚本。
|
|
|
|
文档: https://api.kuaidi100.com/document/603cb649a62a19500e19866b
|
|
|
|
流程:
|
|
1. 从 SDS 取一单国内工厂订单(factoryIds=1161),提取收/寄件人地址(收件人=订单 address,寄件人=发货仓 issuing_bay)
|
|
2. 逐家查价(method=price),选最便宜的快递公司
|
|
3. 下单(method=bOrder)拿 taskId/orderId/kuaidinum
|
|
4. 取消订单(method=cancel)收尾,避免遗留测试单
|
|
|
|
鉴权: 公共参数 method/key/sign/t,param 为 JSON 字符串;
|
|
sign = MD5(param + t + key + secret) 转 32 位大写。
|
|
测试环境: http://e-test.kuaidilab.com/api/order/borderapi.do(测试 key/secret 需向快递100技术支持申请)
|
|
|
|
用法:
|
|
uv run python scripts/kuaidi100_pickup_test.py --key KEY --secret SECRET [--customer CUSTOMER]
|
|
# 或把 JSON {"key":"...","secret":"...","customer":"..."} 存进 Prefect Secret Block kuaidi100-test 后:
|
|
uv run python scripts/kuaidi100_pickup_test.py --from-secret
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parent.parent
|
|
sys.path.insert(0, str(REPO_ROOT))
|
|
|
|
TEST_BASE_URL = "http://e-test.kuaidilab.com/api/order/borderapi.do"
|
|
PROD_BASE_URL = "https://poll.kuaidi100.com/order/borderapi.do"
|
|
|
|
# 文档列出的本 API 支持的快递公司编码(小写)
|
|
SUPPORTED_COURIERS = [
|
|
"jd",
|
|
"debangkuaidi",
|
|
"shunfeng",
|
|
"jtexpress",
|
|
"yuantong",
|
|
"shentong",
|
|
"zhongtong",
|
|
"zhongtongkuaiyun",
|
|
"yunda",
|
|
"kuayue",
|
|
"shunfengkuaiyun",
|
|
"ems",
|
|
"yimidida",
|
|
"kuaidi100zhixuan",
|
|
]
|
|
|
|
COURIER_NAMES = {
|
|
"jd": "京东",
|
|
"debangkuaidi": "德邦快递",
|
|
"shunfeng": "顺丰速运",
|
|
"jtexpress": "极兔速递",
|
|
"yuantong": "圆通速递",
|
|
"shentong": "申通快递",
|
|
"zhongtong": "中通快递",
|
|
"zhongtongkuaiyun": "中通快运",
|
|
"yunda": "韵达快递",
|
|
"kuayue": "跨越速运",
|
|
"shunfengkuaiyun": "顺丰快运",
|
|
"ems": "EMS",
|
|
"yimidida": "壹米滴答",
|
|
"kuaidi100zhixuan": "快递100智选",
|
|
}
|
|
|
|
|
|
# ---------- 快递100 公共调用 ----------
|
|
|
|
|
|
def make_sign(param: str, t: str, key: str, secret: str) -> str:
|
|
"""文档签名算法: MD5(param + t + key + secret) 转 32 位大写。"""
|
|
raw = f"{param}{t}{key}{secret}"
|
|
return hashlib.md5(raw.encode("utf-8")).hexdigest().upper()
|
|
|
|
|
|
def call_api(
|
|
method: str,
|
|
param: dict[str, Any],
|
|
auth: dict[str, str],
|
|
base_url: str = TEST_BASE_URL,
|
|
timeout: float = 30.0,
|
|
) -> dict[str, Any]:
|
|
"""POST application/x-www-form-urlencoded,公共参数 method/key/sign/t/param。"""
|
|
param_str = json.dumps(param, ensure_ascii=False, separators=(",", ":"))
|
|
t = str(int(time.time() * 1000))
|
|
body = {
|
|
"method": method,
|
|
"key": auth["key"],
|
|
"sign": make_sign(param_str, t, auth["key"], auth["secret"]),
|
|
"t": t,
|
|
"param": param_str,
|
|
}
|
|
resp = httpx.post(base_url, data=body, timeout=timeout)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|
|
|
|
def check(resp: dict[str, Any], action: str) -> dict[str, Any]:
|
|
if resp.get("result") is not True or str(resp.get("returnCode")) != "200":
|
|
raise RuntimeError(f"[{action}] 失败: {json.dumps(resp, ensure_ascii=False)[:500]}")
|
|
print(f"[{action}] 成功: {resp.get('message')}")
|
|
return resp
|
|
|
|
|
|
# ---------- SDS 取单 ----------
|
|
|
|
|
|
def calc_order_weight_grams(client: Any, order: Any) -> tuple[float | None, list[str]]:
|
|
"""委托 SdsMngClient.get_order_weight_grams(children 变体克重,24h 缓存)。"""
|
|
return client.get_order_weight_grams(order)
|
|
|
|
|
|
def fetch_sds_domestic_order(client: Any | None = None) -> dict[str, str]:
|
|
"""取一单国内工厂(factoryIds=1161)订单,返回寄件/收件人字段+结构化克重。"""
|
|
if client is None:
|
|
from api import SdsMngClient
|
|
from prefect.blocks.system import Secret
|
|
|
|
value = Secret.load("sds-credentials").get()
|
|
cred = json.loads(value) if isinstance(value, str) else value
|
|
client = SdsMngClient(cred["username"], cred["password"])
|
|
from api.sds.models import OrderPageRequest
|
|
|
|
resp = client.get_orders(OrderPageRequest(factory_ids=[1161], size=1, page=1))
|
|
if not resp.records:
|
|
raise RuntimeError("SDS 未找到 factoryIds=1161 的订单")
|
|
order = resp.records[0]
|
|
addr = order.address
|
|
bay = order.issuing_bay
|
|
if not addr or not bay:
|
|
raise RuntimeError("订单缺少收件地址或发货仓信息")
|
|
|
|
send_print_addr = _join_addr(bay.province, bay.city, bay.detail)
|
|
rec_print_addr = _join_addr(addr.province, addr.city, addr.detail)
|
|
cargo = order.items[0].product_name or "日用品" if order.items else "日用品"
|
|
|
|
info = {
|
|
"order_no": order.no or order.id,
|
|
"send_name": bay.name or "印美达",
|
|
"send_addr": send_print_addr,
|
|
"rec_name": addr.consignee or "收件人",
|
|
"rec_mobile": addr.mobile_phone or "",
|
|
"rec_addr": rec_print_addr,
|
|
"cargo": cargo,
|
|
}
|
|
grams, missed = calc_order_weight_grams(client, order)
|
|
if grams:
|
|
info["weight_kg"] = f"{grams / 1000:.2f}"
|
|
print(f"[weight] 结构化克重合计 {round(grams)}g = {info['weight_kg']}kg")
|
|
else:
|
|
print(f"[weight] 未能取到结构化克重({'; '.join(missed) or '无商品行'}),下单重量回退为默认 1kg")
|
|
print(
|
|
f"[SDS] 选中订单 {info['order_no']}\n"
|
|
f" 寄件: {info['send_name']} / {info['send_addr']}\n"
|
|
f" 收件: {info['rec_name']} {info['rec_mobile']} / {info['rec_addr']}\n"
|
|
f" 物品: {info['cargo']}"
|
|
)
|
|
return info
|
|
|
|
|
|
# ---------- 业务步骤 ----------
|
|
|
|
|
|
def compare_prices(auth: dict[str, str], order_info: dict[str, str], base_url: str, weight: str) -> tuple[str, list[dict[str, Any]]]:
|
|
"""逐家查价,按 defPrice(默认总价)升序,返回 (最便宜编码, 全部结果)。"""
|
|
results: list[dict[str, Any]] = []
|
|
for com in SUPPORTED_COURIERS:
|
|
param = {
|
|
"kuaidiCom": com,
|
|
"sendManPrintAddr": order_info["send_addr"],
|
|
"recManPrintAddr": order_info["rec_addr"],
|
|
"weight": weight,
|
|
}
|
|
try:
|
|
resp = call_api("price", param, auth, base_url)
|
|
if resp.get("result") is True:
|
|
data = resp.get("data", {})
|
|
results.append({"com": com, **data})
|
|
print(
|
|
f"[price] {COURIER_NAMES.get(com, com):8s} 首重={data.get('defFirstPrice')} "
|
|
f"续重={data.get('defOverPrice')} 总价={data.get('defPrice')}"
|
|
)
|
|
else:
|
|
print(f"[price] {COURIER_NAMES.get(com, com):8s} 不支持: {resp.get('message')}")
|
|
except Exception as err: # noqa: BLE001
|
|
print(f"[price] {COURIER_NAMES.get(com, com):8s} 查询异常: {str(err)[:120]}")
|
|
time.sleep(0.3)
|
|
|
|
priced = [r for r in results if str(r.get("defPrice", "") not in ("", None)) and _to_float(r.get("defPrice")) is not None]
|
|
if not priced:
|
|
raise RuntimeError("没有任何快递公司返回有效价格")
|
|
priced.sort(key=lambda r: _to_float(r["defPrice"]))
|
|
best = priced[0]
|
|
print(
|
|
f"[price] 共 {len(priced)} 家报价,最便宜: {COURIER_NAMES.get(best['com'], best['com'])} "
|
|
f"({best['com']}) 总价 {best['defPrice']} 元"
|
|
)
|
|
return best["com"], results
|
|
|
|
|
|
def _join_addr(province: str | None, city: str | None, detail: str | None) -> str:
|
|
"""拼省+市+详细地址;detail 已含省/市名时跳过对应前缀,避免重复。"""
|
|
detail = detail or ""
|
|
prefix = ""
|
|
for part in (province or "", city or ""):
|
|
if part and part not in detail and part not in prefix:
|
|
prefix += part
|
|
return prefix + detail
|
|
|
|
|
|
def _to_float(v: Any) -> float | None:
|
|
try:
|
|
return float(v)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def place_order(
|
|
auth: dict[str, str], order_info: dict[str, str], base_url: str, kuaidicom: str, weight: str,
|
|
sender_mobile: str, callback_url: str, dry_run: bool = False,
|
|
) -> dict[str, Any]:
|
|
param = {
|
|
"kuaidicom": kuaidicom,
|
|
"recManName": order_info["rec_name"],
|
|
"recManMobile": order_info["rec_mobile"],
|
|
"recManPrintAddr": order_info["rec_addr"],
|
|
"sendManName": order_info["send_name"],
|
|
"sendManMobile": sender_mobile,
|
|
"sendManPrintAddr": order_info["send_addr"],
|
|
"callBackUrl": callback_url,
|
|
"cargo": order_info["cargo"],
|
|
"weight": weight,
|
|
"payment": "SHIPPER",
|
|
"remark": f"kuaidi100-e2e-test {order_info['order_no']}",
|
|
"salt": "inkreach-test",
|
|
"op": 0,
|
|
"resultv2": "0",
|
|
}
|
|
if kuaidicom in ("shunfeng", "shunfengkuaiyun"): # 文档要求顺丰必填预约取件时间窗(间隔≥1小时)
|
|
now = time.localtime()
|
|
start_h = (now.tm_hour + 1) % 24
|
|
param.update(
|
|
dayType="1",
|
|
pickupStartTime=f"{start_h:02d}:00",
|
|
pickupEndTime=f"{(start_h + 1) % 24:02d}:00",
|
|
)
|
|
if dry_run:
|
|
print("[bOrder dry-run] 以下为将提交的完整 param(未发送请求):")
|
|
print(json.dumps(param, ensure_ascii=False, indent=2))
|
|
return {}
|
|
resp = check(call_api("bOrder", param, auth, base_url), "bOrder 下单")
|
|
data = resp.get("data") or {}
|
|
print(
|
|
f"[bOrder] taskId={data.get('taskId')} orderId={data.get('orderId')} "
|
|
f"kuaidinum={data.get('kuaidinum')} pollToken={str(data.get('pollToken'))[:20]}"
|
|
)
|
|
return data
|
|
|
|
|
|
def query_detail(auth: dict[str, str], base_url: str, task_id: Any) -> dict[str, Any] | None:
|
|
"""订单详情查询(method=detail,只传 taskId),校验订单已正确落库。"""
|
|
try:
|
|
resp = check(call_api("detail", {"taskId": task_id}, auth, base_url), "detail 订单详情")
|
|
data = resp.get("data") or {}
|
|
print(
|
|
f"[detail] status={data.get('status')} kuaidinum={data.get('kuaidiNum')} "
|
|
f"serviceType={data.get('serviceType')} payStatus={data.get('payStatus')}"
|
|
)
|
|
return data
|
|
except Exception as err: # noqa: BLE001
|
|
print(f"[detail] 查询失败(不中断测试): {str(err)[:160]}")
|
|
return None
|
|
|
|
|
|
def cancel_order(auth: dict[str, str], base_url: str, task_id: Any, order_id: Any) -> None:
|
|
"""取消订单。taskId/orderId 原样回传;cancelMsg 用文档示例文案(实测含"测试"字样会被判订单信息错误)。"""
|
|
param = {"taskId": task_id, "orderId": order_id, "cancelMsg": "暂时不寄件了"}
|
|
check(call_api("cancel", param, auth, base_url), "cancel 取消")
|
|
|
|
|
|
def track_query(auth: dict[str, str], com: str, num: str, phone: str = "") -> dict[str, Any] | None:
|
|
"""实时轨迹查询(poll/query.do): sign=MD5(param+key+customer),需 customer 参数。"""
|
|
if not auth.get("customer"):
|
|
print("[track] 未提供 customer,跳过轨迹查询")
|
|
return None
|
|
url = base_query_url(auth)
|
|
param = json.dumps({"com": com, "num": num, "phone": phone}, ensure_ascii=False, separators=(",", ":"))
|
|
customer = auth["customer"]
|
|
sign = hashlib.md5(f"{param}{auth['key']}{customer}".encode()).hexdigest().upper()
|
|
resp = httpx.post(
|
|
url,
|
|
data={"customer": customer, "sign": sign, "param": param},
|
|
timeout=30,
|
|
).json()
|
|
print(f"[track] {json.dumps(resp, ensure_ascii=False)[:300]}")
|
|
return resp
|
|
|
|
|
|
def base_query_url(auth: dict[str, str]) -> str:
|
|
base = TEST_BASE_URL.replace("/order/borderapi.do", "")
|
|
return f"{base}/poll/query.do"
|
|
|
|
|
|
# ---------- 入口 ----------
|
|
|
|
|
|
def load_auth(args: argparse.Namespace) -> dict[str, str]:
|
|
if args.from_secret:
|
|
from prefect.blocks.system import Secret
|
|
|
|
value = Secret.load(args.secret_block).get()
|
|
auth = json.loads(value) if isinstance(value, str) else value
|
|
else:
|
|
auth = {"key": args.key or "", "secret": args.secret or "", "customer": args.customer or ""}
|
|
missing = [k for k in ("key", "secret") if not auth.get(k)]
|
|
if missing:
|
|
raise SystemExit(f"缺少快递100鉴权参数: {missing}。用 --key/--secret 或 --from-secret 提供。")
|
|
return auth
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description="快递100上门取件API端到端测试")
|
|
parser.add_argument("--key", help="企业版授权码 key")
|
|
parser.add_argument("--secret", help="签名密钥 secret")
|
|
parser.add_argument("--customer", default="", help="轨迹查询用 customer(可选)")
|
|
parser.add_argument("--from-secret", action="store_true", help="从 Prefect Secret Block 读鉴权")
|
|
parser.add_argument("--secret-block", default="kuaidi100-test", help="Secret Block 名(默认 kuaidi100-test)")
|
|
parser.add_argument("--base-url", default=TEST_BASE_URL, help=f"API 地址(默认测试环境 {TEST_BASE_URL})")
|
|
parser.add_argument("--courier", default="yuantong", help="快递编码(默认固定圆通 yuantong)")
|
|
parser.add_argument("--auto-cheapest", action="store_true", help="按比价结果自动选最便宜(默认用 --courier 固定值)")
|
|
parser.add_argument("--skip-price", action="store_true", help="跳过比价查询")
|
|
parser.add_argument("--weight", default=None, help="重量 KG;默认用 SDS 变体克重自动汇总,取不到回退 1")
|
|
parser.add_argument(
|
|
"--sender-name", default="张文风",
|
|
help="寄件人姓名(固定,覆盖 SDS 发货仓名)",
|
|
)
|
|
parser.add_argument(
|
|
"--sender-mobile", default="18320690872",
|
|
help="寄件人手机号(固定)",
|
|
)
|
|
parser.add_argument(
|
|
"--sender-addr", default="广东省广州市番禺区南村镇新基村新盈科技园3栋2楼",
|
|
help="寄件地址(固定,覆盖 SDS 发货仓地址)",
|
|
)
|
|
parser.add_argument("--dry-run", action="store_true", help="只构造并打印下单参数,不实际下单")
|
|
parser.add_argument(
|
|
"--callback-url",
|
|
default="http://8.138.172.239:18090/kuaidi100/order",
|
|
help="订单回调地址(默认走专属 frpc 实例 kuaidi100 穿透到本机回调服务)",
|
|
)
|
|
parser.add_argument("--keep-order", action="store_true", help="下单后不取消")
|
|
args = parser.parse_args()
|
|
|
|
auth = load_auth(args)
|
|
order_info = fetch_sds_domestic_order()
|
|
|
|
# 寄件人固定为广州仓联系人(命令行可覆盖),收件人保持 SDS 订单动态获取
|
|
order_info["send_name"] = args.sender_name
|
|
order_info["send_addr"] = args.sender_addr
|
|
print(f"[sender] 寄件人固定: {order_info['send_name']} {args.sender_mobile} / {order_info['send_addr']}")
|
|
|
|
weight = args.weight or order_info.get("weight_kg") or "1"
|
|
print(f"[weight] 下单申报重量: {weight}kg")
|
|
|
|
kuaidicom = args.courier
|
|
if not args.skip_price:
|
|
cheapest, _ = compare_prices(auth, order_info, args.base_url, weight) # 比价仅作参考
|
|
if args.auto_cheapest:
|
|
kuaidicom = cheapest
|
|
print(f"[price] --auto-cheapest 生效,切到最便宜: {COURIER_NAMES.get(cheapest, cheapest)} ({cheapest})")
|
|
print(f"[courier] 固定使用: {COURIER_NAMES.get(kuaidicom, kuaidicom)} ({kuaidicom})")
|
|
|
|
data = place_order(
|
|
auth, order_info, args.base_url, kuaidicom, weight,
|
|
args.sender_mobile, args.callback_url, dry_run=args.dry_run,
|
|
)
|
|
if args.dry_run:
|
|
print("[done] dry-run 结束,未下单 ✓")
|
|
return
|
|
task_id = data.get("taskId") # 保持原始类型回传,cancel 接口对类型敏感
|
|
order_id = data.get("orderId")
|
|
kuaidinum = str(data.get("kuaidinum") or "")
|
|
|
|
if kuaidinum:
|
|
track_query(auth, kuaidicom, kuaidinum, order_info["rec_mobile"])
|
|
|
|
if task_id:
|
|
query_detail(auth, args.base_url, task_id)
|
|
|
|
if not args.keep_order and task_id and order_id:
|
|
cancel_order(auth, args.base_url, task_id, order_id)
|
|
else:
|
|
print(f"[done] 保留测试单: taskId={task_id} orderId={order_id}")
|
|
|
|
print("[done] 端到端测试完成 ✓")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|