分析中 已回复 待规划 {{opt.name}}

分析中 已回复 待规划

Python 获取 1688 关键词搜索商品与价格接口实战指南

1688（阿里巴巴旗下 B2B 批发平台）提供了丰富的 API 接口，允许开发者通过关键词检索商品信息、获取实时价格与库存数据。本文将深入讲解如何使用 Python 合规接入 1688 开放平台，实现关键词搜索商品列表及价格数据的完整流程。

一、1688 开放平台接口体系概览

1688 开放平台提供了多层次的商品数据接口，核心包括：

二、前置准备：账号与权限申请

2.1 注册与认证流程

1. 访问 1688 开放平台 注册开发者账号
2. 完成企业实名认证（个人认证功能受限）
3. 创建应用，获取 AppKey 和 AppSecret
4. 申请 alibaba.product.search 接口权限（需提交搜索场景说明，审核 1-2 个工作日）
5. 配置 IP 白名单（遗漏会导致 403 Forbidden）

2.2 核心凭证

三、签名机制：MD5 加密详解

1688 API 采用 MD5 签名 确保请求安全，签名生成规则如下：

1. 将所有请求参数（除 sign 外）按参数名 ASCII 升序排序
2. 拼接为 key1value1key2value2... 格式（无连接符）
3. 在字符串首尾各拼接一次 app_secret
4. 进行 MD5 加密，结果转大写

import hashlib
import urllib.parse

def generate_sign(params: dict, app_secret: str) -> str:
    """
    生成 1688 API 请求签名（MD5）
    
    规则：app_secret + 排序后参数拼接 + app_secret，整体 MD5 大写
    """
    # 1. 按参数名 ASCII 升序排序
    sorted_params = sorted(params.items())
    
    # 2. 拼接为 keyvalue 格式（无连接符）
    param_str = ''.join([f"{k}{v}" for k, v in sorted_params])
    
    # 3. 首尾拼接 app_secret
    sign_str = f"{app_secret}{param_str}{app_secret}"
    
    # 4. MD5 加密并转大写
    sign = hashlib.md5(sign_str.encode('utf-8')).hexdigest().upper()
    
    return sign


# 验证签名生成
if __name__ == "__main__":
    test_params = {
        "method": "com.alibaba.product.search",
        "app_key": "12345678",
        "timestamp": "2026-05-25 12:00:00",
        "keywords": "蓝牙耳机",
        "page_no": 1,
        "page_size": 20,
        "format": "json",
        "v": "2.0"
    }
    secret = "your_app_secret_here"
    print(f"签名结果: {generate_sign(test_params, secret)}")
    

四、核心接口实战：关键词搜索商品

4.1 接口参数详解

alibaba.product.search 是关键词搜索的核心接口，2026 年主推 2.0 版本，1.0 版本已不返回批发价、起批量等核心字段。

4.2 完整 Python 实现

import requests
import hashlib
import time
import json
from typing import List, Optional, Dict
from dataclasses import dataclass
from urllib.parse import urlencode


@dataclass
class ProductItem:
    """1688 商品数据结构"""
    product_id: str
    title: str
    price: float
    price_range: str          # 价格区间（如"10.00-50.00"）
    moq: int                  # 最小起订量
    unit: str                 # 计量单位
    quantity: int             # 可售数量/库存
    sale_info: str            # 销量信息
    supplier_name: str        # 供应商名称
    supplier_id: str          # 供应商 ID
    location: str             # 发货地
    image_url: str            # 主图 URL
    detail_url: str           # 商品详情页链接
    is_support_mix: bool      # 是否支持混批
    credit_level: str         # 诚信通等级


class Alibaba1688Client:
    """
    1688 开放平台 API 客户端
    支持关键词搜索、商品详情获取
    """
    
    # 1688 API 网关地址
    BASE_URL = "https://gw.open.1688.com/openapi/param2/2/portals.open/api.listOfferDetail/"
    # 或: "https://api.1688.com/router/rest"
    
    def __init__(self, app_key: str, app_secret: str, access_token: Optional[str] = None):
        self.app_key = app_key
        self.app_secret = app_secret
        self.access_token = access_token
        self.session = requests.Session()
        self.session.headers.update({
            "Accept": "application/json",
            "Content-Type": "application/x-www-form-urlencoded;charset=utf-8"
        })
    
    def _generate_sign(self, params: dict) -> str:
        """生成 MD5 签名"""
        sorted_params = sorted(params.items())
        param_str = ''.join([f"{k}{v}" for k, v in sorted_params])
        sign_str = f"{self.app_secret}{param_str}{self.app_secret}"
        return hashlib.md5(sign_str.encode('utf-8')).hexdigest().upper()
    
    def _build_params(self, **kwargs) -> dict:
        """构建公共参数 + 业务参数"""
        params = {
            "method": kwargs.pop("method", "com.alibaba.product.search"),
            "app_key": self.app_key,
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
            "v": "2.0",
            "format": "json",
            "sign_method": "md5"
        }
        # 合并业务参数
        params.update(kwargs)
        # 生成签名
        params["sign"] = self._generate_sign(params)
        return params
    
    def search_by_keyword(
        self,
        keyword: str,
        page_no: int = 1,
        page_size: int = 20,
        sort_type: str = "total",
        price_start: Optional[float] = None,
        price_end: Optional[float] = None,
        category_id: Optional[int] = None
    ) -> Dict:
        """
        关键词搜索商品列表
        
        Args:
            keyword: 搜索关键词（如"蓝牙耳机"、"连衣裙"）
            page_no: 页码
            page_size: 每页数量（最大 50）
            sort_type: 排序方式
                - "total": 按销量（默认）
                - "price_asc": 价格升序
                - "price_desc": 价格降序
                - "credit": 信用排序
            price_start: 价格区间下限
            price_end: 价格区间上限
            category_id: 类目 ID 过滤
        
        Returns:
            API 原始响应字典
        """
        business_params = {
            "keywords": keyword,
            "page_no": page_no,
            "page_size": page_size,
            "sort_type": sort_type
        }
        
        if price_start is not None:
            business_params["price_start"] = price_start
        if price_end is not None:
            business_params["price_end"] = price_end
        if category_id is not None:
            business_params["category_id"] = category_id
            
        params = self._build_params(**business_params)
        
        try:
            response = self.session.get(self.BASE_URL, params=params, timeout=30)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            print(f"请求异常: {e}")
            return {"error": str(e)}
    
    def parse_products(self, api_response: dict) -> List[ProductItem]:
        """
        解析 API 响应，提取结构化商品数据
        """
        products = []
        
        if not api_response or "result" not in api_response:
            print(f"响应异常: {api_response.get('error', '未知错误')}")
            return products
        
        result = api_response["result"]
        product_list = result.get("products", [])
        
        for item in product_list:
            try:
                product = ProductItem(
                    product_id=str(item.get("productId", "")),
                    title=item.get("subject", ""),
                    price=float(item.get("price", 0) or 0),
                    price_range=item.get("priceRange", ""),
                    moq=int(item.get("minOrderQuantity", 1) or 1),
                    unit=item.get("unit", "件"),
                    quantity=int(item.get("quantity", 0) or 0),
                    sale_info=item.get("saleInfo", ""),
                    supplier_name=item.get("supplierName", ""),
                    supplier_id=str(item.get("supplierId", "")),
                    location=item.get("location", ""),
                    image_url=item.get("imageUrl", ""),
                    detail_url=item.get("detailUrl", ""),
                    is_support_mix=item.get("isSupportMix", False),
                    credit_level=item.get("creditLevel", "")
                )
                products.append(product)
            except (ValueError, TypeError) as e:
                print(f"解析商品数据异常: {e}, 原始数据: {item}")
                continue
                
        return products


# ==================== 使用示例 ====================

if __name__ == "__main__":
    # 初始化客户端（替换为真实凭证）
    client = Alibaba1688Client(
        app_key="your_app_key",
        app_secret="your_app_secret"
    )
    
    # 示例 1：基础关键词搜索
    print("=" * 60)
    print("示例 1：搜索'蓝牙耳机'，按销量排序")
    print("=" * 60)
    
    response = client.search_by_keyword(
        keyword="蓝牙耳机",
        page_no=1,
        page_size=20,
        sort_type="total"
    )
    
    products = client.parse_products(response)
    
    print(f"共获取 {len(products)} 个商品：\n")
    for idx, p in enumerate(products[:5], 1):
        print(f"【{idx}】{p.title[:40]}...")
        print(f"    商品ID: {p.product_id}")
        print(f"    价格: ¥{p.price} ({p.price_range})")
        print(f"    起订量: {p.moq} {p.unit}")
        print(f"    库存: {p.quantity}")
        print(f"    供应商: {p.supplier_name} ({p.location})")
        print(f"    诚信通: {p.credit_level}")
        print(f"    链接: {p.detail_url}")
        print()
    
    # 示例 2：带价格筛选的搜索
    print("=" * 60)
    print("示例 2：搜索'连衣裙'，价格 50-100 元，按价格升序")
    print("=" * 60)
    
    response = client.search_by_keyword(
        keyword="连衣裙",
        page_no=1,
        page_size=20,
        sort_type="price_asc",
        price_start=50.0,
        price_end=100.0
    )
    
    products = client.parse_products(response)
    print(f"筛选后共 {len(products)} 个商品\n")
    for p in products[:3]:
        print(f"¥{p.price} | {p.title[:30]}... | MOQ:{p.moq}{p.unit}")
    
    # 示例 3：分页遍历（获取前 100 条）
    print("\n" + "=" * 60)
    print("示例 3：分页获取前 100 条商品")
    print("=" * 60)
    
    all_products = []
    for page in range(1, 6):  # 5 页 × 20 = 100 条
        resp = client.search_by_keyword(
            keyword="手机壳",
            page_no=page,
            page_size=20
        )
        page_products = client.parse_products(resp)
        all_products.extend(page_products)
        
        # 限流控制：个人开发者 50 次/分钟
        if page < 5:
            time.sleep(1.5)
    
    print(f"累计获取 {len(all_products)} 个商品")
    
    # 价格分析
    if all_products:
        prices = [p.price for p in all_products if p.price > 0]
        avg_price = sum(prices) / len(prices)
        min_price = min(prices)
        max_price = max(prices)
        print(f"价格统计: 最低¥{min_price}, 最高¥{max_price}, 平均¥{avg_price:.2f}")
        

五、商品详情接口：获取价格与 SKU 深度数据

搜索接口返回的是商品摘要，如需获取完整价格体系（阶梯价、SKU 价）、库存、图文详情，需调用 alibaba.product.get 接口。

5.1 商品详情接口实

class Alibaba1688Client:
    # ... 接上文 ...
    
    def get_product_detail(
        self,
        product_id: str,
        fields: Optional[str] = None
    ) -> Dict:
        """
        获取商品详情（含价格、SKU、库存、图文等）
        
        Args:
            product_id: 1688 商品数字 ID
            fields: 指定返回字段，逗号分隔
                如: "shippingInfo,saleInfo,skuInfos,productInfo"
                不传则返回全部字段
        
        Returns:
            商品详情字典
        """
        business_params = {
            "method": "com.alibaba.product.get",
            "productId": product_id
        }
        
        if fields:
            business_params["fields"] = fields
            
        params = self._build_params(**business_params)
        
        try:
            response = self.session.get(self.BASE_URL, params=params, timeout=30)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            return {"error": str(e)}
    
    def parse_price_info(self, detail_response: dict) -> Dict:
        """
        解析商品价格体系（批发价、阶梯价、SKU 价）
        """
        price_info = {
            "product_id": "",
            "title": "",
            "reference_price": 0.0,      # 参考价/划线价
            "sale_price": 0.0,           # 当前售价
            "price_range": "",           # 价格区间
            "moq": 1,                    # 最小起订量
            "unit": "件",
            "wholesale_tiers": [],       # 阶梯批发价
            "sku_prices": []             # SKU 规格价格
        }
        
        if "result" not in detail_response:
            return price_info
        
        result = detail_response["result"]
        product = result.get("productInfo", {})
        
        price_info["product_id"] = str(product.get("id", ""))
        price_info["title"] = product.get("subject", "")
        price_info["reference_price"] = float(product.get("referencePrice", 0) or 0)
        price_info["sale_price"] = float(product.get("saleInfo", {}).get("price", 0) or 0)
        price_info["price_range"] = product.get("priceRange", "")
        price_info["moq"] = int(product.get("saleInfo", {}).get("minQuantity", 1) or 1)
        price_info["unit"] = product.get("saleInfo", {}).get("unit", "件")
        
        # 解析阶梯批发价
        sale_info = product.get("saleInfo", {})
        if "priceRanges" in sale_info:
            for tier in sale_info["priceRanges"]:
                price_info["wholesale_tiers"].append({
                    "min_quantity": tier.get("startQuantity", 0),
                    "max_quantity": tier.get("endQuantity", 0),
                    "price": float(tier.get("price", 0) or 0),
                    "unit": tier.get("unit", "件")
                })
        
        # 解析 SKU 价格
        sku_infos = product.get("skuInfos", [])
        for sku in sku_infos:
            sku_price = {
                "sku_id": sku.get("skuId", ""),
                "attributes": [],          # 规格属性（颜色、尺码等）
                "price": float(sku.get("price", 0) or 0),
                "quantity": int(sku.get("quantity", 0) or 0),
                "sku_code": sku.get("skuCode", "")
            }
            
            # 解析规格属性
            if "attributes" in sku:
                for attr in sku["attributes"]:
                    sku_price["attributes"].append({
                        "name": attr.get("attributeName", ""),
                        "value": attr.get("attributeValue", "")
                    })
            
            price_info["sku_prices"].append(sku_price)
        
        return price_info


# 商品详情使用示例
if __name__ == "__main__":
    client = Alibaba1688Client(
        app_key="your_app_key",
        app_secret="your_app_secret"
    )
    
    # 获取商品详情
    product_id = "619899292404"  # 示例商品 ID
    detail = client.get_product_detail(product_id)
    
    # 解析价格信息
    price_data = client.parse_price_info(detail)
    
    print(f"商品: {price_data['title']}")
    print(f"参考价: ¥{price_data['reference_price']}")
    print(f"售价: ¥{price_data['sale_price']}")
    print(f"价格区间: {price_data['price_range']}")
    print(f"起订量: {price_data['moq']} {price_data['unit']}")
    
    print("\n【阶梯批发价】")
    for tier in price_data["wholesale_tiers"]:
        print(f"  {tier['min_quantity']}-{tier['max_quantity']} {tier['unit']}: ¥{tier['price']}")
    
    print("\n【SKU 规格价格】")
    for sku in price_data["sku_prices"]:
        attrs = " | ".join([f"{a['name']}:{a['value']}" for a in sku["attributes"]])
        print(f"  SKU {sku['sku_id']}: {attrs} → ¥{sku['price']} (库存:{sku['quantity']})")
        

六、寻源通接口：跨境采购场景

1688 寻源通（alibaba.wholesale.goods.search）专为采购商设计，支持供应商资质筛选、跨境选品等高级功能。

class Alibaba1688Client:
    # ... 接上文 ...
    
    def wholesale_search(
        self,
        keywords: str,
        page_no: int = 1,
        page_size: int = 50,
        price_range: Optional[str] = None,   # "10,100" 表示 10-100 元
        is_pre_sale: bool = False,
        sort: str = "price_asc",              # price_asc, price_desc, sale_desc
        high_quality: bool = True               # 是否筛选实力商家
    ) -> Dict:
        """
        寻源通商品搜索（跨境采购场景）
        """
        business_params = {
            "method": "alibaba.wholesale.goods.search",
            "keywords": keywords,
            "page_no": page_no,
            "page_size": page_size,
            "sort": sort,
            "is_pre_sale": "true" if is_pre_sale else "false",
            "high_quality": "true" if high_quality else "false"
        }
        
        if price_range:
            business_params["price_range"] = price_range
            
        params = self._build_params(**business_params)
        
        try:
            # 寻源通使用不同网关
            url = "https://gw.open.1688.com/openapi/param2/1/com.alibaba.trade/alibaba.wholesale.goods.search"
            response = self.session.get(url, params=params, timeout=30)
            return response.json()
        except requests.exceptions.RequestException as e:
            return {"error": str(e)}
            

七、异常处理与限流控制

7.1 常见错误码

7.2 限流与重试机制

import time
import random
from functools import wraps

def rate_limited(max_per_minute=50):
    """限流装饰器：控制调用频率"""
    min_interval = 60.0 / max_per_minute
    last_called = [0.0]
    
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            elapsed = time.time() - last_called[0]
            if elapsed < min_interval:
                time.sleep(min_interval - elapsed)
            result = func(*args, **kwargs)
            last_called[0] = time.time()
            return result
        return wrapper
    return decorator


def exponential_backoff(max_retries=3, base_delay=1.0):
    """指数退避重试"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except requests.exceptions.HTTPError as e:
                    if e.response.status_code == 429:
                        delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
                        print(f"触发限流，等待 {delay:.1f}s 后重试 ({attempt+1}/{max_retries})")
                        time.sleep(delay)
                    else:
                        raise
            return func(*args, **kwargs)  # 最后一次尝试
        return wrapper
    return decorator


class Alibaba1688Client:
    # ... 在原有方法上应用装饰器 ...
    
    @rate_limited(max_per_minute=50)  # 个人开发者默认 50 次/分钟
    @exponential_backoff(max_retries=3)
    def search_by_keyword(self, **kwargs):
        # ... 原有实现 ...
        pass
        

八、数据持久化与价格监控

将采集数据存入数据库，实现价格变动监控：

import sqlite3
from datetime import datetime

class PriceMonitorDB:
    """价格监控数据库"""
    
    def __init__(self, db_path: str = "1688_monitor.db"):
        self.conn = sqlite3.connect(db_path)
        self._init_tables()
    
    def _init_tables(self):
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS products (
                product_id TEXT PRIMARY KEY,
                title TEXT,
                current_price REAL,
                price_range TEXT,
                moq INTEGER,
                supplier TEXT,
                location TEXT,
                last_updated TIMESTAMP
            )
        """)
        
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS price_history (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                product_id TEXT,
                price REAL,
                recorded_at TIMESTAMP,
                FOREIGN KEY (product_id) REFERENCES products(product_id)
            )
        """)
        self.conn.commit()
    
    def save_product(self, product: ProductItem):
        """保存或更新商品信息"""
        self.conn.execute("""
            INSERT OR REPLACE INTO products 
            (product_id, title, current_price, price_range, moq, supplier, location, last_updated)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
        """, (
            product.product_id, product.title, product.price,
            product.price_range, product.moq, product.supplier_name,
            product.location, datetime.now()
        ))
        
        # 记录价格历史
        self.conn.execute("""
            INSERT INTO price_history (product_id, price, recorded_at)
            VALUES (?, ?, ?)
        """, (product.product_id, product.price, datetime.now()))
        
        self.conn.commit()
    
    def get_price_trend(self, product_id: str, days: int = 7) -> List[Dict]:
        """获取价格趋势"""
        cursor = self.conn.execute("""
            SELECT price, recorded_at FROM price_history
            WHERE product_id = ? AND recorded_at > datetime('now', '-{} days')
            ORDER BY recorded_at
        """.format(days), (product_id,))
        
        return [{"price": row[0], "time": row[1]} for row in cursor.fetchall()]
    
    def detect_price_change(self, product_id: str, new_price: float) -> Optional[float]:
        """检测价格变动，返回旧价格"""
        cursor = self.conn.execute(
            "SELECT current_price FROM products WHERE product_id = ?", 
            (product_id,)
        )
        row = cursor.fetchone()
        if row and row[0] != new_price:
            return row[0]
        return None
        

九、完整监控任务示例

def run_monitor_task():
    """定时监控任务"""
    client = Alibaba1688Client(
        app_key="your_app_key",
        app_secret="your_app_secret"
    )
    db = PriceMonitorDB()
    
    # 监控关键词列表
    keywords = ["蓝牙耳机", "手机壳", "数据线"]
    
    for keyword in keywords:
        print(f"\n正在监控关键词: {keyword}")
        
        response = client.search_by_keyword(
            keyword=keyword,
            page_no=1,
            page_size=50,
            sort_type="total"
        )
        
        products = client.parse_products(response)
        
        for product in products:
            # 保存数据
            db.save_product(product)
            
            # 检测价格变动
            old_price = db.detect_price_change(product.product_id, product.price)
            if old_price is not None:
                change_pct = (product.price - old_price) / old_price * 100
                print(f"⚠️ 价格变动: {product.title[:30]}...")
                print(f"   ¥{old_price} → ¥{product.price} ({change_pct:+.1f}%)")
            
            # 获取详情（深度监控前 10 名商品）
            if product.price > 0:
                detail = client.get_product_detail(product.product_id)
                price_detail = client.parse_price_info(detail)
                # 可进一步分析阶梯价、SKU 库存等
        
        time.sleep(2)  # 关键词间间隔


if __name__ == "__main__":
    run_monitor_task()
    

十、关键注意事项

十一、总结

如遇任何疑问或有进一步的需求，请随时与我私信或者评论联系。