×

美客多关键词搜索接口:多站点兼容与分页漂移处理实战

Ace Ace 发表于2026-09-04 14:39:09 浏览12 评论0

抢沙发发表评论

前言

美客多作为拉美主流跨境平台,很多对接示例只做单站点简单请求,忽略多站点切换、offset 分页漂移、结果重复、无效商品过滤这些生产环境问题。本文不做简单的请求演示,重点解决业务选品场景下,跨站点搜索、本地去重、限流退避、脏数据裁剪,给出可直接用于业务服务的封装代码。

前置准备

对接前需要在美客多开发者后台创建应用,获取access_token;不同国家对应不同 site_id,墨西哥 MLM、阿根廷 MLA、巴西 MLB。搜索接口最大单页返回 50 条,offset 超过 1000 后会出现结果错乱,不能单纯循环页码拉取全部数据GitHub。

点击获取key和secret

核心代码实现

python

import requests
import time

class MercadoSearchClient:
    def __init__(self, token):
        self.token = token
        self.headers = {"Authorization": f"Bearer {self.token}"}
        self.seen_item_ids = set()  # 内存去重集合

    def keyword_search(self, site_id, keyword, limit=50, max_offset=800):
        base_url = f"https://api.mercadolibre.com/sites/{site_id}/search"
        all_goods = []
        offset = 0
        while True:
            params = {"q": keyword, "limit": limit, "offset": offset}
            try:
                resp = requests.get(base_url, headers=self.headers, params=params, timeout=25)
                if resp.status_code == 429:
                    time.sleep(60) # 限流触发,休眠退避
                    continue
                resp.raise_for_status()
                res_json = resp.json()
            except Exception as e:
                print(f"请求异常:{str(e)}")
                break
            results = res_json.get("results", [])
            if not results:
                break
            for item in results:
                item_id = item.get("id")
                if not item_id or item_id in self.seen_item_ids:
                    continue
                self.seen_item_ids.add(item_id)
                # 业务字段裁剪,过滤无用原始字段
                simple_data = {
                    "item_id": item_id,
                    "title": item.get("title"),
                    "price": item.get("price"),
                    "stock": item.get("available_quantity"),
                    "sold": item.get("sold_quantity"),
                    "free_shipping": item.get("shipping",{}).get("free_shipping"),
                    "seller_id": item.get("seller",{}).get("id")
                }
                all_goods.append(simple_data)
            paging = res_json.get("paging",{})
            total = paging.get("total",0)
            offset += limit
            if offset >= total or offset >= max_offset:
                break
            time.sleep(0.35)
        return all_goods

if __name__ == "__main__":
    tk = "your_access_token_here"
    client = MercadoSearchClient(tk)
    data = client.keyword_search(site_id="MLM", keyword="bluetooth earphone")
    print(f"有效商品总数:{len(data)}")


代码逻辑解析


  1. 内置seen_item_ids集合做内存去重,解决 offset 分页漂移带来的商品重复问题,过滤重复返回条目。

  2. 设置max_offset阈值,不无限向后翻页,规避官方 offset 大于 1000 后返回异常的限制。

  3. 捕获 429 限流状态码,使用休眠退避策略,配合请求间隔,降低接口报错概率GitHub。

  4. 对返回结果做字段精简,只保留选品业务核心字段,减少内存占用,避免原始大 JSON 冗余。

  5. 支持传入不同 site_id,快速切换墨西哥、巴西、阿根廷各个站点搜索。


对接踩坑总结

第一,不要无限制增大 offset,平台会对深度分页做结果截断,超出阈值拿到的数据不可信。第二,不同站点返回字段略有差异,取值时必须使用.get()防止键不存在直接抛异常。第三,access_token 存在有效期,业务服务需要增加令牌过期自动刷新逻辑。第四,接口返回会包含下架、库存为 0 的商品,业务层需要自行二次过滤。

群贤毕至

访客