完成开发者账号认证,获取唯一Developer ID;
通过/api/v1/token接口获取临时访问令牌(有效期 1 小时,需定时刷新)。
注册并获取API Key:https://o0b.cn/lin

import requestsfrom datetime import datetime, timedeltaclass AmazonSPAPI:def __init__(self, client_id, client_secret, refresh_token):self.client_id = client_idself.client_secret = client_secretself.refresh_token = refresh_tokenself.access_token = Noneself.token_expire = datetime.mindef _get_access_token(self):"""合规获取访问令牌,自动刷新过期token"""if datetime.now() < self.token_expire:return self.access_tokenresp = 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|
请求参数 |
类型 |
说明(合规使用要点) |
|
keywords |
String |
支持多词组合,最多 20 个字符 |
|
marketplace_ids |
String |
必须指定站点(如 US、DE),不可跨站查询 |
|
item_count |
Integer |
单次请求最多返回 50 条数据 |
|
filters |
Object |
可按价格区间、评分等过滤,不可过滤竞品店铺 |
{"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 torchfrom 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
来自 “ ITPUB博客 ” ,链接:https://blog.itpub.net/70047598/viewspace-3097935/ ,如需转载,请注明出处,否则将追究法律责任。