当当网（Dangdang.com）作为国内知名的综合性电商平台，其开放平台（当当开发者平台）提供了一系列 API 接口，覆盖商品查询、订单管理、库存查询、促销活动等核心电商功能，主要服务于第三方开发者、企业客户及合作伙伴。以下从接口体系、认证机制、核心功能展开分析，并提供 Python 调用实现（以商品搜索和详情查询为例）。
一、当当 API 核心特性分析
1. 接口体系与功能域
2. 认证与签名机制
3. 核心接口参数与响应示例
二、Python 脚本实现
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字详情
