深度分析当当API接口,用Python脚本实现

简介: 当当网开放平台提供商品、库存、订单、促销等RESTful API,支持商品查询与管理,适用于比价工具、图书信息聚合等场景。采用appkey+签名认证,Python示例展示商品搜索与详情调用实现。

当当网(Dangdang.com)作为国内知名的综合性电商平台,其开放平台(当当开发者平台)提供了一系列 API 接口,覆盖商品查询、订单管理、库存查询、促销活动等核心电商功能,主要服务于第三方开发者、企业客户及合作伙伴。以下从接口体系、认证机制、核心功能展开分析,并提供 Python 调用实现(以商品搜索和详情查询为例)。
一、当当 API 核心特性分析

  1. 接口体系与功能域
    当当 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。
  2. 认证与签名机制
    当当 API 采用 “appkey + 签名” 的认证方式,无需复杂的 OAuth 流程,核心步骤:
    参数准备:公共参数(app_key、method、timestamp、format等)+ 业务参数;
    签名生成:
    按参数名 ASCII 升序排序;
    拼接为key=value&key=value格式;
    末尾拼接app_secret(应用密钥),通过 MD5 加密生成 32 位大写签名(sign);
    请求发送:参数(含签名)通过 GET 或 POST 提交至网关,推荐使用 GET。
  3. 核心接口参数与响应示例
    以 商品搜索(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字详情

三、关键技术解析

  1. 签名生成逻辑
    当当 API 的签名生成是接口调用的核心,需注意:
    参数排序严格按 ASCII 升序(如app_key在method之前);
    签名字符串需包含所有公共参数和业务参数,不可遗漏;
    加密结果为 32 位大写 MD5 值(与京东 API 的小写不同)。
  2. 接口权限与限制
    权限申请:需在当当开发者平台注册账号,创建应用并申请接口权限(个人 / 企业均可);
    调用限制:默认 QPS 为 5(每秒 5 次),超过返回 429 Too Many Requests,需控制调用频率;
    数据范围:仅返回当当自营及合作商家的商品,部分第三方卖家商品可能无法获取。
  3. 典型错误处理
    错误码 说明 解决方案
    1001 签名错误 检查参数排序、app_secret及加密逻辑
    1002 app_key 无效 确认app_key是否已在开放平台激活
    2001 商品 ID 不存在 检查goods_id是否为当当有效商品 ID
    3001 接口权限不足 在开放平台为应用添加对应接口的权限
    四、应用场景与扩展
  4. 典型场景
    比价工具:通过商品搜索接口获取同类商品价格,实现跨平台比价;
    图书信息聚合:针对当当强项的图书品类,聚合书籍详情、作者、ISBN 等信息;
    库存监控:定期调用库存接口,对目标商品设置库存预警(如低于阈值时通知)。
  5. 扩展建议
    实现订单管理功能(dd.order.create和dd.order.get),需申请订单权限;
    增加促销信息查询(dd.promotion.get),获取商品优惠券和活动信息;
    添加代理 IP 池和请求重试机制,应对 IP 封禁和网络波动;
    结合数据库缓存热门商品信息,减少重复调用。
    总结
    当当 API 接口设计简洁,认证流程简单(无需 OAuth),适合快速开发轻量级应用,尤其在图书类商品数据方面有优势。Python 实现需重点处理签名生成和错误响应,结合业务场景合理控制调用频率。使用时需遵守当当开放平台规范,避免批量爬取或商业滥用导致权限受限。
相关文章
|
22天前
|
人工智能 运维 监控
IT运维数字化转型:不是换工具,而是换思路
IT运维数字化转型:不是换工具,而是换思路
83 9
|
22天前
|
JSON 监控 API
深度分析快手API接口,用Python脚本实现
快手开放平台提供丰富API接口,覆盖内容管理、用户互动、直播运营等场景,服务于企业开发者及内容创作者。本文解析其接口体系、OAuth 2.0认证机制,并示例Python调用实现。
|
22天前
|
安全 前端开发 Java
Java包管理与访问控制权限详解
本文深入讲解Java包管理和访问控制,涵盖包的创建与使用、访问权限的四个层级,并结合实战案例分析如何设计合理的包结构和访问权限,帮助开发者提升代码的安全性与可维护性。
|
22天前
|
数据采集 Web App开发 存储
用Python的Requests+BeautifulSoup爬取微博热搜榜及话题内容
用Python的Requests+BeautifulSoup爬取微博热搜榜及话题内容
|
前端开发 API
uniapp中为什么会出现跨域问题,如何解决
uniapp中为什么会出现跨域问题,如何解决
4247 0
|
22天前
|
数据采集 JSON 监控
Python高效工作必备:20个实用脚本推荐!
Python是提升效率的终极自动化利器!本文精选20个实用脚本,覆盖文件批量处理、数据清洗转换、网络爬取、邮件通知、系统监控等高频场景,每项均附完整代码,可直接复制使用。无需深厚编程基础,用几行代码就能节省数小时手动操作,让你的工作流全面自动化,轻松成为高效能人士!
|
29天前
|
安全 数据挖掘 数据安全/隐私保护
Python条件判断:从基础到进阶的实用指南
本文通过15个实战案例,系统讲解Python条件判断的使用技巧,涵盖基础语法、多条件分支、嵌套判断、逻辑运算符、三元表达式及match-case等核心内容,帮助开发者构建清晰高效的决策逻辑,提升代码可读性与可维护性。
93 1
|
1月前
|
数据采集 边缘计算 缓存
从流量到留量:ESA 安全加速守护零售行业交易全链路
零售业正经历数字技术驱动的深度变革,电商蓬勃发展,消费持续升级。阿里云边缘云推出零售交易行业解决方案,通过分布式边缘计算、智能路由与安全防护,助力企业应对跨地域交易挑战,实现安全高效发展。
115 14
|
测试技术 Nacos 数据库
Nacos 配置中心(命名空间切换) | 学习笔记
快速学习 Nacos 配置中心(命名空间切换)