前言 义乌购作为小商品批发平台,和零售电商搜索接口最大差异在于起订量、档口门店、多阶梯报价这三类批发专属字段。常规搜索接口 demo 大多只获取标题与售价,忽略批发业务字段解析,直接使用会导致采购比价系统数据失真。本文从批发业务视角,封装关键词商品列表请求,增加阶梯价格解析、最小起订过滤、重复商品降噪逻辑,适合采购选品、行情监控场景。 前置准备 调用前需要在开放平台申请 核心代码实现 代码逻辑解析 封装请求会话,复用连接减少 TCP 握手开销,设置超时防止长时间阻塞。 入参增加 提取批发特有字段:档口市场地址、最小起订量,不再只拿基础标题价格。 使用字典推导式基于商品 ID 去重,解决多店铺同款商品重复返回的问题。 踩坑要点 关键词编码:中文关键词会自动 urlencode,不要手动二次编码,否则出现检索为空。 限流策略:接口有 QPS 限制,批量分页轮询时,每页请求增加休眠,短时间高频调用会触发 429 限流。 阶梯价格:返回 price 字段为基础单价,如需多档报价,需要结合商品详情接口补充。 缓存开关:调试阶段 cache 设置 no 获取实时数据;生产环境开启缓存,降低接口消耗。key与secret密钥,接口采用 GET 请求,返回 JSON 结构。重点关注:义乌购搜索结果存在同商品多档口重复铺货,需要在业务层做数据去重;价格字段不是单一零售价,会附带起批量,必须单独解析。
import requests
import json
class YiwugoSearch:
def __init__(self, key, secret):
self.key = key
self.secret = secret
self.api_url = "https://api-gw.onebound.cn/yiwugo/item_search"
self.session = requests.Session()
def search_items(self, keyword, page=1, min_order=1):
params = {
"key": self.key,
"secret": self.secret,
"api_name": "item_search",
"q": keyword,
"page": page,
"result_type": "json",
"cache": "no"
}
resp = self.session.get(self.api_url, params=params, timeout=15)
res_data = resp.json()
item_list = []
if res_data.get("status") != 200:
print("接口请求异常:", res_data.get("msg"))
return []
# 遍历商品,过滤低于最小起订量的商品
for item in res_data["data"]["items"]:
if int(item.get("min_order", 0)) >= min_order:
goods = {
"item_id": item["num_iid"],
"title": item["title"],
"price": item["price"],
"min_order": item["min_order"],
"shop_name": item["shop"],
"market_addr": item["market_address"],
"pic_url": item["pic_url"]
}
item_list.append(goods)
# 简单去重,按商品ID去重
unique_items = list({v["item_id"]:v for v in item_list}.values())
return unique_items
if __name__ == "__main__":
# 替换为自己的密钥
client = YiwugoSearch(key="your_key", secret="your_secret")
result = client.search_items(keyword="亚克力饰品", page=1, min_order=100)
print(json.dumps(result, ensure_ascii=False, indent=2))min_order,可以按业务需求过滤起订门槛过高的商品,适配小批量采购场景。