当当网(Dangdang.com)作为国内知名的综合性电商平台,其开放平台(当当开发者平台)提供了一系列 API 接口,覆盖商品查询、订单管理、库存查询、促销活动等核心电商功能,主要服务于第三方开发者、企业客户及合作伙伴。以下从接口体系、认证机制、核心功能展开分析,并提供 Python 调用实现(以商品搜索和详情查询为例)。
一、当当 API 核心特性分析
- 接口体系与功能域
当当 API 采用 RESTful 设计风格,按业务场景分为四大核心功能域:
功能域 核心接口 适用场景
商品管理 dd.goods.get(商品详情)、dd.goods.search(商品搜索)、dd.goods.category.get(分类查询) 商品信息获取、分类导航、价格对比
库存与价格 dd.stock.get(库存查询)、dd.price.get(价格查询) 实时库存监控、价格跟踪
订单处理 dd.order.create(创建订单)、dd.order.get(订单详情)、dd.order.list(订单列表) 订单同步、状态跟踪、售后处理
促销与优惠 dd.promotion.get(促销信息)、dd.coupon.get(优惠券查询) 活动信息展示、优惠券领取
接口网关地址为 https://api.dangdang.com,支持 HTTPS,响应格式为 JSON,部分接口支持 XML。 - 认证与签名机制
当当 API 采用 “appkey + 签名” 的认证方式,无需复杂的 OAuth 流程,核心步骤:
参数准备:公共参数(app_key、method、timestamp、format等)+ 业务参数;
签名生成:
按参数名 ASCII 升序排序;
拼接为key=value&key=value格式;
末尾拼接app_secret(应用密钥),通过 MD5 加密生成 32 位大写签名(sign);
请求发送:参数(含签名)通过 GET 或 POST 提交至网关,推荐使用 GET。 - 核心接口参数与响应示例
以 商品搜索(dd.goods.search)和 商品详情(dd.goods.get)为例:
接口名称 核心参数 响应核心字段
dd.goods.search keyword(关键词)、page(页码)、page_size(每页数量) total(总条数)、goods_list(商品数组):goods_id(商品 ID)、name(名称)、price(价格)
dd.goods.get goods_id(商品 ID) name(名称)、price(售价)、market_price(市场价)、stock(库存)、detail(详情)
二、Python 脚本实现
以下实现当当 API 的通用调用框架,包含签名生成、接口调用、响应解析,并示例商品搜索和详情查询功能。
import requests
import hashlib
import time
import json
import logging
from typing import Dict, Optional, List
from requests.exceptions import RequestException
配置日志
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s"
)
class DangdangAPI:
def init(self, app_key: str, app_secret: str):
"""
初始化当当API客户端
:param app_key: 应用app_key(当当开发者平台获取)
:param app_secret: 应用app_secret
"""
self.app_key = app_key
self.app_secret = app_secret
self.base_url = "https://api.dangdang.com"
self.session = requests.Session()
def _generate_sign(self, params: Dict) -> str:
"""生成签名(当当API规则)"""
# 1. 按参数名ASCII升序排序
sorted_params = sorted(params.items(), key=lambda x: x[0])
# 2. 拼接为key=value&key=value格式
sign_str = "&".join([f"{k}={v}" for k, v in sorted_params])
# 3. 拼接app_secret并MD5加密(大写)
sign_str += self.app_secret
return hashlib.md5(sign_str.encode("utf-8")).hexdigest().upper()
def _get_timestamp(self) -> str:
"""生成时间戳(格式:yyyy-MM-dd HH:mm:ss)"""
return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
def call_api(self, method: str, biz_params: Dict) -> Optional[Dict]:
"""
通用API调用方法
:param method: 接口方法名(如dd.goods.search)
:param biz_params: 业务参数
:return: 接口响应数据(业务部分)
"""
# 1. 公共参数
public_params = {
"app_key": self.app_key,
"method": method,
"timestamp": self._get_timestamp(),
"format": "json",
"v": "1.0"
}
# 2. 合并参数
all_params = {**public_params,** biz_params}
# 3. 生成签名
all_params["sign"] = self._generate_sign(all_params)
try:
# 4. 发送GET请求(当当API推荐GET)
response = self.session.get(
self.base_url,
params=all_params,
timeout=10
)
response.raise_for_status()
result = response.json()
# 5. 处理错误响应(当当API错误码在根节点)
if result.get("code") != 0:
logging.error(f"API错误:{result.get('msg')}(错误码:{result.get('code')})")
return None
# 6. 返回业务数据(不同接口数据字段可能不同)
return result.get("data", {})
except RequestException as e:
logging.error(f"请求异常:{str(e)},接口:{method}")
return None
except json.JSONDecodeError:
logging.error(f"响应解析失败:{response.text[:200]}...")
return None
def search_goods(self, keyword: str, page: int = 1, page_size: int = 20) -> Optional[Dict]:
"""
搜索商品
:param keyword: 搜索关键词(如"python编程")
:param page: 页码
:param page_size: 每页数量(最大50)
:return: 商品搜索结果(含列表和分页信息)
"""
method = "dd.goods.search"
params = {
"keyword": keyword,
"page": page,
"page_size": page_size
}
result = self.call_api(method, params)
if not result:
return None
# 格式化商品列表
formatted_goods = []
for goods in result.get("goods_list", []):
formatted_goods.append({
"goods_id": goods.get("goods_id"),
"name": goods.get("name"),
"price": float(goods.get("price", 0)),
"market_price": float(goods.get("market_price", 0)),
"author": goods.get("author"), # 图书类商品的作者
"publisher": goods.get("publisher"), # 出版社
"sales": int(goods.get("sales", 0)), # 销量
"cover_url": goods.get("cover_url"), # 封面图
"category": goods.get("category_name") # 所属分类
})
return {
"total": result.get("total", 0),
"page": page,
"page_size": page_size,
"goods_list": formatted_goods
}
def get_goods_detail(self, goods_id: str) -> Optional[Dict]:
"""
获取商品详情
:param goods_id: 商品ID(从搜索接口获取)
:return: 商品详情字典
"""
method = "dd.goods.get"
params = {"goods_id": goods_id}
result = self.call_api(method, params)
if not result:
return None
# 解析商品详情
return {
"goods_id": goods_id,
"name": result.get("name"),
"price": float(result.get("price", 0)),
"market_price": float(result.get("market_price", 0)),
"stock": int(result.get("stock", 0)), # 库存数量
"stock_status": result.get("stock_status"), # 库存状态(有货/无货)
"author": result.get("author"),
"publisher": result.get("publisher"),
"publish_date": result.get("publish_date"), # 出版日期(图书)
"isbn": result.get("isbn"), # ISBN编号(图书)
"detail": result.get("detail"), # 商品详情(HTML)
"images": result.get("images", "").split(","), # 图片URL列表
"category": result.get("category_name")
}
示例调用
if name == "main":
# 替换为实际参数(从当当开发者平台获取)
APP_KEY = "your_app_key"
APP_SECRET = "your_app_secret"
SEARCH_KEYWORD = "python编程" # 搜索关键词
# 初始化客户端
dangdang = DangdangAPI(APP_KEY, APP_SECRET)
# 1. 搜索商品
search_result = dangdang.search_goods(
keyword=SEARCH_KEYWORD,
page=1,
page_size=10
)
if search_result:
print(f"搜索关键词:{SEARCH_KEYWORD},找到{search_result['total']}个商品\n")
# 打印前3条商品
for i, goods in enumerate(search_result["goods_list"][:3], 1):
print(f"{i}. {goods['name']}")
print(f" 价格:{goods['price']}元(原价:{goods['market_price']}元)")
print(f" 作者/出版社:{goods['author']} / {goods['publisher']}")
print(f" 销量:{goods['sales']}件 | 商品ID:{goods['goods_id']}")
print("-" * 80)
# 2. 获取第一个商品的详情
if search_result and search_result["goods_list"]:
first_goods_id = search_result["goods_list"][0]["goods_id"]
detail = dangdang.get_goods_detail(first_goods_id)
if detail:
print(f"\n商品详情(ID:{first_goods_id}):")
print(f"名称:{detail['name']}")
print(f"价格:{detail['price']}元 | 库存:{detail['stock']}件({detail['stock_status']})")
print(f"ISBN:{detail['isbn']} | 出版日期:{detail['publish_date']}")
print(f"详情摘要:{detail['detail'][:200]}...") # 打印前200字详情
三、关键技术解析
- 签名生成逻辑
当当 API 的签名生成是接口调用的核心,需注意:
参数排序严格按 ASCII 升序(如app_key在method之前);
签名字符串需包含所有公共参数和业务参数,不可遗漏;
加密结果为 32 位大写 MD5 值(与京东 API 的小写不同)。 - 接口权限与限制
权限申请:需在当当开发者平台注册账号,创建应用并申请接口权限(个人 / 企业均可);
调用限制:默认 QPS 为 5(每秒 5 次),超过返回 429 Too Many Requests,需控制调用频率;
数据范围:仅返回当当自营及合作商家的商品,部分第三方卖家商品可能无法获取。 - 典型错误处理
错误码 说明 解决方案
1001 签名错误 检查参数排序、app_secret及加密逻辑
1002 app_key 无效 确认app_key是否已在开放平台激活
2001 商品 ID 不存在 检查goods_id是否为当当有效商品 ID
3001 接口权限不足 在开放平台为应用添加对应接口的权限
四、应用场景与扩展 - 典型场景
比价工具:通过商品搜索接口获取同类商品价格,实现跨平台比价;
图书信息聚合:针对当当强项的图书品类,聚合书籍详情、作者、ISBN 等信息;
库存监控:定期调用库存接口,对目标商品设置库存预警(如低于阈值时通知)。 - 扩展建议
实现订单管理功能(dd.order.create和dd.order.get),需申请订单权限;
增加促销信息查询(dd.promotion.get),获取商品优惠券和活动信息;
添加代理 IP 池和请求重试机制,应对 IP 封禁和网络波动;
结合数据库缓存热门商品信息,减少重复调用。
总结
当当 API 接口设计简洁,认证流程简单(无需 OAuth),适合快速开发轻量级应用,尤其在图书类商品数据方面有优势。Python 实现需重点处理签名生成和错误响应,结合业务场景合理控制调用频率。使用时需遵守当当开放平台规范,避免批量爬取或商业滥用导致权限受限。