作為國內知名綜合性電商平臺，當當網商品詳情接口涵蓋圖書、家居、電子等全品類核心數據，是構建電商比價系統、商品分析平臺、第三方導購應用的關鍵支撐。本文從接口基礎配置、OAuth 2.0 認證落地、簽名生成規范，到 Python 代碼實現、數據結構化解析，再到企業級對接優化，提供全流程技術方案，幫助開發者規避認證失敗、數據混亂、請求超限等常見問題，實現高效合規對接。

一、接口核心基礎信息

1. 基礎調用配置

二、OAuth 2.0 認證機制深度解析

當當開放平臺采用 OAuth 2.0 認證體系，所有接口調用需先獲取有效access_token，再通過簽名驗證確保請求合法性，核心流程分為 “token 獲取 - 緩存 - 自動刷新” 三階段。

1. access_token 生命周期管理

?有效期：默認 2 小時（7200 秒），過期后需重新請求

?獲取方式：通過client_credentials授權模式，提交partner_id（app_key）與app_secret獲取

?刷新邏輯：本地維護 token 過期時間，到期前自動發起新請求，避免業務中斷

?

2. 緩存策略設計（Redis 集成）

為減少重復認證請求、提升效率，采用 “內存 + Redis” 多級緩存：

?內存緩存：服務運行時在內存中維護 token 狀態，減少 Redis 訪問頻次

?Redis 緩存：分布式場景下共享 token，緩存過期時間比實際 token 早 300 秒（避免網絡延遲導致的 token 失效）

?緩存失效處理：緩存讀取失敗時，自動降級為實時請求 token，確保業務連續性

?

三、簽名生成規范與核心參數

1. 必選核心參數（接口調用基礎）

2. 簽名生成 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

6.

四、Python 實戰實現（含緩存 + 日志）

1. 核心類設計（高內聚低耦合）

（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")  # 數據解析時間        }

2. 調用示例（即拿即用）

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))

五、數據提取最佳實踐（企業級優化）

1. 結構化解析核心原則

?分層分類：按 “基礎信息 - 價格 - 庫存 - 圖片 - 品類特有信息” 分層，避免數據混亂

?類型統一：將銷量、庫存、頁數等轉為 int 類型，價格轉為 float 類型，確保數據一致性

?空值處理：對缺失字段設置默認值（如銷量默認 0、標題默認空字符串），避免業務報錯

?

2. 圖書品類重點字段利用

當當以圖書為核心品類，解析時需重點關注以下字段，支撐圖書類業務場景：

?ISBN：用于圖書唯一標識，可關聯圖書元數據（如內容簡介、作者背景）

?出版信息：出版社、出版日期可用于篩選新版 / 經典圖書，輔助選品決策

?作者：可按作者分類聚合圖書，構建作者專題或推薦系統

?

3. 數據緩存策略（減少重復請求）

根據商品品類特性差異化設置緩存周期，平衡數據新鮮度與接口調用成本：

?圖書類商品：更新頻率低，建議緩存 24 小時

?3C / 家居類商品：價格 / 庫存變動較頻繁，建議緩存 6-12 小時

?促銷商品：需實時同步價格，建議緩存 1-2 小時（或監聽促銷活動狀態）

?

六、企業級對接避坑與優化建議

1. 請求頻率控制（合規核心）

?當當接口對調用頻率有明確限制，建議單個partner_id的 QPS 控制在 10 以內

?批量獲取商品詳情時，采用 “隊列 + 定時任務” 模式，避免短時間內請求量突增

?新增請求失敗重試機制，重試間隔按 “1 秒→3 秒→5 秒” 階梯遞增，避免無效重試

?

2. 異常處理增強（提升穩定性）

3. 日志與監控（問題快速定位）

?記錄全鏈路日志：包含請求參數、響應數據、耗時、錯誤信息，便于追溯問題

?新增監控指標：接口成功率、平均響應時間、token 過期次數，設置閾值告警

?定期分析日志：識別高頻失敗的商品 ID、峰值請求時段，優化調用策略

?通過本文提供的方案，可實現當當網商品詳情接口的企業級合規對接。代碼設計遵循 “高內聚、低耦合” 原則，認證與接口調用邏輯分離，便于后續擴展（如新增商品列表接口、訂單接口）；數據解析聚焦 “結構化 + 品類差異化”，可直接支撐比價系統、數據分析平臺、導購應用等各類業務場景，為底層數據獲取提供可靠保障。

歡迎各位大佬們評論互動，小編必回

審核編輯 黃宇