亚马逊 SP-API 深度开发：关键字搜索接口的购物意图挖掘与合规竞品分析

import requests
from datetime import datetime, timedelta
class AmazonSPAPI:
    def __init__(self, client_id, client_secret, refresh_token):
        self.client_id = client_id
        self.client_secret = client_secret
        self.refresh_token = refresh_token
        self.access_token = None
        self.token_expire = datetime.min
    def _get_access_token(self):
        """合规获取访问令牌，自动刷新过期token"""
        if datetime.now() < self.token_expire:
            return self.access_token
        resp = requests.post(
            "https://api.amazon.com/auth/o2/token",
            data={
                "grant_type": "refresh_token",
                "client_id": self.client_id,
                "client_secret": self.client_secret,
                "refresh_token": self.refresh_token
            },
            timeout=10
        )
        data = resp.json()
        self.access_token = data["access_token"]
        self.token_expire = datetime.now() + timedelta(seconds=data["expires_in"] - 60)  # 预留60秒缓冲
        return self.access_token

{
  "search_result": {
    "items": [
      {
        "asin": "B015DWSQ3W",
        "title": "ECCO Terracruise Lt Waterproof Hiking Boot",
        "price": 149.99,
        "sales_rank": 128,  // 品类排名
        "match_score": 0.89,  // 与关键词匹配度
        "attributes": {"waterproof": true, "shaft_height": "6 inches"}
      }
    ],
    "intent_tags": ["outdoor", "hiking", "rainy_day"],  // COSMO算法生成的意图标签
    "related_keywords": ["mens waterproof hiking boots", "lightweight hiking boots"]
  }
}

import torch
from sentence_transformers import SentenceTransformer
# 加载语义模型（模拟亚马逊Embedding逻辑）
model = SentenceTransformer('all-MiniLM-L6-v2')
def parse_search_intent(keywords, intent_tags):
    """解析购物意图并分级"""
    # 生成关键词向量
    keyword_vec = model.encode(keywords)
    # 意图标签匹配
    intent_vecs = model.encode(intent_tags)
    similarities = torch.nn.functional.cosine_similarity(
        torch.tensor(keyword_vec).unsqueeze(0),
        torch.tensor(intent_vecs)
    )
    # 意图分级：核心意图（相似度>0.7）、关联意图（0.5-0.7）
    core_intents = [intent_tags[i] for i, sim in enumerate(similarities) if sim > 0.7]
    related_intents = [intent_tags[i] for i, sim in enumerate(similarities) if 0.5 <= sim <= 0.7]
    return {"core": core_intents, "related": related_intents}
# 示例：解析"waterproof hiking boots"意图
intent_result = parse_search_intent(
    "waterproof hiking boots",
    ["outdoor", "hiking", "rainy_day", "fashion", "running"]
)
# 返回：{"core":["outdoor","hiking","rainy_day"],"related":[]}

def get_competitor_keywords(api, competitor_asin, marketplace):
    """获取竞品高价值关键词"""
    resp = requests.get(
        "https://api.amazon.com/product-search/v1/items",
        headers={"Authorization": f"Bearer {api._get_access_token()}"},
        params={
            "asin": competitor_asin,
            "marketplace_ids": marketplace,
            "fields": "related_keywords,sales_rank"
        },
        timeout=10
    )
    data = resp.json()
    # 筛选规则：相关度>0.8且竞品排名前50的关键词
    high_value_kw = [
        kw for kw, score in zip(
            data["related_keywords"], data["keyword_relevance_scores"]
        ) if score > 0.8 and data["sales_rank"] <= 50
    ]
    return high_value_kw

┌─────────────┐  热点数据  ┌─────────────┐
│  本地缓存   │◄─────────►│  Redis缓存  │
│（内存，10s）│           │（15分钟）   │
└─────────────┘           └─────────────┘

def dynamic_rate_limit(api, request_count):
    """动态限流：根据剩余配额调整请求速度"""
    # 获取当前配额
    resp = requests.get(
        "https://api.amazon.com/usage/v1/quotas",
        headers={"Authorization": f"Bearer {api._get_access_token()}"},
        timeout=5
    )
    remaining_quota = resp.json()["product-search"]["remaining"]
    # 若剩余配额<50，降低请求频率
    if remaining_quota < 50:
        time.sleep(20)  # 延长间隔至20秒
    return request_count + 1