亚马逊 SP-API 商品详情接口实战：多维度规格解析 + 站点兼容 + 令牌续期（Python 合规版）

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

class AmazonItemDetail:
    def __init__(self, client_id, client_secret, refresh_token, ak, sk):
        self.client_id = client_id
        self.client_secret = client_secret
        self.refresh_token = refresh_token
        self.ak = ak
        self.sk = sk
        self.token = None
        self.token_expire = None
        # 主流站点映射
        self.site_map = {"us":"na","uk":"eu","de":"eu","jp":"jp"}
        self._refresh_token()

    def _refresh_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
        }
        res = requests.post(url, data=data, timeout=10).json()
        self.token = res.get("access_token")
        self.token_expire = datetime.utcnow() + timedelta(seconds=res["expires_in"]-60)

    def _sigv4_sign(self, url, headers):
        """标准SigV4签名"""
        region = url.split("-")[-1].split(".")[0]
        req = AWSRequest(method="GET", url=url, headers=headers)
        SigV4Auth({"access_key":self.ak,"secret_key":self.sk},
                  "sellingpartnerapi", region).add_auth(req)
        return dict(req.headers)

    def get_detail(self, asin, site="us"):
        """获取商品详情、变体、图文、卖点"""
        if datetime.utcnow() >= self.token_expire:
            self._refresh_token()
        if site not in self.site_map:
            return {"code":-1,"msg":"暂不支持该站点"}
        region = self.site_map[site]
        url = f"https://sellingpartnerapi-{region}.amazon.com/catalog/2022-04-01/items/{asin}"
        params = {"includedData":"summaries,images,variations,offers"}
        headers = {"x-amz-access-token":self.token,"Host":url.split("/")[2]}
        headers = self._sigv4_sign(url, headers)

        try:
            resp = requests.get(url, headers=headers, params=params, timeout=15)
            raw = resp.json()
            if resp.status_code != 200:
                return {"code":-2,"msg":raw.get("message","接口请求失败")}

            item = raw.get("items",[{}])[0]
            summary = item.get("summaries",[{}])[0]
            offer = item.get("offers",[{}])[0].get("price",{})
            variations = item.get("variations",[])
            img_list = [img.get("link") for img in item.get("images",[{}])[0].get("images",[])]

            # 结构化数据输出
            ret = {
                "asin":asin,"title":summary.get("title"),"brand":summary.get("brand",{}).get("name"),
                "price":offer.get("amount"),"currency":offer.get("currencyCode"),
                "main_images":img_list,"variations":variations
            }
            time.sleep(0.5)
            return {"code":200,"data":ret}
        except Exception as e:
            return {"code":-3,"msg":f"程序异常：{str(e)}"}

if __name__ == "__main__":
    api = AmazonItemDetail(
        client_id="你的ClientID",client_secret="你的ClientSecret",
        refresh_token="你的RefreshToken",ak="你的AWS AK",sk="你的AWS SK"
    )
    result = api.get_detail(asin="B0XXXXXXXXX", site="us")
    print(json.dumps(result, ensure_ascii=False, indent=2))