一、接口核心基础信息
基础调用配置
配置项 说明 规范要求
接口地址 商品详情请求入口 固定为 api.dangdang.com/product/det…
请求方式 数据提交方式 仅支持 POST 方法
数据格式 请求与响应数据类型 支持 JSON/XML，默认推荐 JSON
适用范围 可获取的商品品类 图书、家居、3C 电子等全平台商品
超时建议 网络请求超时设置 15 秒（避免因网络波动导致请求失败）
二、OAuth 2.0 认证机制深度解析
当当开放平台采用 OAuth 2.0 认证体系，所有接口调用需先获取有效access_token，再通过签名验证确保请求合法性，核心流程分为 “token 获取 - 缓存 - 自动刷新” 三阶段。access_token 生命周期管理
•有效期：默认 2 小时（7200 秒），过期后需重新请求
•获取方式：通过client_credentials授权模式，提交partner_id（app_key）与app_secret获取
•刷新逻辑：本地维护 token 过期时间，到期前自动发起新请求，避免业务中断
•
- 缓存策略设计（Redis 集成）
为减少重复认证请求、提升效率，采用 “内存 + Redis” 多级缓存：
•内存缓存：服务运行时在内存中维护 token 状态，减少 Redis 访问频次
•Redis 缓存：分布式场景下共享 token，缓存过期时间比实际 token 早 300 秒（避免网络延迟导致的 token 失效）
•缓存失效处理：缓存读取失败时，自动降级为实时请求 token，确保业务连续性
•
三、签名生成规范与核心参数
- 必选核心参数（接口调用基础）
参数名称 类型 说明 注意事项
partner_id string 合作伙伴 ID（即平台分配的 app_key） 需在开放平台完成资质申请后获取
access_token string 认证令牌 需通过 OAuth 流程获取，过期需刷新
product_id string 当当网商品唯一编号 可从商品详情页 URL 或平台数据中提取
timestamp long 毫秒级时间戳 与平台服务器时间差需≤5 分钟
sign string 签名串 按平台规则生成，确保请求未篡改 - 签名生成 5 步流程（关键避坑点）
1.收集参数：整理所有请求参数（含access_token，不含sign本身）
2.排序参数：按参数名 ASCII 码升序排序（如access_token在partner_id之前）
3.拼接字符串：按 “key=value” 格式拼接排序后参数，用 “&” 连接（例：access_token=xxx&partner_id=yyy）
4.追加密钥：在拼接字符串末尾直接追加app_secret（无分隔符）
5.加密处理：对最终字符串进行 UTF-8 编码后，执行 MD5 加密并转为大写，结果即为sign
四、Python 实战实现（含缓存 + 日志）
- 核心类设计（高内聚低耦合）
（1）认证管理类（DangDangAuth）
负责access_token的获取、缓存与过期刷新，独立于接口调用逻辑，便于复用。
import requestsimport jsonimport loggingfrom datetime import datetime, timedeltafrom typing import Optional# 配置日志（便于问题排查）logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(module)s - %(levelname)s - %(message)s')logger = logging.getLogger('dangdang-auth')class DangDangAuth: """当当网接口认证管理类：处理access_token的获取、缓存与刷新""" def init(self, partner_id: str, app_secret: str, redis_client=None): self.partner_id = partner_id # 合作伙伴ID（app_key） self.app_secret = app_secret # 接口密钥 self.auth_url = "https://api.dangdang.com/oauth2/token" # 认证请求地址 self.access_token: Optional[str] = None # 当前有效token self.expires_at: Optional[datetime] = None # token过期时间 self.redis_client = redis_client # Redis客户端（可选，用于分布式缓存） self.token_cache_key = f"dangdang:access_token:{partner_id}" # 缓存键名 # 初始化时尝试从缓存加载token self._load_token_from_cache() def _load_token_from_cache(self) -> bool: """从Redis缓存加载token，避免重复请求""" if not self.redis_client: logger.info("未配置Redis，不加载缓存token") return False try: cached_token = self.redis_client.get(self.token_cache_key) if not cached_token: logger.info("缓存中无有效token") return False # 解析缓存的token信息 token_info = json.loads(cached_token) self.access_token = token_info.get("access_token") self.expires_at = datetime.fromtimestamp(token_info.get("expires_at")) # 校验token是否未过期 if datetime.now() < self.expires_at: logger.info(f"从缓存加载token成功，有效期至：{self.expires_at.strftime('%Y-%m-%d %H:%M:%S')}") return True else: logger.info("缓存token已过期，需重新获取") return False except Exception as e: logger.warning(f"加载缓存token失败：{str(e)}", exc_info=True) return False def _save_token_to_cache(self, access_token: str, expires_in: int) -> None: """将token保存到Redis，设置过期时间（提前300秒失效）""" if not self.redis_client: return try: # 计算实际过期时间戳（提前300秒，避免网络延迟导致失效） expire_timestamp = (datetime.now() + timedelta(seconds=expires_in - 300)).timestamp() token_info = { "access_token": access_token, "expires_at": expire_timestamp, "update_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S") } # 存入Redis并设置过期时间 self.redis_client.setex( name=self.token_cache_key, time=expires_in - 300, value=json.dumps(token_info, ensure_ascii=False) ) logger.info("token已保存至Redis缓存") except Exception as e: logger.warning(f"保存token到缓存失败：{str(e)}", exc_info=True) def get_valid_token(self) -> Optional[str]: """获取有效token：未过期则返回，过期则重新请求""" # 1. 校验当前token是否有效 if self.access_token and self.expires_at and datetime.now() < self.expires_at: return self.access_token # 2. 重新请求token try: request_params = { "grant_type": "client_credentials", # 客户端凭证模式 "client_id": self.partner_id, "client_secret": self.app_secret } # 发送POST请求获取token response = requests.post( url=self.auth_url, params=request_params, timeout=10, headers={"User-Agent": "DangDangAuth/1.0"} ) response.raise_for_status() # 捕获HTTP错误（如401、500） result = response.json() if "access_token" not in result: logger.error(f"获取token失败：{result.get('error_description', '未知错误')}") return None # 3. 更新token状态并缓存 self.access_token = result["access_token"] expires_in = result.get("expires_in", 7200) # 默认2小时有效期 self.expires_at = datetime.now() + timedelta(seconds=expires_in) self._save_token_to_cache(self.access_token, expires_in) logger.info(f"获取新token成功，有效期至：{self.expires_at.strftime('%Y-%m-%d %H:%M:%S')}") return self.access_token except Exception as e: logger.error(f"获取token异常：{str(e)}", exc_info=True) return None
（2）商品接口客户端（DangDangProductClient）
整合认证、签名、请求发送与数据解析，提供统一的商品详情获取入口。
import hashlibimport timeimport jsonimport requestsfrom typing import Dict, Optionalfrom datetime import datetimeclass DangDangProductClient: """当当网商品详情接口客户端：封装请求、签名与数据解析""" def init(self, partner_id: str, app_secret: str, redis_client=None): self.partner_id = partner_id self.app_secret = app_secret self.base_url = "https://api.dangdang.com/product/detail" # 商品详情接口地址 self.timeout = 15 # 请求超时时间（秒） # 初始化认证实例（复用token管理） self.auth = DangDangAuth(partner_id, app_secret, redis_client) def _generate_sign(self, params: Dict[str, str]) -> str: """生成签名：按平台规则确保请求完整性""" # 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并加密 sign_str += self.app_secret md5 = hashlib.md5() md5.update(sign_str.encode("utf-8")) return md5.hexdigest().upper() def get_product_detail(self, product_id: str, resp_format: str = "json") -> Optional[Dict]: """ 核心方法：获取商品详情并结构化解析 :param product_id: 当当商品ID :param resp_format: 响应格式（仅支持json，xml需额外扩展） :return: 结构化商品数据（None表示失败） """ # 1. 获取有效token（认证前置） access_token = self.auth.get_valid_token() if not access_token: logger.error("无有效认证token，终止商品详情请求") return None # 2. 构建基础请求参数 base_params = { "partner_id": self.partner_id, "access_token": access_token, "product_id": product_id, "timestamp": str(int(time.time() * 1000)), # 毫秒级时间戳 "format": resp_format.lower() } # 3. 生成签名（防篡改） base_params["sign"] = self._generate_sign(base_params) try: logger.info(f"发起商品详情请求：product_id={product_id}") # 4. 发送POST请求 response = requests.post( url=self.base_url, json=base_params, # JSON格式提交参数 timeout=self.timeout, headers={"User-Agent": "DangDangProductClient/1.0"} ) response.raise_for_status() # 5. 解析响应数据 if resp_format.lower() == "json": result = response.json() else: logger.error("暂不支持XML格式，仅支持JSON") return None # 6. 处理业务响应（status=0表示成功） if result.get("status") != 0: logger.error( f"商品详情请求失败：product_id={product_id}，" f"错误码={result.get('status')}，" f"错误信息={result.get('message', '未知错误')}" ) return None logger.info(f"商品详情请求成功：product_id={product_id}") # 7. 结构化解析原始数据 return self._parse_raw_data(result.get("data", {})) except requests.exceptions.RequestException as e: logger.error(f"商品详情请求网络异常：product_id={product_id}，异常信息={str(e)}", exc_info=True) return None except json.JSONDecodeError: logger.error(f"商品详情响应解析失败：product_id={product_id}，响应内容非JSON格式") return None def _parse_raw_data(self, raw_data: Dict) -> Dict: """将接口返回的原始数据解析为结构化格式""" if not isinstance(raw_data, Dict): return {} # 1. 基础商品信息 base_info = { "product_id": raw_data.get("product_id", ""), # 商品唯一ID "title": raw_data.get("title", ""), # 商品主标题 "sub_title": raw_data.get("sub_title", ""), # 商品副标题 "brand": raw_data.get("brand", {}).get("name", ""), # 品牌名称 "category": [cat.get("name") for cat in raw_data.get("category", []) if cat.get("name")], # 所属分类 "publish_time": raw_data.get("publish_time", ""), # 上架时间 "sales_volume": int(raw_data.get("sales_volume", 0)) # 销量 } # 2. 价格信息（含折扣） price_info = { "current_price": raw_data.get("price", {}).get("current_price", 0.0), # 当前售价 "original_price": raw_data.get("price", {}).get("original_price", 0.0),# 原价 "discount": raw_data.get("price", {}).get("discount", ""), # 折扣信息（如"8折"） "price_unit": raw_data.get("price", {}).get("unit", "") # 价格单位（如"元/本"） } # 3. 库存信息 stock_info = { "stock_count": int(raw_data.get("stock", {}).get("stock_count", 0)), # 库存数量 "stock_status": raw_data.get("stock", {}).get("status", "未知"), # 库存状态（如"有货"） "limit_buy": int(raw_data.get("stock", {}).get("limit_buy", 0)) # 限购数量（0表示不限购） } # 4. 图片信息（主图+详情图+缩略图） image_info = { "main_images": raw_data.get("images", {}).get("main", []), # 主图URL列表 "detail_images": raw_data.get("images", {}).get("detail", []), # 详情图URL列表 "thumbnail": raw_data.get("images", {}).get("thumbnail", "") # 缩略图URL } # 5. 图书特有信息（当当核心品类，单独解析） book_info = {} if raw_data.get("product_type") == "book": book_info = { "author": raw_data.get("book_info", {}).get("author", ""), # 作者 "publisher": raw_data.get("book_info", {}).get("publisher", ""), # 出版社 "publish_date": raw_data.get("book_info", {}).get("publish_date", ""), # 出版日期 "isbn": raw_data.get("book_info", {}).get("isbn", ""), # ISBN编号 "pages": int(raw_data.get("book_info", {}).get("pages", 0)), # 页数 "language": raw_data.get("book_info", {}).get("language", "") # 语言（如"中文"） } # 6. 规格信息（多规格商品，如尺寸、颜色） spec_info = [] for spec in raw_data.get("specs", []): spec_info.append({ "spec_id": spec.get("spec_id", ""), "spec_name": spec.get("spec_name", ""), "options": [ { "option_id": opt.get("option_id", ""), "option_name": opt.get("option_name", ""), "price": opt.get("price", 0.0), "stock": opt.get("stock", 0), "image": opt.get("image", "") } for opt in spec.get("options", []) ] }) # 整合所有结构化数据 return { "base_info": base_info, "price_info": price_info, "stock_info": stock_info, "image_info": image_info, "book_info": book_info, "spec_info": spec_info, "parse_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S") # 数据解析时间 }
- 调用示例（即拿即用）
import redisif name == "main": # 1. 配置基础参数（替换为自身在当当开放平台申请的资质） PARTNER_ID = "your_partner_id" # 合作伙伴ID（app_key） APP_SECRET = "your_app_secret" # 接口密钥 # 2. 初始化Redis客户端（可选，用于token缓存；无需缓存可设为None） try: redis_client = redis.Redis( host="localhost", # Redis服务地址 port=6379, # 端口 db=0, # 数据库编号 decode_responses=True, # 响应转为字符串 timeout=5 # 连接超时时间 ) redis_client.ping() # 测试连接 logger.info("Redis客户端初始化成功") except Exception as e: logger.warning(f"Redis连接失败，将不启用缓存：{str(e)}") redis_client = None # 3. 初始化商品接口客户端 product_client = DangDangProductClient( partner_id=PARTNER_ID, app_secret=APP_SECRET, redis_client=redis_client ) # 4. 获取商品详情（示例商品ID，替换为实际需要查询的ID） TARGET_PRODUCT_ID = "29383846" product_detail = product_client.get_product_detail(TARGET_PRODUCT_ID) # 5. 打印结果（实际业务中可替换为数据存储/业务处理逻辑） if product_detail: print("\n=== 商品基础信息 ===") print(json.dumps(product_detail["base_info"], ensure_ascii=False, indent=2)) print("\n=== 价格与库存信息 ===") print(f"当前售价：{product_detail['price_info']['current_price']} {product_detail['price_info']['price_unit']}") print(f"原价：{product_detail['price_info']['original_price']} {product_detail['price_info']['price_unit']}") print(f"库存状态：{product_detail['stock_info']['stock_status']}（剩余{product_detail['stock_info']['stock_count']}件）") # 若为图书，打印图书特有信息 if product_detail["book_info"]: print("\n=== 图书特有信息 ===") print(json.dumps(product_detail["book_info"], ensure_ascii=False, indent=2))
五、数据提取最佳实践（企业级优化）
- 结构化解析核心原则
•分层分类：按 “基础信息 - 价格 - 库存 - 图片 - 品类特有信息” 分层，避免数据混乱
•类型统一：将销量、库存、页数等转为 int 类型，价格转为 float 类型，确保数据一致性
•空值处理：对缺失字段设置默认值（如销量默认 0、标题默认空字符串），避免业务报错
•
- 图书品类重点字段利用
当当以图书为核心品类，解析时需重点关注以下字段，支撑图书类业务场景：
•ISBN：用于图书唯一标识，可关联图书元数据（如内容简介、作者背景）
•出版信息：出版社、出版日期可用于筛选新版 / 经典图书，辅助选品决策
•作者：可按作者分类聚合图书，构建作者专题或推荐系统
•
- 数据缓存策略（减少重复请求）
根据商品品类特性差异化设置缓存周期，平衡数据新鲜度与接口调用成本：
•图书类商品：更新频率低，建议缓存 24 小时
•3C / 家居类商品：价格 / 库存变动较频繁，建议缓存 6-12 小时
•促销商品：需实时同步价格，建议缓存 1-2 小时（或监听促销活动状态）
•
六、企业级对接避坑与优化建议
- 请求频率控制（合规核心）
•当当接口对调用频率有明确限制，建议单个partner_id的 QPS 控制在 10 以内
•批量获取商品详情时，采用 “队列 + 定时任务” 模式，避免短时间内请求量突增
•新增请求失败重试机制，重试间隔按 “1 秒→3 秒→5 秒” 阶梯递增，避免无效重试
•
- 异常处理增强（提升稳定性）
异常类型 处理方案
token 获取失败 触发告警（邮件 / 短信），人工介入排查资质
商品不存在 标记该商品 ID 为无效，短期内不再重复请求
网络超时 自动重试 2-3 次，仍失败则降级为缓存数据
签名错误 日志记录完整请求参数，排查参数排序 / 密钥正确性 - 日志与监控（问题快速定位）
•记录全链路日志：包含请求参数、响应数据、耗时、错误信息，便于追溯问题
•新增监控指标：接口成功率、平均响应时间、token 过期次数，设置阈值告警
•定期分析日志：识别高频失败的商品 ID、峰值请求时段，优化调用策略
•通过本文提供的方案，可实现当当网商品详情接口的企业级合规对接。代码设计遵循 “高内聚、低耦合” 原则，认证与接口调用逻辑分离，便于后续扩展（如新增商品列表接口、订单接口）；数据解析聚焦 “结构化 + 品类差异化”，可直接支撑比价系统、数据分析平台、导购应用等各类业务场景，为底层数据获取提供可靠保障。
欢迎各位大佬们评论互动，小编必回
