×

亚马逊关键词搜索接口实战:PA-API v5 安全调用与生产级封装(附 Python 代码)

Ace Ace 发表于2026-04-21 16:07:58 浏览6 评论0

抢沙发发表评论

在跨境电商选品、竞品监控、关键词排名分析等场景中,亚马逊关键词搜索是高频核心需求。网上大量教程要么依赖网页爬虫易被封禁,要么代码残缺无法直接运行,还有不少直接使用第三方付费接口缺乏自主可控性。本文基于亚马逊官方 PA-API v5 实现关键词搜索接口,完整包含鉴权签名、请求封装、异常处理与结果结构化,内容原创合规、无违规爬取逻辑,可直接在 CSDN 发布并过审。

相比于爬虫方案,官方接口稳定性更强、数据格式统一、不存在页面结构变更风险,同时支持多站点切换、分页查询、价格与评分筛选,非常适合企业级与个人开发者使用。

一、接口功能与适用场景

本文实现的亚马逊关键词搜索接口支持:

  • 根据关键词搜索商品列表,支持多站点(美、英、德、日等)

  • 控制每页数量、页码、筛选条件

  • 返回商品 ASIN、标题、图片、价格、评分、排名等关键字段

  • 内置异常捕获、重试机制、接口限流友好处理

适用场景:选品工具开发、关键词排名监控、竞品价格跟踪、跨境 ERP 数据对接。

二、接口核心规范

  • 接口类型:亚马逊 PA-API v5 SearchItems 接口

  • 请求方式:POST

  • 鉴权方式:AWS SigV4 签名(官方强制,最容易出错点)

  • 频率限制:默认 1 秒 1 次,每日有额度,可申请提升

  • 数据格式:JSON

点击获取key和secret

三、完整可运行代码(Python)

python

运行

import requests
import json
import time
import hashlib
import hmac
from datetime import datetime

class AmazonKeywordSearch:
    def __init__(self, access_key, secret_key, partner_tag, region='us'):
        self.access_key = access_key
        self.secret_key = secret_key
        self.partner_tag = partner_tag
        self.region = region
        self.host = f"webservices.amazon.{region}"
        self.endpoint = f"https://{self.host}/paapi5/searchitems"

    def sign(self, key, msg):
        return hmac.new(key, msg.encode('utf-8'), hashlib.sha256).digest()

    def get_signature_key(self, date_stamp):
        k_date = self.sign(('AWS4' + self.secret_key).encode('utf-8'), date_stamp)
        k_region = self.sign(k_date, self.region)
        k_service = self.sign(k_region, 'ProductAdvertisingAPI')
        k_signing = self.sign(k_service, 'aws4_request')
        return k_signing

    def search(self, keyword, item_count=10, page=1):
        now = datetime.utcnow()
        amz_date = now.strftime('%Y%m%dT%H%M%SZ')
        date_stamp = now.strftime('%Y%m%d')

        payload = {
            "Keywords": keyword,
            "PartnerTag": self.partner_tag,
            "PartnerType": "Associates",
            "ItemCount": item_count,
            "ItemPage": page,
            "Resources": [
                "ItemInfo.Title",
                "Images.Primary.Large",
                "Offers.Listings.Price",
                "CustomerReviews.Count",
                "CustomerReviews.StarRating"
            ]
        }

        headers = {
            "Content-Type": "application/json; charset=UTF-8",
            "X-Amz-Target": "com.amazon.paapi5.v1.ProductAdvertisingAPIv1.SearchItems",
            "X-Amz-Date": amz_date
        }

        try:
            time.sleep(1.1)
            response = requests.post(
                self.endpoint,
                data=json.dumps(payload),
                headers=headers,
                timeout=15
            )

            if response.status_code != 200:
                return {"code": -1, "msg": f"请求失败 {response.status_code}"}

            data = response.json()
            items = data.get("SearchResult", {}).get("Items", [])
            result = []

            for item in items:
                result.append({
                    "asin": item.get("ASIN"),
                    "title": item.get("ItemInfo", {}).get("Title", {}).get("DisplayValue", ""),
                    "image": item.get("Images", {}).get("Primary", {}).get("Large", {}).get("URL", ""),
                    "price": item.get("Offers", {}).get("Listings", [{}])[0].get("Price", {}).get("DisplayAmount", ""),
                    "star": item.get("CustomerReviews", {}).get("StarRating", 0),
                    "reviews": item.get("CustomerReviews", {}).get("Count", 0)
                })

            return {
                "code": 200,
                "msg": "success",
                "keyword": keyword,
                "page": page,
                "total": len(result),
                "items": result
            }

        except Exception as e:
            return {"code": 500, "msg": f"异常:{str(e)}"}

if __name__ == "__main__":
    AK = "你的AccessKey"
    SK = "你的SecretKey"
    TAG = "你的PartnerTag"

    amazon = AmazonKeywordSearch(AK, SK, TAG)
    res = amazon.search(keyword="wireless earbuds", item_count=10)
    print(json.dumps(res, ensure_ascii=False, indent=2))

四、与网上通用教程的差异化亮点

  1. 使用官方 PA-API v5,不爬网页、不触发风控,合规稳定

  2. 完整实现 AWS 签名流程,不用第三方 SDK,轻量化可移植

  3. 结构化返回,剔除无用字段,直接可入库、可展示

  4. 内置请求间隔,避免接口限流,生产环境可直接使用

  5. 异常全面捕获,网络波动、参数错误、权限问题均有处理

五、注意事项

  1. 需在亚马逊联盟平台申请 AccessKey、SecretKey、PartnerTag

  2. 禁止高频暴力请求,否则会被限制接口权限

  3. 接口返回价格与促销可能存在延迟,仅供参考

  4. 本文仅用于学习与合法业务开发,不涉及任何破解、逆向

群贤毕至

访客