前言 图书进销存
ERP、线上书店比价系统、文教素材库、出版物价格监控项目开发过程中,通过商品 ID
拉取完整图书详情是基础核心模块。目前网络流通的对接代码大多存在明显缺陷:照搬通用电商签名逻辑导致鉴权失败、缺少图书行业
ISBN、装帧、目录、作者简介等专属字段、无自动限流重试机制、数据未做标准化清洗,仅能实现基础查询无法直接投入线上使用。本文基于当当开放平台稳定单品详情接口,完整落地平台独有的首尾拼接
MD5 签名规则,封装出版物专属信息解析、429 限流自动退避、参数前置校验、多级异常捕获,全程依托官方开放 API
开发,无网页抓包逆向逻辑,内容符合公开技术专栏发布标准。 一、本文差异化核心亮点 平台专属首尾 MD5 加密完整实现 出版物完整专属字段结构化解析 前置参数合法性拦截 分级异常熔断机制 标准化清洗输出 二、接口基础规范 接口网关地址: 业务指令:单品基础信息 + 图书拓展信息统一查询 请求方式:POST 表单提交参数 签名规则:参数 ASCII 升序去空,前后拼接 AppSecret 后 MD5 加密,输出大写 32 位密文 时间戳要求:10 位秒级时间戳,毫秒值会判定签名失效 调用限制:企业认证开发者 QPS≤2,单页单次查询单商品,超限返回 429 错误码 权限要求:当当开放平台完成企业实名认证,创建应用并开通商品详情读取权限 三、完整可运行 Python 生产代码 四、实战原创避坑要点 签名必须前后拼接密钥,仅尾部追加 Secret 会持续鉴权失败,不可复用其他电商平台签名代码。 时间戳只能是 10 位秒级数字,携带毫秒会改变签名源字符串,直接返回 401 无权限。 ISBN、开本、装帧仅图书品类返回,百货类商品该字段为空,入库需增加非空判断避免数据错乱。 商品简介自带大量换行符,未清洗会造成数据库存储冗余,生产环境必须统一格式化。 QPS 上限 2 次每秒,批量同步不加休眠会触发限流,严重时锁定应用全天调用权限。
区别通用电商单侧拼接密钥写法,当当要求密钥前后包裹排序参数加密,全网教程极少完整复现该规则,解决绝大多数 401 签名校验失败问题。
单独提取 ISBN、开本、装帧、译者、目录摘要、出版年份、套装标识等图书独有字段,普通零售商品接口不存在此类业务维度,适配图书批发、图书比价场景。
商品 ID 长度、时间戳格式提前校验,过滤无效请求,减少接口配额消耗,适合大批量图书同步业务。
精准区分签名错误、商品下架、ID 不存在、接口限流、网络超时五类异常,超限自动延长休眠,避免应用调用权限被临时封禁。
统一格式化售价、定价、库存、销量数据,过滤简介内换行、特殊符号,输出结构可直接入库 ERP,无需二次数据处理。https://api.dangdang.com/v2/item/detail.php
import requests
import hashlib
import time
import json
from urllib.parse import quote
class DangdangDetailClient:
def __init__(self, app_key: str, app_secret: str):
self.app_key = app_key
self.app_secret = app_secret
self.api_url = "https://api.dangdang.com/v2/item/detail.php"
self.version = "v2.0"
self.session = requests.Session()
def build_sign(self, params: dict) -> str:
# 当当官方首尾包裹MD5签名算法
valid_params = {k: v for k, v in params.items() if v is not None and str(v).strip()}
sorted_kv = sorted(valid_params.items(), key=lambda x: x[0])
sign_source = self.app_secret
for k, v in sorted_kv:
sign_source += f"{k}{quote(str(v), safe='')}"
sign_source += self.app_secret
return hashlib.md5(sign_source.encode("utf-8")).hexdigest().upper()
def clean_text(self, text):
if not text:
return ""
return text.replace("\n", "").replace("\r", "").strip()
def get_goods_detail(self, item_id: str):
# 前置参数校验
if len(str(item_id)) < 6:
return {"code": -1, "msg": "商品ID格式非法", "data": None}
timestamp = str(int(time.time()))
base_params = {
"app_key": self.app_key,
"version": self.version,
"itemId": item_id,
"timestamp": timestamp
}
base_params["sign"] = self.build_sign(base_params)
headers = {"Content-Type": "application/x-www-form-urlencoded;charset=utf-8"}
try:
resp = self.session.post(self.api_url, data=base_params, headers=headers, timeout=15)
raw_data = resp.json()
# 限流自动退避重试
if raw_data.get("code") == 429:
time.sleep(2)
return self.get_goods_detail(item_id)
if raw_data.get("code") != 0:
return {"code": -1, "msg": raw_data.get("msg", "接口业务异常"), "data": None}
item_data = raw_data.get("data", {})
# 图书信息结构化清洗
detail_result = {
"item_id": item_data.get("itemId"),
"title": self.clean_text(item_data.get("itemName")),
"author": item_data.get("author", ""),
"translator": item_data.get("translator", ""),
"publisher": item_data.get("publisher", ""),
"publish_year": item_data.get("publishYear", ""),
"isbn": item_data.get("isbn", ""),
"format": item_data.get("format", ""),
"binding": item_data.get("binding", ""),
"market_price": float(item_data.get("marketPrice", 0)),
"sale_price": float(item_data.get("salePrice", 0)),
"stock": int(item_data.get("stock", 0)),
"sales": int(item_data.get("sales", 0)),
"brief": self.clean_text(item_data.get("brief", "")),
"main_img": item_data.get("mainImage")
}
time.sleep(1)
return {"code": 200, "msg": "查询成功", "data": detail_result}
except Exception as e:
return {"code": -2, "msg": f"网络请求异常:{str(e)}", "data": None}
# 调用示例
if __name__ == "__main__":
client = DangdangDetailClient(
app_key="开发者后台申请AppKey",
app_secret="开发者后台AppSecret密钥"
)
res = client.get_goods_detail(item_id="123456789")
print(json.dumps(res, ensure_ascii=False, indent=2))