亚马逊PA-API v5商品详情接口实战：签名避坑+生产级落地（附完整Python代码）

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

def generate_aws_sigv4_signature(secret_key, region, service, params):
    """生成AWS SigV4签名（PA-API v5强制要求，核心避坑点）"""
    # 1. 基础配置与时间格式
    timestamp = datetime.utcnow().strftime("%Y%m%dT%H%M%SZ")
    date_stamp = timestamp[:8]
    # 2. 拼接规范请求字符串
    sorted_params = sorted(params.items())
    query_string = "&".join([f"{k}={v}" for k, v in sorted_params])
    canonical_request = f"GET\n/paapi5/getitems\n{query_string}\nhost:webservices.amazon.com\nx-amz-date:{timestamp}\n\nhost;x-amz-date\n{e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855}"
    # 3. 计算签名密钥与最终签名
    signing_key = hmac.new(f"AWS4{secret_key}".encode(), date_stamp.encode(), hashlib.sha256).digest()
    signing_key = hmac.new(signing_key, region.encode(), hashlib.sha256).digest()
    signing_key = hmac.new(signing_key, service.encode(), hashlib.sha256).digest()
    signing_key = hmac.new(signing_key, "aws4_request".encode(), hashlib.sha256).digest()
    signature = hmac.new(signing_key, canonical_request.encode(), hashlib.sha256).hexdigest()
    return signature, timestamp

def get_amazon_product_detail(access_key, secret_key, partner_tag, asin, region="us"):
    """
    亚马逊PA-API v5商品详情获取（生产级，含签名+重试）
    :param access_key: 亚马逊开发者Access Key
    :param secret_key: 亚马逊开发者Secret Key
    :param partner_tag: 关联标签（开发者平台创建）
    :param asin: 商品ASIN编码
    :param region: 站点代码（us:美国, de:德国, jp:日本）
    :return: 结构化商品详情JSON
    """
    url = "https://webservices.amazon.com/paapi5/getitems"
    service = "ProductAdvertisingAPI"
    # 基础请求参数
    params = {
        "AccessKey": access_key,
        "PartnerTag": partner_tag,
        "PartnerType": "Associates",
        "ItemId": asin,
        "Resources": "Images.Primary.Large,ItemInfo.Title,Offers.Listings.Price"
    }
    # 生成签名与时间戳
    signature, timestamp = generate_aws_sigv4_signature(secret_key, region, service, params)
    params["Timestamp"] = timestamp
    params["Signature"] = signature

    # 重试机制（规避网络波动与临时限流）
    for attempt in range(3):
        try:
            response = requests.get(url, params=params, timeout=10)
            response.raise_for_status()
            result = response.json()

            # 标准化返回，过滤冗余字段
            if "ItemsResult" in result and "Items" in result["ItemsResult"]:
                item = result["ItemsResult"]["Items"][0]
                return {
                    "code": 200,
                    "msg": "success",
                    "data": {
                        "asin": asin,
                        "title": item["ItemInfo"]["Title"]["DisplayValue"],
                        "main_image": item["Images"]["Primary"]["Large"]["URL"],
                        "price": item["Offers"]["Listings"][0]["Price"]["DisplayValue"],
                        "region": region
                    }
                }
            else:
                return {"code": 400, "msg": f"商品信息获取失败：{result.get('Errors', [{}])[0].get('Message', '未知错误')}"}
        except Exception as e:
            if attempt == 2:
                return {"code": 500, "msg": f"请求异常：{str(e)}"}
            time.sleep(1.5)

# 调用示例
if __name__ == "__main__":
    # 替换为亚马逊开发者平台获取的真实信息
    ACCESS_KEY = "your_access_key"
    SECRET_KEY = "your_secret_key"
    PARTNER_TAG = "your_partner_tag"
    PRODUCT_ASIN = "B0CXXXK9"  # 示例ASIN
    REGION = "us"  # 美国站

    product_data = get_amazon_product_detail(ACCESS_KEY, SECRET_KEY, PARTNER_TAG, PRODUCT_ASIN, REGION)
    print(json.dumps(product_data, ensure_ascii=False, indent=2))