×

得力商城开放 API 关键词商品搜索:分页容错与商品类目过滤实践

Ace Ace 发表于2026-10-19 17:16:08 浏览13 评论0

抢沙发发表评论

前言

常规电商搜索接口大多只返回商品基础标题和售价,但得力办公用品场景存在特殊痛点:搜索结果混杂文具、设备、耗材,很多商品带有多规格包装,批量拉取分页时容易出现页码越界、空页返回,直接套通用电商代码会造成脏数据。本文不再单纯演示基础请求,重点解决分页自动终止、类目过滤、空数据容错,适合企业采购系统、比价平台对接。

前置准备

Python3.7 以上,安装 requests 库。前往得力开放平台注册开发者,创建应用获取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)

核心思路解析

  1. 分页自动中断逻辑:循环拉取页面,检测返回商品为空就立刻终止,避免无效请求触发限流;

  2. 类目黑名单过滤:在接口返回后直接剔除无关类目,减少下游业务数据清洗工作量;

  3. 异常捕获封装:网络超时、接口异常统一捕获,防止单次接口失败导致整个批量任务中断;

  4. HMAC 签名构建:参数按 key 排序拼接字符串生成签名,和 MD5 签名逻辑区分,适配得力开放平台校验规则。

群贤毕至

访客