嵌入式多任务状态机设计与优化实践
2026/5/4 1:32:59
1688 作为阿里巴巴旗下的批发电商平台,其开放的 API 接口为开发者提供了合规获取商品数据、商家信息的渠道。本文将从 API 接入准备、接口调用开发、数据抓取与解析等维度,完整拆解 1688 API 接口开发全流程,并结合实战代码实现商品数据抓取,同时探讨合规与反爬注意事项。
ApiKey和ApiSecret(核心凭证,需妥善保管)。1688 API 基于 HTTP/HTTPS 协议,支持 RESTful 风格,数据格式以 JSON 为主,核心认证方式为签名认证:通过 AppKey、AppSecret、时间戳、请求参数等生成签名,确保请求合法性。
1688 API 请求必须携带签名(sign),生成规则如下:
key1=value1&key2=value2格式;AppSecretkey1=value1&key2=value2AppSecret;所有 API 请求均需携带以下公共参数:
| 参数名 | 说明 | 示例 |
|---|---|---|
| app_key | 应用唯一标识 | 12345678 |
| method | 接口方法名 | alibaba.item.get |
| timestamp | 时间戳(毫秒) | 1718000000000 |
| format | 响应格式 | json |
| v | API 版本 | 2.0 |
| sign | 签名 | E89F95C6089879890987654321 |
以下以 Python 为例,实现 1688 商品详情接口调用,抓取商品标题、价格、规格等核心数据。
安装依赖库:
pip install requests hashlib urllib.parseimport requests import hashlib import time import urllib.parse class Ali1688API: def __init__(self, app_key, app_secret): self.app_key = app_key self.app_secret = app_secret self.base_url = "https://gw.open.1688.com/openapi/param2/1/" def generate_sign(self, params): """生成签名""" # 1. 按参数名ASCII升序排序 sorted_params = sorted(params.items(), key=lambda x: x[0]) # 2. 拼接参数字符串 sign_str = "" for key, value in sorted_params: if value is not None and value != "": sign_str += f"{key}{value}" # 3. 首尾拼接AppSecret sign_str = self.app_secret + sign_str + self.app_secret # 4. MD5加密并转大写 sign = hashlib.md5(sign_str.encode()).hexdigest().upper() return sign def get_item_detail(self, item_id): """ 获取商品详情 :param item_id: 商品ID :return: 商品详情字典 """ # 1. 构造请求参数 method = "alibaba.item.get" # 商品详情接口方法名 timestamp = str(int(time.time() * 1000)) # 毫秒时间戳 params = { "app_key": self.app_key, "method": method, "timestamp": timestamp, "format": "json", "v": "2.0", "item_id": item_id, "fields": "item_id,title,price,spec_info,sku_list,shop_name" # 需要返回的字段 } # 2. 生成签名 sign = self.generate_sign(params) params["sign"] = sign # 3. 发送请求 try: response = requests.get(self.base_url + method, params=params, timeout=10) response.raise_for_status() # 抛出HTTP异常 result = response.json() # 4. 解析响应 if result.get("error_response"): print(f"接口调用失败:{result['error_response']['msg']}") return None return result["alibaba_item_get_response"]["item"] except Exception as e: print(f"请求异常:{str(e)}") return None # #################### 调用示例 #################### if __name__ == "__main__": # 替换为自己的AppKey和AppSecret APP_KEY = "你的AppKey" APP_SECRET = "你的AppSecret" # 初始化API对象 api = Ali1688API(APP_KEY, APP_SECRET) # 抓取指定商品ID的详情(替换为实际商品ID) item_id = "699788888888" item_detail = api.get_item_detail(item_id) if item_detail: print("商品详情:") print(f"商品ID:{item_detail.get('item_id')}") print(f"商品标题:{item_detail.get('title')}") print(f"商品价格:{item_detail.get('price')}") print(f"店铺名称:{item_detail.get('shop_name')}") print(f"规格信息:{item_detail.get('spec_info')}")generate_sign方法严格遵循 1688 签名规则,确保请求通过认证;get_item_detail方法封装了商品详情接口的请求逻辑,支持指定返回字段,减少数据传输量;若需批量抓取商品数据(如店铺商品列表、类目商品),可基于上述基础框架扩展:
以 “阿里巴巴店铺商品列表接口” 为例,核心逻辑:
def get_shop_item_list(self, seller_nick, page_no=1, page_size=20): """ 获取店铺商品列表 :param seller_nick: 卖家昵称 :param page_no: 页码 :param page_size: 每页条数 :return: 商品列表 """ method = "alibaba.item.seller.list.get" timestamp = str(int(time.time() * 1000)) params = { "app_key": self.app_key, "method": method, "timestamp": timestamp, "format": "json", "v": "2.0", "seller_nick": seller_nick, "page_no": page_no, "page_size": page_size, "fields": "item_id,title,price" } sign = self.generate_sign(params) params["sign"] = sign try: response = requests.get(self.base_url + method, params=params, timeout=10) result = response.json() if result.get("error_response"): print(f"获取商品列表失败:{result['error_response']['msg']}") return None return result["alibaba_item_seller_list_get_response"]["items"]["item"] except Exception as e: print(f"批量抓取异常:{str(e)}") return None对于大批量数据(如上万条商品),建议使用异步请求提升效率:
import aiohttp import asyncio async def async_get_item_detail(self, item_id): """异步获取商品详情""" # 签名生成逻辑同同步方法 timestamp = str(int(time.time() * 1000)) params = { "app_key": self.app_key, "method": "alibaba.item.get", "timestamp": timestamp, "format": "json", "v": "2.0", "item_id": item_id, "fields": "item_id,title,price" } params["sign"] = self.generate_sign(params) async with aiohttp.ClientSession() as session: try: async with session.get(self.base_url + "alibaba.item.get", params=params, timeout=10) as response: result = await response.json() return result["alibaba_item_get_response"]["item"] except Exception as e: print(f"异步请求失败:{str(e)}") return None # 批量异步调用 async def batch_crawl_items(self, item_ids): tasks = [self.async_get_item_detail(item_id) for item_id in item_ids] results = await asyncio.gather(*tasks) return [res for res in results if res is not None]1688 API 接口开发的核心在于签名认证的正确实现和请求频率的合理控制。本文从前置准备、核心流程、代码实现到批量抓取优化,完整覆盖了商品数据抓取的全流程。在实际开发中,需结合业务场景选择同步 / 异步方案,同时严格遵守平台规则,确保数据抓取的合规性和稳定性。
对于复杂场景(如实时商品监控、多维度数据分析),可基于本文代码扩展数据存储(如 MySQL、MongoDB)、数据清洗、可视化等模块,形成完整的 1688 数据应用体系。