外观
Python 完整示例
此文件根据实际使用的 test.py 整理，仅把 API Key 和 Secret 替换为占位符。
支持：
- 自动获取所有分页中的
sku_code - 关联商品标题、描述、价格和库存
- 使用
sku_code + quantity下单 - 自动轮询订单直到交付完成
- 按
\n拆分fulfillment.payload - 按
order_<order_no>.txt保存，每行一条购买数据
命令
powershell
pip install -r requirements.txt
python test.py
python test.py order SKU-1 2完整源码
python
import argparse
import time
import hmac
import hashlib
import json
from pathlib import Path
import requests
# ---------- 配置区域（请修改为你的实际信息） ----------
BASE_URL = "https://cheapemail.cc" # 你的 dujiao-next 站点域名
API_KEY = "your_api_key" # 后台生成的 API Key
API_SECRET = "your_api_secret" # 后台生成的 API Secret
ORDER_DATA_DIR = Path(__file__).resolve().parent / "orders"
REQUEST_TIMEOUT_SECONDS = 30
ORDER_POLL_INTERVAL_SECONDS = 5
ORDER_COMPLETION_TIMEOUT_SECONDS = 600
# ----------------------------------------------------
def generate_signature(method, path, timestamp, body_bytes):
body_md5 = hashlib.md5(body_bytes).hexdigest()
sign_string = f"{method}\n{path}\n{timestamp}\n{body_md5}"
secret_bytes = API_SECRET.encode("utf-8")
sign_bytes = sign_string.encode("utf-8")
signature = hmac.new(secret_bytes, sign_bytes, hashlib.sha256).hexdigest()
return signature
def build_headers(method, path, body_bytes):
timestamp = str(int(time.time()))
signature = generate_signature(method, path, timestamp, body_bytes)
return {
"Dujiao-Next-Api-Key": API_KEY,
"Dujiao-Next-Timestamp": timestamp,
"Dujiao-Next-Signature": signature,
"Content-Type": "application/json",
}
def list_products(page=1, page_size=100):
path = "/api/v1/upstream/products"
url = BASE_URL + path
method = "GET"
body_bytes = b''
headers = build_headers(method, path, body_bytes)
response = requests.get(
url,
headers=headers,
params={"page": page, "page_size": page_size},
timeout=REQUEST_TIMEOUT_SECONDS,
)
return response.json()
def get_all_products():
products = []
page = 1
while True:
result = list_products(page=page)
if not result.get("ok"):
raise RuntimeError(f"获取商品列表失败: {result.get('error_message', result)}")
items = result.get("items", [])
products.extend(items)
total = result.get("total", len(products))
if not items or len(products) >= total:
return products
page += 1
def localized_text(value, language="zh-CN"):
if isinstance(value, str):
return value.strip()
if not isinstance(value, dict):
return ""
for key in (language, "zh-CN", "zh-TW", "en-US", "en"):
text = value.get(key)
if isinstance(text, str) and text.strip():
return text.strip()
return next(
(text.strip() for text in value.values() if isinstance(text, str) and text.strip()),
"",
)
def get_sku_list():
sku_list = []
for product in get_all_products():
for sku in product.get("skus", []):
if sku.get("is_active") is False or not sku.get("sku_code"):
continue
sku_list.append({
"sku_code": sku["sku_code"],
"title": localized_text(product.get("title")),
"description": localized_text(product.get("description")),
"price": sku.get("price_amount", product.get("price_amount", "")),
"stock_status": sku.get("stock_status", "unknown"),
"stock_quantity": sku.get("stock_quantity", 0),
"sku_id": sku.get("id"),
})
return sorted(sku_list, key=lambda item: item["sku_code"])
def print_sku_list():
sku_list = get_sku_list()
print(f"共获取到 {len(sku_list)} 个 SKU：")
for item in sku_list:
print(
f"{item['sku_code']} | {item['title']} | {item['description']} | "
f"价格: {item['price']} | 库存: {item['stock_status']} "
f"({item['stock_quantity']})"
)
def create_order(sku_id, quantity):
path = "/api/v1/upstream/orders"
url = BASE_URL + path
method = "POST"
payload = {"sku_id": sku_id, "quantity": quantity}
body_bytes = json.dumps(payload, separators=(',', ':')).encode('utf-8')
headers = build_headers(method, path, body_bytes)
response = requests.post(url, headers=headers, data=body_bytes, timeout=REQUEST_TIMEOUT_SECONDS)
return response.json()
def create_order_by_sku_code(sku_code, quantity):
"""根据 SKU 代码下单（自动查找 sku_id）"""
sku = next((item for item in get_sku_list() if item["sku_code"] == sku_code), None)
if sku is None:
raise RuntimeError(f"未找到 SKU 代码: {sku_code}")
if sku["stock_status"] == "out_of_stock":
raise RuntimeError(f"SKU '{sku_code}' 库存不足")
return create_order(sku_id=sku["sku_id"], quantity=quantity)
def get_order(order_id):
path = f"/api/v1/upstream/orders/{order_id}"
url = BASE_URL + path
method = "GET"
body_bytes = b''
headers = build_headers(method, path, body_bytes)
response = requests.get(url, headers=headers, timeout=REQUEST_TIMEOUT_SECONDS)
return response.json()
def wait_for_order_completion(order_id, poll_interval=ORDER_POLL_INTERVAL_SECONDS,
timeout=ORDER_COMPLETION_TIMEOUT_SECONDS):
"""轮询订单，直到交付完成或等待超时。"""
deadline = time.monotonic() + timeout
while True:
detail = get_order(order_id)
if not detail.get("ok"):
raise RuntimeError(f"查询订单失败: {detail.get('error_message', detail)}")
status = detail.get("status")
if status in {"delivered", "completed"} and detail.get("fulfillment"):
return detail
if status == "canceled":
raise RuntimeError(f"订单 {order_id} 已取消")
remaining = deadline - time.monotonic()
if remaining <= 0:
raise TimeoutError(
f"等待订单 {order_id} 完成超时，最后状态: {status or 'unknown'}"
)
time.sleep(min(poll_interval, remaining))
def save_order_data(order_detail, output_dir=ORDER_DATA_DIR):
"""将 fulfillment.payload 按每条数据一行保存为 UTF-8 TXT 文件。"""
order_no = order_detail.get("order_no")
if not order_no:
raise ValueError("订单详情缺少 order_no，无法保存")
payload = order_detail.get("fulfillment", {}).get("payload")
if not isinstance(payload, str) or not payload.strip():
raise ValueError("订单详情缺少 fulfillment.payload，无法保存")
# 同时兼容真实换行和 payload 中以文本形式出现的 \n / \r\n。
normalized_payload = (
payload.replace("\\r\\n", "\n")
.replace("\\n", "\n")
.replace("\\r", "\n")
)
items = [line.strip() for line in normalized_payload.splitlines() if line.strip()]
if not items:
raise ValueError("fulfillment.payload 中没有可保存的数据")
output_dir = Path(output_dir)
output_dir.mkdir(parents=True, exist_ok=True)
safe_order_no = "".join(
char if char.isalnum() or char in "-_" else "_" for char in str(order_no)
).strip("_")
if not safe_order_no:
raise ValueError("order_no 无法转换为有效文件名")
output_path = output_dir / f"order_{safe_order_no}.txt"
temp_path = output_path.with_suffix(".txt.tmp")
with temp_path.open("w", encoding="utf-8") as file:
file.write("\n".join(items) + "\n")
temp_path.replace(output_path)
return output_path
def run_order(sku_code, quantity):
print(f"正在使用 SKU '{sku_code}' 购买 {quantity} 件...")
try:
order_result = create_order_by_sku_code(sku_code, quantity)
print("下单响应:", json.dumps(order_result, indent=2, ensure_ascii=False))
if order_result.get("ok"):
order_id = order_result.get("order_id")
print(f"下单成功，订单ID: {order_id}")
print("正在等待订单交付完成...")
detail = wait_for_order_completion(order_id)
output_path = save_order_data(detail)
print(f"购买数据已保存到: {output_path}")
else:
print("下单失败:", order_result.get("error_message"))
except Exception as e:
print("发生错误:", str(e))
def main():
parser = argparse.ArgumentParser(description="CheapEmail API 示例")
subparsers = parser.add_subparsers(dest="command")
subparsers.add_parser("skus", help="获取全部 sku_code 和商品描述")
order_parser = subparsers.add_parser("order", help="使用 sku_code 下单")
order_parser.add_argument("sku_code", help="要购买的 SKU 代码")
order_parser.add_argument("quantity", type=int, help="购买数量")
args = parser.parse_args()
if args.command in {None, "skus"}:
print_sku_list()
elif args.command == "order":
if args.quantity < 1:
parser.error("quantity 必须大于等于 1")
run_order(args.sku_code, args.quantity)
if __name__ == "__main__":
main()