技术拆解：1688 API 接口开发全流程与商品数据抓取方案

pip install requests hashlib urllib.parse

import 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')}")

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]