用关键字搜索亚马逊商品列表，是选品调研、竞品监控的核心场景。网上常规技术贴多停留在“调用SearchItems API获取全量数据”，存在三大致命问题：结果冗余（无关商品混杂）、价值滞后（需后续二次筛选）、合规缺失（忽视调用限制）。本文创新性提出“关键字权重优化+商业维度前置筛选+智能限流合规”的全流程方案，基于亚马逊SP-API最新版本实现，让关键字搜商品从“数据采集”直接升级为“价值筛选”，代码可直接落地复用。
一、核心认知：SP-API关键字搜索的本质的是“商业筛选”
很多开发者误解“关键字搜商品”是“找全相关商品”，实则跨境电商的核心需求是“快速找到高价值商品”（如低竞争、高毛利、高好评）。亚马逊SearchItems API（v2020-08-26）支持丰富的筛选参数，但90%的方案只用到了“keywords”和“marketplaceIds”两个基础参数，浪费了接口的商业筛选能力。本文方案的核心差异的是：把“商业筛选逻辑”嵌入接口调用阶段，而非数据采集后，大幅提升效率。
二、差异化方案实现：3大核心模块
方案基于亚马逊SP-API的SearchItems API构建，核心包含“关键字权重优化器”“商业维度前置筛选器”“合规限流请求器”三大模块，解决精准度、价值度、稳定性三大问题。
1. 关键字权重优化器：提升搜索精准度
常规方案直接传入原始关键字（如“wireless earbuds”），易返回无关商品。本模块通过“核心词+属性词权重拆分”“同义词扩展”优化关键字组合，让搜索结果更贴合真实需求：
from typing import List, Dict import re class KeywordWeightOptimizer: """关键字权重优化器：拆分核心词/属性词，扩展同义词，提升搜索精准度""" def __init__(self): # 常见品类同义词映射（可按自身品类扩展） self.synonym_map = { "wireless earbuds": ["bluetooth earbuds", "wireless headphones"], "yoga mat": ["yoga towel", "exercise mat"], "pet bed": ["dog bed", "cat bed", "pet cushion"] } # 权重标识：核心词（^）、属性词（+），参考亚马逊搜索权重规则 self.weight_markers = {"core": "^", "attribute": "+"} def split_keyword(self, raw_keyword: str) -> Dict: """拆分核心词与属性词（支持手动标注或自动识别）""" # 手动标注示例：^wireless earbuds +noise cancelling +waterproof if self.weight_markers["core"] in raw_keyword: core_part = re.findall(rf"{self.weight_markers['core']}([^+]+)", raw_keyword)[0].strip() attr_parts = re.findall(rf"{self.weight_markers['attribute']}([^+]+)", raw_keyword) attr_parts = [attr.strip() for attr in attr_parts] else: # 自动识别（简单规则：长度较长、品类相关的为核心词） words = raw_keyword.split() core_part = " ".join(words[:2]) # 前2个词为核心词 attr_parts = words[2:] # 后续为属性词 return { "core_keyword": core_part, "attribute_keywords": attr_parts } def expand_synonyms(self, core_keyword: str) -> List[str]: """扩展核心词同义词，提升搜索覆盖面""" return self.synonym_map.get(core_keyword, [core_keyword]) def build_optimized_keywords(self, raw_keyword: str) -> List[str]: """构建优化后的关键字列表（核心词+同义词+属性词组合）""" split_result = self.split_keyword(raw_keyword) core_synonyms = self.expand_synonyms(split_result["core_keyword"]) attr_str = " ".join([f"+{attr}" for attr in split_result["attribute_keywords"]]) if split_result["attribute_keywords"] else "" optimized_keywords = [] for synonym in core_synonyms: optimized = f"{synonym} {attr_str}".strip() if optimized: optimized_keywords.append(optimized) return optimized_keywords # 示例：优化原始关键字 optimizer = KeywordWeightOptimizer() raw_keyword = "^wireless earbuds +noise cancelling +waterproof" print(optimizer.build_optimized_keywords(raw_keyword)) # 输出：['wireless earbuds +noise cancelling +waterproof', 'bluetooth earbuds +noise cancelling +waterproof', 'wireless headphones +noise cancelling +waterproof']
2. 商业维度前置筛选器：接口阶段过滤低价值商品
这是本方案的核心创新点：利用SearchItems API的“filterParameters”参数，在接口调用时直接过滤低价值商品（如低评分、高竞争、非FBA商品），减少无效数据传输。支持的商业筛选维度包括：评分、评论数、履约方式、价格区间、BSR排名等：
class BusinessFilter: """商业维度前置筛选器：构建API筛选参数，前置过滤低价值商品""" def __init__(self): # 筛选参数映射（对接SearchItems API的filterParameters字段） self.filter_map = { "min_rating": "average_rating", "min_review_count": "review_count", "fulfillment_type": "fulfillment_channel", "min_price": "price", "max_price": "price", "max_bsr": "sales_rank" } def build_filter_params(self, filter_config: Dict) -> List[Dict]: """构建筛选参数列表（适配SP-API要求格式）""" filter_params = [] for key, value in filter_config.items(): if key not in self.filter_map: continue api_filter_key = self.filter_map[key] # 构建筛选条件（支持范围、枚举值） if key == "min_rating": filter_params.append({"filterType": api_filter_key, "conditionType": "GREATER_THAN_OR_EQUAL_TO", "conditionValue": str(value)}) elif key == "min_review_count": filter_params.append({"filterType": api_filter_key, "conditionType": "GREATER_THAN_OR_EQUAL_TO", "conditionValue": str(value)}) elif key == "fulfillment_type": # 支持FBA、FBM等，多个值用逗号分隔 filter_params.append({"filterType": api_filter_key, "conditionType": "EQUALS", "conditionValue": value}) elif key == "min_price": filter_params.append({"filterType": api_filter_key, "conditionType": "GREATER_THAN_OR_EQUAL_TO", "conditionValue": str(value)}) elif key == "max_price": filter_params.append({"filterType": api_filter_key, "conditionType": "LESS_THAN_OR_EQUAL_TO", "conditionValue": str(value)}) elif key == "max_bsr": filter_params.append({"filterType": api_filter_key, "conditionType": "LESS_THAN_OR_EQUAL_TO", "conditionValue": str(value)}) return filter_params # 示例：构建筛选参数（筛选FBA、评分≥4.0、评论数≥100、价格10-50美元的商品） filter = BusinessFilter() filter_config = { "fulfillment_type": "AMAZON", # FBA履约（AMAZON=FBA，MERCHANT=FBM） "min_rating": 4.0, "min_review_count": 100, "min_price": 10.0, "max_price": 50.0, "max_bsr": 10000 # BSR排名≤10000 } print(filter.build_filter_params(filter_config)) # 输出符合SP-API要求的筛选参数列表
3. 合规限流请求器：保障调用稳定性
亚马逊SP-API对SearchItems API有严格的调用限制（默认QPS=5，日调用上限随开发者等级），常规方案易因限流导致调用失败。本模块实现“智能限流+自动重试+授权管理”，保障合规稳定调用：
from sp_api.api import SearchItems from sp_api.base import SellingApiException, Marketplaces import time import os from datetime import datetime, timedelta from dotenv import load_dotenv # 加载环境变量（避免硬编码授权信息） load_dotenv() class ComplianceRequester: """合规限流请求器：处理授权、限流、自动重试""" def __init__(self, marketplace=Marketplaces.US): self.credentials = { "refresh_token": os.getenv("AMAZON_REFRESH_TOKEN"), "lwa_app_id": os.getenv("AMAZON_CLIENT_ID"), "lwa_client_secret": os.getenv("AMAZON_CLIENT_SECRET"), "aws_access_key": os.getenv("AWS_ACCESS_KEY"), "aws_secret_key": os.getenv("AWS_SECRET_KEY") } self.marketplace = marketplace self.api = SearchItems(credentials=self.credentials, marketplace=marketplace) self.qps_limit = 5 # SP-API默认QPS限制 self.last_request_time = datetime.min self.retry_limit = 3 # 最大重试次数 self.retry_delay = 2 # 初始重试延迟（秒） def _control_rate(self): """限流控制：确保不超过QPS限制""" current_time = datetime.now() interval = 1 / self.qps_limit time_since_last = (current_time - self.last_request_time).total_seconds() if time_since_last < interval: time.sleep(interval - time_since_last) self.last_request_time = current_time def search_products(self, keywords: str, filter_params: List[Dict], page_size=10) -> Dict: """搜索商品列表：含限流、重试、参数校验""" retry_count = 0 while retry_count < self.retry_limit: try: # 限流控制 self._control_rate() # 调用SearchItems API response = self.api.search_items( keywords=keywords, filterParameters=filter_params, itemCount=page_size, includedData=["identifiers,pricing,ratings,fulfillment,salesRank"] # 只返回核心商业字段 ) return response.payload except SellingApiException as e: retry_count += 1 error_msg = str(e) # 特殊错误处理：限流（429）、授权失效（401） if "429" in error_msg: print(f"触发限流，延迟{self.retry_delay * (retry_count + 1)}秒重试...") time.sleep(self.retry_delay * (retry_count + 1)) elif "401" in error_msg: print("授权失效，重新初始化API...") self.api = SearchItems(credentials=self.credentials, marketplace=self.marketplace) time.sleep(self.retry_delay) else: print(f"搜索失败（非重试场景）：{error_msg}") break raise Exception(f"搜索失败（已耗尽重试次数）：{error_msg}") # 示例：初始化请求器 requester = ComplianceRequester(marketplace=Marketplaces.US)
点击获取key和secret
三、全流程实战：关键字搜商品+价值筛选
整合三大模块，实现从关键字优化、商业筛选到商品列表获取的全流程实战，直接输出高价值商品数据：
import json def main(): # 配置参数（按自身需求调整） RAW_KEYWORD = "^wireless earbuds +noise cancelling +waterproof" # 原始关键字 FILTER_CONFIG = { "fulfillment_type": "AMAZON", "min_rating": 4.0, "min_review_count": 100, "min_price": 10.0, "max_price": 50.0, "max_bsr": 10000 } MARKETPLACE = Marketplaces.US # 目标站点（美亚） PAGE_SIZE = 20 # 每页返回商品数 SAVE_PATH = "./amazon_keyword_product_list.json" try: # 1. 初始化核心模块 keyword_optimizer = KeywordWeightOptimizer() business_filter = BusinessFilter() requester = ComplianceRequester(marketplace=MARKETPLACE) # 2. 优化关键字 optimized_keywords = keyword_optimizer.build_optimized_keywords(RAW_KEYWORD) print(f"优化后关键字列表：{optimized_keywords}") # 3. 构建商业筛选参数 filter_params = business_filter.build_filter_params(FILTER_CONFIG) print(f"商业筛选参数：{filter_params}") # 4. 批量搜索（遍历优化后的关键字，聚合结果） all_products = [] for keyword in optimized_keywords: print(f"正在用关键字搜索：{keyword}") result = requester.search_products(keyword, filter_params, PAGE_SIZE) # 提取核心商品信息 products = result.get("Items", []) if not products: print(f"关键字{keyword}未搜索到符合条件的商品") continue # 结构化处理商品数据 structured_products = [] for product in products: structured_products.append({ "asin": product.get("Identifiers", {}).get("MarketplaceASIN", {}).get("ASIN"), "title": product.get("Summaries", [{}])[0].get("Title"), "brand": product.get("Summaries", [{}])[0].get("Brand"), "price": product.get("AttributeSets", [{}])[0].get("ListPrice", {}).get("Amount"), "currency": product.get("AttributeSets", [{}])[0].get("ListPrice", {}).get("CurrencyCode"), "avg_rating": product.get("CustomerReviews", {}).get("AverageRating"), "review_count": product.get("CustomerReviews", {}).get("ReviewCount"), "fulfillment_type": product.get("FulfillmentOptions", [{}])[0].get("FulfillmentChannel"), "bsr": product.get("AttributeSets", [{}])[0].get("SalesRank", [{}])[0].get("Rank"), "search_keyword": keyword }) all_products.extend(structured_products) # 5. 去重（同一ASIN可能被多个关键字匹配） unique_products = [] seen_asins = set() for product in all_products: asin = product.get("asin") if asin and asin not in seen_asins: seen_asins.add(asin) unique_products.append(product) # 6. 保存结果 final_result = { "search_config": { "raw_keyword": RAW_KEYWORD, "optimized_keywords": optimized_keywords, "filter_config": FILTER_CONFIG, "marketplace": MARKETPLACE.value, "search_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S") }, "product_count": len(unique_products), "products": unique_products } with open(SAVE_PATH, "w", encoding="utf-8") as f: json.dump(final_result, f, ensure_ascii=False, indent=2) print(f"搜索完成！共获取{len(unique_products)}个高价值商品，结果已保存至{SAVE_PATH}") except Exception as e: print(f"执行失败：{str(e)}") if __name__ == "__main__": main()
四、核心避坑与扩展建议
1. 避坑指南（合规+精准度）
关键字权重：亚马逊搜索中“核心词前置”“属性词用+标注”可提升精准度，避免无权重的杂乱组合；
筛选参数：不要过度筛选（如同时设置“BSR≤1000”和“评论数≥1000”），会导致无结果，建议逐步收紧；
授权管理：refresh_token有效期较长，但需定期更新，建议添加授权过期监控；
数据去重：同一ASIN可能被多个同义词匹配，必须去重避免数据冗余。
2. 扩展方向
多站点适配：新增站点参数，支持美亚、德亚、日亚等多站点搜索；
价格趋势：对接Product Pricing API，补充商品价格历史趋势，提升选品精准度；
竞争度分析：新增“价格竞争力”“BSR趋势”等指标，输出选品优先级排序。
本方案的核心价值在于“把商业筛选前置到接口调用阶段”，区别于网上“先采集全量数据再筛选”的低效方案，大幅提升选品调研效率。同时兼顾合规性与稳定性，所有代码基于官方SP-API SDK实现，避开爬虫等违规方式，适合中小团队长期落地使用。
