亚马逊关键词搜索API接口全面解析与实战指南

pip install amazon-paapi5

from paapi5_python_sdk.api.default_api import DefaultApifrom paapi5_python_sdk.models.search_request import SearchRequestfrom paapi5_python_sdk.models.partner_type import PartnerTypeimport osclass AmazonKeywordSearcher:
    def __init__(self):
        self.access_key = os.getenv('AMAZON_ACCESS_KEY')
        self.secret_key = os.getenv('AMAZON_SECRET_KEY')
        self.partner_tag = os.getenv('AMAZON_PARTNER_TAG')
        self.host = "webservices.amazon.com"
        self.region = "us-east-1"
        
    def search_by_keyword(self, keyword, search_index="All", item_count=10):
        """
        关键词搜索商品
        """
        request = SearchRequest(
            partner_tag=self.partner_tag,
            partner_type=PartnerType.ASSOCIATES,
            keywords=keyword,
            search_index=search_index,
            item_count=item_count,
            sort_by="SalesRank",  # 按销量排序
            resources=[
                "Images.Primary.Medium",
                "Images.Variants.Small",
                "ItemInfo.Title",
                "ItemInfo.Features",
                "ItemInfo.ProductInfo",
                "Offers.Listings.Price",
                "Offers.Listings.SavingBasis",
                "CustomerReviews.StarRating",
                "BrowseNodeInfo.BrowseNodes"
            ]
        )
        
        api = DefaultApi(
            access_key=self.access_key,
            secret_key=self.secret_key,
            host=self.host,
            region=self.region        )
        
        try:
            response = api.search_items(request)
            return self._parse_response(response)
        except Exception as e:
            print(f"搜索失败: {str(e)}")
            return None
    
    def _parse_response(self, response):
        """解析响应数据"""
        results = []
        if response.search_result and response.search_result.items:
            for item in response.search_result.items:
                product = {
                    'asin': item.asin,
                    'title': item.item_info.title.display_value if item.item_info else None,
                    'price': item.offers.listings[0].price.display_amount if item.offers else None,
                    'image': item.images.primary.medium.url if item.images else None,
                    'detail_url': item.detail_page_url,
                    'sales_rank': item.browse_node_info.browse_nodes[0].sales_rank if item.browse_node_info else None
                }
                results.append(product)
        return results# 使用示例if __name__ == "__main__":
    searcher = AmazonKeywordSearcher()
    products = searcher.search_by_keyword("wireless headphones", "Electronics", 10)
    
    for p in products:
        print(f"ASIN: {p['asin']}, 标题: {p['title']}, 价格: {p['price']}")

{
    "search_result": {
        "items": [
            {
                "asin": "B08N5WRWNW",
                "item_info": {
                    "title": {"display_value": "Echo Dot (4th Gen)"},
                    "features": {"display_values": ["Smart speaker with Alexa", "Charcoal"]}
                },
                "offers": {
                    "listings": [{
                        "price": {"display_amount": "$29.99", "currency": "USD"},
                        "saving_basis": {"display_amount": "$49.99"}
                    }]
                },
                "customer_reviews": {
                    "star_rating": {"value": 4.7, "count": 456789}
                },
                "images": {
                    "primary": {"medium": {"url": "https://m.media-amazon.com/..."}}
                }
            }
        ],
        "total_result_count": 5000
    }}

import requestsimport boto3from botocore.auth import SigV4Authfrom botocore.awsrequest import AWSRequestclass SPAPIKeywordSearcher:
    def __init__(self, refresh_token, client_id, client_secret):
        self.refresh_token = refresh_token
        self.client_id = client_id
        self.client_secret = client_secret
        self.access_token = self._get_access_token()
        
    def _get_access_token(self):
        """获取访问令牌"""
        url = "https://api.amazon.com/auth/o2/token"
        data = {
            "grant_type": "refresh_token",
            "refresh_token": self.refresh_token,
            "client_id": self.client_id,
            "client_secret": self.client_secret        }
        response = requests.post(url, data=data)
        return response.json().get("access_token")
    
    def search_catalog_items(self, keywords, marketplace_id="ATVPDKIKX0DER"):
        """
        搜索商品目录
        """
        url = f"https://sellingpartnerapi-na.amazon.com/catalog/2020-12-01/items"
        
        headers = {
            "x-amz-access-token": self.access_token,
            "Content-Type": "application/json"
        }
        
        params = {
            "keywords": keywords,
            "marketplaceIds": marketplace_id,
            "includedData": ["summaries", "images", "salesRanks"],
            "pageSize": 20
        }
        
        response = requests.get(url, headers=headers, params=params)
        
        if response.status_code == 200:
            return self._parse_sp_api_response(response.json())
        else:
            print(f"请求失败: {response.status_code}, {response.text}")
            return None
    
    def _parse_sp_api_response(self, data):
        """解析SP-API响应（含COSMO意图标签）"""
        results = []
        for item in data.get("items", []):
            product = {
                "asin": item.get("asin"),
                "title": item.get("summaries", [{}])[0].get("itemName"),
                "brand": item.get("summaries", [{}])[0].get("brand"),
                "price": item.get("summaries", [{}])[0].get("price", {}).get("amount"),
                "sales_rank": item.get("salesRanks", [{}])[0].get("displayGroupRanks", [{}])[0].get("rank"),
                # COSMO算法生成的意图标签（如存在）
                "intent_tags": item.get("classifications", [{}])[0].get("intentTags", []),
                "related_keywords": item.get("classifications", [{}])[0].get("relatedKeywords", [])
            }
            results.append(product)
        return results

{
    "items": [{
        "asin": "B015DWSQ3W",
        "summaries": [{"itemName": "ECCO Terracruise Lt Waterproof Hiking Boot"}],
        "salesRanks": [{
            "displayGroupRanks": [{"rank": 128}]
        }],
        "classifications": [{
            "intentTags": ["outdoor", "hiking", "rainy_day"],
            "relatedKeywords": ["mens waterproof hiking boots", "lightweight hiking boots"],
            "matchScore": 0.89
        }]
    }]}

import requestsdef pangolin_keyword_search(keyword, zipcode="10041"):
    url = "https://scrapeapi.pangolinfo.com/api/v1/scrape"
    
    payload = {
        "url": f"https://www.amazon.com/s?k={keyword.replace(' ', '+')}",
        "formats": ["json"],
        "parserName": "amzKeyword",
        "bizContext": {"zipcode": zipcode}  # 指定邮编获取本地化价格
    }
    
    headers = {
        "Authorization": "Bearer YOUR_PANGOLIN_TOKEN",
        "Content-Type": "application/json"
    }
    
    response = requests.post(url, json=payload, headers=headers)
    return response.json()

import hashlibimport hmacimport datetimeimport urllib.parsedef generate_pa_api_signature(access_key, secret_key, payload, host, region, service="ProductAdvertisingAPI"):
    """
    生成AWS Signature V4
    """
    # 1. 创建规范请求
    t = datetime.datetime.utcnow()
    amz_date = t.strftime('%Y%m%dT%H%M%SZ')
    date_stamp = t.strftime('%Y%m%d')
    
    # 2. 创建待签名字符串
    credential_scope = f"{date_stamp}/{region}/{service}/aws4_request"
    string_to_sign = f"AWS4-HMAC-SHA256\n{amz_date}\n{credential_scope}\n" + \
                     hashlib.sha256(payload.encode('utf-8')).hexdigest()
    
    # 3. 计算签名密钥
    def get_signature_key(key, date_stamp, region_name, service_name):
        k_date = hmac.new(('AWS4' + key).encode('utf-8'), date_stamp.encode('utf-8'), hashlib.sha256).digest()
        k_region = hmac.new(k_date, region_name.encode('utf-8'), hashlib.sha256).digest()
        k_service = hmac.new(k_region, service_name.encode('utf-8'), hashlib.sha256).digest()
        k_signing = hmac.new(k_service, 'aws4_request'.encode('utf-8'), hashlib.sha256).digest()
        return k_signing
    
    signing_key = get_signature_key(secret_key, date_stamp, region, service)
    signature = hmac.new(signing_key, string_to_sign.encode('utf-8'), hashlib.sha256).hexdigest()
    
    # 4. 构建Authorization头
    auth_header = f"AWS4-HMAC-SHA256 Credential={access_key}/{credential_scope}, " \                  f"SignedHeaders=host;x-amz-date, Signature={signature}"
    
    return auth_header, amz_date

import timeimport randomdef retry_with_backoff(func, max_retries=3, base_delay=1):
    """指数退避重试"""
    for attempt in range(max_retries):
        try:
            return func()
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
            print(f"请求失败，{delay:.2f}秒后重试...")
            time.sleep(delay)

def optimize_search_params(base_keyword):
    """
    搜索参数优化建议
    """
    return {
        "primary": base_keyword,
        "variations": [
            f"{base_keyword} for women",  # 人群细分
            f"{base_keyword} prime",       # Prime筛选
            f"{base_keyword} best seller"  # 畅销筛选
        ],
        "filters": {
            "min_rating": 4.0,      # 评分过滤
            "max_price": 100.00,    # 价格上限
            "availability": "Available"  # 有货筛选
        }
    }