常规电商搜索接口大多只返回商品基础标题和售价,但得力办公用品场景存在特殊痛点:搜索结果混杂文具、设备、耗材,很多商品带有多规格包装,批量拉取分页时容易出现页码越界、空页返回,直接套通用电商代码会造成脏数据。本文不再单纯演示基础请求,重点解决分页自动终止、类目过滤、空数据容错,适合企业采购系统、比价平台对接。 Python3.7 以上,安装 requests 库。前往得力开放平台注册开发者,创建应用获取 分页自动中断逻辑:循环拉取页面,检测返回商品为空就立刻终止,避免无效请求触发限流; 类目黑名单过滤:在接口返回后直接剔除无关类目,减少下游业务数据清洗工作量; 异常捕获封装:网络超时、接口异常统一捕获,防止单次接口失败导致整个批量任务中断; HMAC 签名构建:参数按 key 排序拼接字符串生成签名,和 MD5 签名逻辑区分,适配得力开放平台校验规则。前言
前置准备
app_id和app_secret。接口鉴权采用参数排序 + HMAC-SHA256 签名,每次请求必须携带时间戳防重放攻击。
点击获取key和secret
代码实现
import requests
import time
import hmac
import hashlib
class DeliProductSearch:
def __init__(self, app_id, app_secret):
self.app_id = app_id
self.app_secret = app_secret
self.api_url = "https://open.deli-mall.com/api/product/search"
self.session = requests.Session()
# 过滤不需要的类目,按需修改
self.filter_category = {"安防设备", "考勤机具"}
def create_sign(self, params):
sorted_items = sorted(params.items(), key=lambda x: x[0])
raw_str = "".join(f"{k}{v}" for k, v in sorted_items)
return hmac.new(self.app_secret.encode(), raw_str.encode(), hashlib.sha256).hexdigest()
def fetch_page(self, keyword, page, size=20):
params = {
"app_id": self.app_id,
"keyword": keyword,
"page_num": page,
"page_size": size,
"timestamp": str(int(time.time()*1000))
}
params["sign"] = self.create_sign(params)
try:
resp = self.session.get(self.api_url, params=params, timeout=12)
res_json = resp.json()
except Exception as e:
return {"code": -1, "msg": str(e), "list": []}
return res_json
def search_all(self, keyword, max_page=5):
result_list = []
for page in range(1, max_page+1):
resp_data = self.fetch_page(keyword, page)
if resp_data.get("code") != 0:
break
item_list = resp_data.get("data", {}).get("items", [])
if not item_list:
break
for item in item_list:
cat_name = item.get("category_name", "")
if cat_name in self.filter_category:
continue
result_list.append({
"spu": item.get("spu_id"),
"title": item.get("title"),
"price": item.get("sale_price"),
"category": cat_name,
"spec": item.get("spec_info")
})
return result_list
if __name__ == "__main__":
client = DeliProductSearch(app_id="your_appid", app_secret="your_secret")
data = client.search_all(keyword="中性笔", max_page=3)
for row in data:
print(row)核心思路解析