1688平台商品数据采集：API调用与批量分析

在B2B电商场景下，1688平台汇聚了数百万级SKU的批发商品信息。对于供应链选品、价格监控、竞品调研等业务需求，手动采集页面数据效率极低且易触发反爬。本文从技术实现角度出发，详细介绍基于1688开放平台API的商品详情与关键词搜索接口的调用方法，并提供完整的Python工程化代码，涵盖签名构造、分页迭代、异常重试及数据落库等环节。

一、API调用前的准备：应用凭证与权限

1688开放平台采用OAuth 2.0授权体系，但针对商品查询类接口（只读），通常使用AppKey + AppSecret的签名认证方式即可。开发者需完成以下步骤：

1. 登录1688开放平台（需企业实名认证）。

2. 创建应用，获取App Key和App Secret。

3. 在应用管理后台订阅“商品信息查询”与“搜索服务”相关API权限。

4. 注意：部分接口需绑定IP白名单，生产环境务必配置。

二、商品详情API：单SKU深度信息获取

2.1 接口说明

• 请求地址（网关）:https://api.1688.com/param2/1/com.alibaba.product/...（具体路径以官方文档为准，因平台版本更新，本文使用模拟通用REST风格）

• 请求方法: GET

• 必填参数:productId（商品ID）

• 返回字段: 商品标题、一级/二级类目、属性集合（规格、颜色、尺寸）、价格区间（起批量对应阶梯价）、库存总量、详情描述图片URL、店铺信息等。

2.2 签名机制（重点）

1688 API采用签名算法防止请求被篡改。通用规则如下：

• 将请求参数（含公共参数app_key,timestamp,sign_method,format,v等）按key升序排列，拼接成字符串str1。

• 在str1前后加上App Secret，计算MD5或HMAC-SHA256得到sign值。

Python签名示例：

import hashlib import time import urllib.parse def generate_sign(params, secret, method='md5'): """生成1688风格签名（简化版）""" sorted_keys = sorted(params.keys()) sign_str = secret for k in sorted_keys: sign_str += k + str(params[k]) sign_str += secret if method == 'md5': return hashlib.md5(sign_str.encode('utf-8')).hexdigest().upper() else: # HMAC-SHA256 可按需实现 pass

实际调用时，需将sign加入请求参数中。

2.3 商品详情接口调用完整代码

以下代码封装了签名、发送请求、解析响应的通用函数，并添加了重试机制与超时控制。

import requests import time import hashlib import json from typing import Dict, Optional class Alibaba1688Client: def __init__(self, app_key: str, app_secret: str, gateway: str = "https://api.1688.com"): self.app_key = app_key self.app_secret = app_secret self.gateway = gateway self.session = requests.Session() self.session.timeout = (5, 30) # 连接超时5s，读取超时30s def _build_common_params(self) -> Dict: """构造公共参数""" return { 'app_key': self.app_key, 'timestamp': str(int(time.time() * 1000)), # 毫秒时间戳 'format': 'json', 'v': '2.0', 'sign_method': 'md5' } def _sign(self, params: Dict) -> str: sorted_keys = sorted(params.keys()) raw = self.app_secret for k in sorted_keys: raw += k + str(params[k]) raw += self.app_secret return hashlib.md5(raw.encode('utf-8')).hexdigest().upper() def _request(self, api_path: str, biz_params: Dict) -> Dict: """ 通用请求方法 :param api_path: API路径，如 '/com.alibaba.product/detail' :param biz_params: 业务参数dict """ full_url = self.gateway + api_path params = self._build_common_params() params.update(biz_params) params['sign'] = self._sign(params) # 重试逻辑（最多3次，指数退避） for attempt in range(3): try: resp = self.session.get(full_url, params=params, timeout=30) resp.raise_for_status() result = resp.json() # 检查业务错误码（1688返回中通常包含 errorCode 字段） if result.get('errorCode'): raise Exception(f"API业务错误: {result.get('errorMessage')}") return result except (requests.exceptions.RequestException, Exception) as e: if attempt == 2: raise time.sleep(2 ** attempt) return {} def get_product_detail(self, product_id: str) -> Optional[Dict]: """获取商品详情""" api_path = "/param2/1/com.alibaba.product/detail" # 示例路径，实际请查阅最新文档 biz = {'productId': product_id} return self._request(api_path, biz)

2.4 返回数据清洗要点

• 价格字段通常为字符串区间（如"10.5-20.0"），需正则提取最低价与最高价。

• 库存字段可能包含多个子仓库库存，需汇总。

• 属性attributes为JSON数组，可展开为扁平化字段便于入库。

三、关键词搜索API：批量商品列表采集

3.1 接口参数详解

搜索接口常驻参数：

3.2 分页全量爬取策略

由于平台单页最大100条，且总商品数可能超过10000，需循环请求直到返回空列表。同时需控制请求频率（建议间隔200ms以上），防止限流。

def search_products(client: Alibaba1688Client, keyword: str, max_pages: int = 50, page_size: int = 100): """分页搜索商品，返回生成器""" for page in range(1, max_pages + 1): biz = { 'keywords': keyword, 'pageNo': page, 'pageSize': page_size, 'sortType': 'sale_desc' } resp = client._request("/param2/1/com.alibaba.product/search", biz) items = resp.get('data', {}).get('productList', []) if not items: break yield items time.sleep(0.3) # 避免被限流

3.3 搜索结果的数据结构扁平化

返回的productList每条记录包含productId,title,price,saleCount（月销量）,imageUrl,detailUrl,storeName等。可将其映射为关系型表结构，例如：

CREATE TABLE ods_1688_search_result ( product_id VARCHAR(32) PRIMARY KEY, title VARCHAR(512), price_min DECIMAL(12,2), price_max DECIMAL(12,2), monthly_sales INT, image_url VARCHAR(1024), detail_url VARCHAR(1024), store_name VARCHAR(128), fetch_time DATETIME DEFAULT CURRENT_TIMESTAMP );

四、性能优化与反爬规避

4.1 批量请求与异步IO

若需批量采集多个关键词，可使用asyncio+aiohttp提升效率，但需注意平台QPS上限（通常为60次/分钟）。下面给出异步限流器的简单实现：

import asyncio import aiohttp from aiohttp import ClientTimeout class AsyncRateLimiter: def __init__(self, max_calls: int, period: float): self.max_calls = max_calls self.period = period self.semaphore = asyncio.Semaphore(max_calls) self.last_reset = asyncio.get_event_loop().time() async def acquire(self): await self.semaphore.acquire() # 这里可添加时间窗口控制逻辑 # 具体异步请求代码略（可参考官方aiohttp文档）

4.2 本地缓存策略

对于商品详情这类变化频率较低的数据（价格、库存除外），建议使用Redis缓存，TTL设为6小时。伪代码：

import redis r = redis.Redis(host='localhost', decode_responses=True) def get_product_detail_cached(client, product_id): cache_key = f"1688:detail:{product_id}" cached = r.get(cache_key) if cached: return json.loads(cached) data = client.get_product_detail(product_id) if data: r.setex(cache_key, 21600, json.dumps(data)) return data

五、常见错误码及处理

建议在_request方法中针对不同错误码做差异化重试（如限流错误等待更长时间）。

六、数据存储与分析初步

采集到的原始数据可存入Hive或MySQL，后续可进行以下分析：

• 价格区间分布（按类目、按供应商等级）

• 热销商品属性偏好（通过标题分词提取颜色、材质等）

• 供应商供货能力（通过商品总数、月销量总和评估）

• 竞品监控（定时采集指定商品的价格变动）

示例SQL：统计各价格带商品数量

sql

SELECT CASE WHEN price_min < 10 THEN '0-10' WHEN price_min < 50 THEN '10-50' WHEN price_min < 100 THEN '50-100' ELSE '100+' END AS price_band, COUNT(*) AS product_cnt FROM ods_1688_search_result WHERE fetch_time >= CURDATE() - INTERVAL 1 DAY GROUP BY price_band;

七、法律与合规提醒

• 1688平台《开发者协议》明确禁止将API数据用于转售或竞品恶意打压。采集数据仅限内部经营决策使用，不得公开传播原始数据。

• 严格遵守平台限流规则，不得使用分布式绕过频率限制，否则将被封禁应用。

结语

本文从实战角度梳理了1688商品数据采集的完整技术链路，涵盖签名认证、分页处理、异常重试、缓存优化及存储方案。开发者可根据自身业务需求扩展字段映射和调度策略（如使用Airflow定期执行）。所有代码均已脱敏，请替换为自己的合法凭证后运行。数据是电商运营的底层燃料，合理利用API才能构建高效的供应链决策系统。