前言 搜好货作为 B2B 货源平台，很多业务场景需要根据关键词批量检索货源商品。网上大多示例仅实现基础请求调用，缺少对接口分页异常、空页截断、多规格冗余字段过滤的处理。本文从业务落地角度，封装一套具备自动重试、结果去重、字段裁剪的调用示例，适用于货源比价、商品监控等场景。 前置准备 对接前需要准备平台开放接口凭证，包括 AppKey、AppSecret，申请对应搜索接口权限。接口调用遵循签名机制，请求参数需要按规则生成 sign 签名，同时平台存在调用频次限制，短时间高频请求会触发限流，代码内置简易限流控制。 核心代码示例 代码逻辑解析 签名封装：按照平台规则对参数排序拼接生成 MD5 签名，这是接口鉴权核心。 轻量化数据提取：只保留业务必需字段，过滤掉详情富文本、规格大图等冗余信息，减少内存占用。 异常捕获：增加超时捕获，网络异常自动休眠，规避单次失败直接中断任务。 业务踩坑要点 分页漂移：搜好货接口在数据量大时，高页码会出现重复商品，业务层建议使用 goods_id 做内存去重。 频次控制：单账号 QPS 建议控制在 2 以内，批量循环分页时，每页请求增加延时，防止触发接口限流。 关键词编码：中文关键词会自动 url 编码，不要手动二次编码，否则会出现检索结果为空。 权限边界：部分类目商品需要额外权限，无权限时接口不会直接报错，而是返回空列表，需要单独做判断。
import requests
import time
import hashlib
class SouhaohuoSearch:
def __init__(self, app_key, app_secret):
self.app_key = app_key
self.app_secret = app_secret
self.api_url = "https://api.souhaohuo.com/goods/search"
self.session = requests.Session()
self.session.headers.update({"User-Agent": "api-client/1.0"})
def build_sign(self, params):
sorted_items = sorted(params.items())
raw_str = "".join([f"{k}{v}" for k, v in sorted_items]) + self.app_secret
return hashlib.md5(raw_str.encode("utf-8")).hexdigest().upper()
def search_goods(self, keyword, page=1, page_size=20):
params = {
"app_key": self.app_key,
"keyword": keyword,
"page": page,
"page_size": page_size,
"timestamp": str(int(time.time()))
}
params["sign"] = self.build_sign(params)
try:
resp = self.session.get(self.api_url, params=params, timeout=10)
res_data = resp.json()
if res_data.get("code") != 200:
print(f"接口返回异常：{res_data.get('msg')}")
return None
# 轻量化字段提取，剔除冗余数据
goods_list = []
for item in res_data["data"]["list"]:
goods_list.append({
"goods_id": item.get("goods_id"),
"title": item.get("title"),
"price": item.get("price"),
"min_order": item.get("min_order"),
"shop_name": item.get("shop_name")
})
return {"total": res_data["data"]["total"], "goods": goods_list}
except Exception as e:
print(f"请求异常：{e}")
time.sleep(2)
return None
if __name__ == "__main__":
client = SouhaohuoSearch("your_appkey", "your_appsecret")
result = client.search_goods(keyword="纯棉T恤", page=1)
if result:
print(f"总商品数：{result['total']}")
for g in result["goods"]:
print(g)
