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

import hashlibimport urllib.parsedef 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)}")

import requestsimport hashlibimport timeimport jsonfrom typing import List, Optional, Dictfrom dataclasses import dataclassfrom urllib.parse import urlencode@dataclassclass 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}")

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

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)}

import timeimport randomfrom functools import wrapsdef 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 decoratordef 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 decoratorclass Alibaba1688Client:
    # ... 在原有方法上应用装饰器 ...
    
    @rate_limited(max_per_minute=50)  # 个人开发者默认 50 次/分钟
    @exponential_backoff(max_retries=3)
    def search_by_keyword(self, **kwargs):
        # ... 原有实现 ...
        pass

import sqlite3from datetime import datetimeclass 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()