亚马逊商品搜索接口实战：SP-API 认证 + 多站点分页 + 商品画像结构化（Python 合规版）

import requests
import hmac
import hashlib
import time
import json
from datetime import datetime, timedelta
from botocore.auth import SigV4Auth
from botocore.awsrequest import AWSRequest

class AmazonSPSearchAPI:
    """亚马逊SP-API商品搜索接口（生产级·合规封装）"""
    def __init__(self, client_id, client_secret, refresh_token, aws_access_key, aws_secret_key):
        self.client_id = client_id
        self.client_secret = client_secret
        self.refresh_token = refresh_token
        self.aws_access_key = aws_access_key
        self.aws_secret_key = aws_secret_key
        self.access_token = None
        self.token_expire = None
        # 多站点市场ID映射
        self.marketplaces = {
            "us": ("na", "ATVPDKIKX0DER"),
            "uk": ("eu", "A1F83G8C2ARO7P"),
            "de": ("eu", "A1PA6795UKMFR9"),
            "jp": ("jp", "A1VC38T7YXB528")
        }
        self._refresh_access_token()

    def _refresh_access_token(self):
        """LWA令牌自动刷新（核心：解决过期问题）"""
        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
        }
        try:
            resp = requests.post(url, data=data, timeout=10)
            token_data = resp.json()
            self.access_token = token_data.get("access_token")
            expires_in = token_data.get("expires_in", 3600)
            self.token_expire = datetime.utcnow() + timedelta(seconds=expires_in - 60)
        except Exception as e:
            raise Exception(f"令牌刷新失败：{str(e)}")

    def _sigv4_sign(self, url, method, headers, payload):
        """SigV4标准签名（亚马逊官方要求）"""
        service = "sellingpartnerapi"
        region = url.split("-")[-1].split(".")[0]
        request = AWSRequest(method=method, url=url, headers=headers, data=payload)
        SigV4Auth(
            credentials={"access_key": self.aws_access_key, "secret_key": self.aws_secret_key},
            service_name=service,
            region_name=region
        ).add_auth(request)
        return dict(request.headers)

    def search_items(self, keywords, site="us", page=1, page_size=20):
        """
        商品搜索：支持多站点、分页、结构化解析
        :param keywords: 搜索关键词
        :param site: 站点（us/uk/de/jp）
        :param page: 页码
        :param page_size: 每页数量（最大50）
        :return: 结构化商品数据
        """
        # 令牌过期检查
        if datetime.utcnow() >= self.token_expire:
            self._refresh_access_token()

        if site not in self.marketplaces:
            return {"code": -1, "msg": "不支持的站点"}
        region, marketplace_id = self.marketplaces[site]
        url = f"https://sellingpartnerapi-{region}.amazon.com/catalog/2022-04-01/items"

        params = {
            "keywords": keywords,
            "marketplaceIds": marketplace_id,
            "includedData": ["summaries", "images", "offers", "ratings"],
            "pageSize": page_size,
            "page": page
        }
        payload = json.dumps(params)
        headers = {
            "x-amz-access-token": self.access_token,
            "Content-Type": "application/json",
            "Host": f"sellingpartnerapi-{region}.amazon.com"
        }
        signed_headers = self._sigv4_sign(url, "POST", headers, payload)

        try:
            resp = requests.post(url, headers=signed_headers, data=payload, timeout=15)
            result = resp.json()

            if resp.status_code != 200:
                return {"code": -2, "msg": f"请求失败：{result.get('message')}", "error": result}

            items = result.get("items", [])
            total = result.get("totalCount", 0)
            product_list = []

            # 结构化商品画像解析
            for item in items:
                summary = item.get("summaries", [{}])[0]
                image = item.get("images", [{}])[0].get("images", [{}])[0]
                offer = item.get("offers", [{}])[0].get("price", {})
                rating = item.get("ratings", {}).get("averageRating", 0)

                product_list.append({
                    "asin": item.get("asin"),
                    "title": summary.get("title", ""),
                    "brand": summary.get("brand", {}).get("name", ""),
                    "price": offer.get("amount", 0),
                    "currency": offer.get("currencyCode", "USD"),
                    "rating": rating,
                    "review_count": item.get("ratings", {}).get("reviewCount", 0),
                    "main_image": image.get("link", ""),
                    "category": summary.get("categories", [{}])[0].get("name", "")
                })

            time.sleep(0.5)
            return {
                "code": 200,
                "msg": "success",
                "data": {
                    "keywords": keywords,
                    "site": site,
                    "total_count": total,
                    "page": page,
                    "product_list": product_list
                }
            }
        except Exception as e:
            return {"code": 500, "msg": f"异常：{str(e)}"}

# ———— 调用示例 ————
if __name__ == "__main__":
    API = AmazonSPSearchAPI(
        client_id="你的LWA客户端ID",
        client_secret="你的LWA客户端密钥",
        refresh_token="你的LWA刷新令牌",
        aws_access_key="你的AWS访问密钥",
        aws_secret_key="你的AWS密钥"
    )
    # 搜索美国站蓝牙耳机
    res = API.search_items("wireless earbuds", site="us", page=1)
    print(json.dumps(res, ensure_ascii=False, indent=2))