欢迎光临
我们一直在努力

【亚马逊API接口】Python 接入亚马逊 API 实现数据采集

亚马逊提供多种 API 接口用于数据采集,其中Product Advertising API (PA API 5.0) 是最常用的商品数据采集接口(如商品信息、价格、评论等)。本文将详细讲解如何通过 Python 接入亚马逊 PA API 5.0,包括前期准备、签名认证、请求构造和响应解析。

一、前期准备

1. 核心凭证获取

亚马逊 API 接入依赖严格的身份认证,需先获取以下凭证:

凭证名称获取途径
Access Key ID & Secret Access Key 亚马逊 IAM 控制台创建用户并分配权限
Associate Tag 亚马逊联盟账号生成(PA API 必需)
Marketplace ID 目标站点 ID(如北美:A1AM78C64UM0Y8,日本:A1VC38T7YXB528)

2. 步骤指引

  • 注册:亚马逊开放平台
  • 申请 PA API 5.0 权限,关联亚马逊联盟账号(生成 Associate Tag)
  • 进入 AWS IAM 控制台,创建用户并授予 PA API 访问权限,下载Access Key ID和Secret Access Key
  • 记录目标站点的Marketplace ID
  • 二、环境准备

    安装核心依赖库:

    bash

    运行

    pip install requests # 发送HTTP请求
    pip install python-dateutil # 时间处理(可选)

    三、核心实现:PA API 5.0 接入

    亚马逊 API 要求使用AWS Signature Version 4签名认证,以下是完整的接入代码(以商品详情查询为例)。

    1. 完整代码

    python

    运行

    import datetime
    import hashlib
    import hmac
    import requests
    from urllib.parse import quote_plus

    # ————————– 配置项(替换为自己的凭证) ————————–
    ACCESS_KEY = "你的Access Key ID"
    SECRET_KEY = "你的Secret Access Key"
    ASSOCIATE_TAG = "你的Associate Tag"
    MARKETPLACE_ID = "A1AM78C64UM0Y8" # 美国站点,根据需求替换
    ENDPOINT = "webservices.amazon.com" # 美国站点端点,其他站点参考官方文档
    REGION = "us-east-1" # 美国站点对应区域,其他站点参考官方文档
    # —————————————————————————–

    def generate_aws4_signature(
    method: str,
    host: str,
    region: str,
    service: str,
    access_key: str,
    secret_key: str,
    params: dict,
    body: str = "",
    timestamp: str = None
    ) -> tuple:
    """
    生成AWS Signature Version 4签名
    :return: (签名, 标准化请求字符串, 待签名字符串)
    """
    # 1. 生成时间戳(UTC)
    if not timestamp:
    t = datetime.datetime.utcnow()
    timestamp = t.strftime("%Y%m%dT%H%M%SZ")
    date_stamp = t.strftime("%Y%m%d")
    else:
    date_stamp = timestamp[:8]

    # 2. 构造规范请求字符串(Canonical Request)
    canonical_uri = "/"
    canonical_querystring = "&".join([
    f"{quote_plus(k)}={quote_plus(str(v))}" for k, v in sorted(params.items())
    ])
    canonical_headers = f"host:{host}\\nx-amz-date:{timestamp}\\n"
    signed_headers = "host;x-amz-date"

    # 计算请求体哈希
    payload_hash = hashlib.sha256(body.encode("utf-8")).hexdigest()

    canonical_request = (
    f"{method}\\n"
    f"{canonical_uri}\\n"
    f"{canonical_querystring}\\n"
    f"{canonical_headers}\\n"
    f"{signed_headers}\\n"
    f"{payload_hash}"
    )

    # 3. 构造待签名字符串(String to Sign)
    algorithm = "AWS4-HMAC-SHA256"
    credential_scope = f"{date_stamp}/{region}/{service}/aws4_request"
    string_to_sign = (
    f"{algorithm}\\n"
    f"{timestamp}\\n"
    f"{credential_scope}\\n"
    f"{hashlib.sha256(canonical_request.encode('utf-8')).hexdigest()}"
    )

    # 4. 计算签名
    def hmac_sha256(key: bytes, msg: str) -> bytes:
    return hmac.new(key, msg.encode("utf-8"), hashlib.sha256).digest()

    date_key = hmac_sha256(f"AWS4{secret_key}".encode("utf-8"), date_stamp)
    region_key = hmac_sha256(date_key, region)
    service_key = hmac_sha256(region_key, service)
    signing_key = hmac_sha256(service_key, "aws4_request")
    signature = hmac_sha256(signing_key, string_to_sign).hex()

    return signature, canonical_request, string_to_sign

    def get_amazon_product_by_asin(asin: str) -> dict:
    """
    根据ASIN查询商品详情(PA API 5.0 ItemLookup操作)
    :param asin: 亚马逊商品ASIN码
    :return: 商品详情字典
    """
    # 1. 构造请求参数
    params = {
    "Action": "ItemLookup",
    "Version": "2013-08-01", # PA API版本固定
    "MarketplaceId": MARKETPLACE_ID,
    "AssociateTag": ASSOCIATE_TAG,
    "ItemId": asin,
    "ResponseGroup": "ItemAttributes,Offers,Images", # 返回字段组,可扩展
    "Service": "AWSECommerceService",
    "Timestamp": datetime.datetime.utcnow().strftime("%Y%m%dT%H%M%SZ"),
    "AWSAccessKeyId": ACCESS_KEY,
    }

    # 2. 生成签名
    signature, _, _ = generate_aws4_signature(
    method="GET",
    host=ENDPOINT,
    region=REGION,
    service="AWSECommerceService",
    access_key=ACCESS_KEY,
    secret_key=SECRET_KEY,
    params=params
    )
    params["Signature"] = signature

    # 3. 发送请求
    url = f"https://{ENDPOINT}/"
    try:
    response = requests.get(url, params=params, timeout=10)
    response.raise_for_status() # 抛出HTTP错误
    return response.json() # PA API默认返回XML,需配置或解析XML(推荐JSON)
    except requests.exceptions.RequestException as e:
    print(f"请求失败:{e}")
    print(f"响应内容:{response.text if 'response' in locals() else '无'}")
    return {}

    # 测试示例
    if __name__ == "__main__":
    # 测试ASIN(亚马逊自营商品)
    test_asin = "B07VGRJDFY"
    product_data = get_amazon_product_by_asin(test_asin)
    if product_data:
    print("商品数据获取成功:")
    # 解析核心字段(根据ResponseGroup调整)
    item = product_data.get("ItemLookupResponse", {}).get("Items", {}).get("Item", {})
    if item:
    print(f"商品标题:{item.get('ItemAttributes', {}).get('Title', '无')}")
    print(f"价格:{item.get('Offers', {}).get('Offer', {}).get('OfferListing', {}).get('Price', {}).get('Amount', '无')} 美分")
    print(f"图片URL:{item.get('Images', {}).get('Primary', {}).get('Large', {}).get('URL', '无')}")
    else:
    print("未找到商品数据")

    2. 关键说明

    (1)签名机制

    亚马逊 API 强制要求 AWS V4 签名,核心步骤:

    • 生成 UTC 时间戳(必须同步,否则签名失效)
    • 构造规范请求字符串(包含 HTTP 方法、URI、参数、头部等)
    • 计算请求体哈希(GET 请求体为空,哈希为固定值)
    • 生成签名密钥并计算最终签名
    (2)请求参数扩展
    • ResponseGroup:控制返回字段,支持ItemAttributes(基础属性)、Offers(价格)、Reviews(评论)、SalesRank(销量排名)等,多个值用逗号分隔。
    • ItemSearch:替换Action为ItemSearch,可通过关键词 / 分类搜索商品,需添加Keywords/Category参数。
    (3)XML 转 JSON

    PA API 默认返回 XML,如需 JSON 格式,可在参数中添加:

    python

    运行

    params["Format"] = "JSON"

    四、常见问题与注意事项

    1. 签名错误(最常见)

    • 时间不同步:确保本地时间为 UTC,或使用 NTP 同步
    • 参数编码:所有参数需 URL 编码(代码中已用quote_plus处理)
    • 凭证错误:检查 Access Key/Secret Key 是否正确,是否关联 PA API 权限

    2. 调用限制

    • PA API 有每日调用次数限制(免费额度约 1000 次 / 天),超限会返回 429 错误
    • 建议添加请求间隔(如time.sleep(1)),避免触发反爬

    3. 地区适配

    不同站点的ENDPOINT/REGION/Marketplace ID需对应:

    站点ENDPOINTREGIONMarketplace ID
    美国 us-east-1 A1AM78C64UM0Y8
    英国 eu-west-1 A1F83G8C2ARO7P
    日本 webservices.amazon.co.jp ap-northeast-1 A1VC38T7YXB528

    4. 合规性

    • 亚马逊 API 禁止爬取数据用于商业竞争、批量抓取等违规场景
    • 必须保留 Associate Tag 关联,否则会被封禁权限
    • 数据使用需遵守《亚马逊开发者协议》

    五、进阶优化

  • 请求重试:添加重试机制(如tenacity库)处理临时网络错误
  • 异步请求:使用aiohttp替代requests,提高批量采集效率
  • 数据解析封装:封装解析函数,统一处理不同 ResponseGroup 的返回格式
  • 日志记录:记录请求 / 响应日志,便于排查问题
  • 六、替代方案(无 API 权限时)

    若无法申请 PA API 权限,可考虑:

  • 亚马逊官方卖家 API(针对卖家账号)
  • 合规的第三方亚马逊数据服务商(如 Jungle Scout、Helium 10)
  • 基于反爬策略的网页爬虫(需严格遵守 robots.txt 和亚马逊条款,风险较高)
  • 赞(0)
    未经允许不得转载:171主机测评 » 【亚马逊API接口】Python 接入亚马逊 API 实现数据采集
    分享到: 更多 (0)

    评论 抢沙发

    • 昵称 (必填)
    • 邮箱 (必填)
    • 网址