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

分析中 已回复 待规划

1688图搜+关键词搜索商品API：从入门到实战，让你效率翻倍

一、概述

1688作为国内最大的B2B批发采购平台，拥有海量商品数据。对于电商开发者、选品工具和供应链系统而言，图片搜索（图搜/拍立淘）和关键词搜索是两种最高效的找货方式。本文将带你从0到1掌握1688图搜API和关键词搜索API的完整调用流程，附赠实战代码，让你的开发效率翻倍。

二、准备工作：账号与权限

2.1 注册1688开放平台账号

1688开放平台主要面向企业开发者，个人开发者权限受限。

1. 访问 1688开放平台 官网
2. 使用企业支付宝账号注册并完成企业实名认证
3. 提交营业执照、法人身份证等资质

2.2 创建应用获取凭证

在控制台创建应用后，你将获得：

• App Key：应用唯一标识
• App Secret：签名密钥（严格保密）

2.3 申请API权限

根据业务需求申请对应接口权限：

⚠️ 注意：部分接口需要通过OAuth2.0授权获取 access_token，特别是涉及店铺数据的操作。

三、签名生成机制

1688开放平台采用 HMAC-SHA1 或 MD5 签名算法，所有请求参数必须按规则生成签名。

3.1 签名规则

1. 排序：将所有参数（除 sign 外）按参数名 ASCII 码升序排列
2. 拼接：格式为 appSecret + key1value1 + key2value2 + ... + appSecret
3. 加密：使用 MD5 加密，结果转大写

3.2 Python签名工具类

import hashlib
import hmac
import time
import urllib.parse
from datetime import datetime

class AlibabaApiSigner:
    """1688开放平台API签名工具类"""
    
    def __init__(self, app_key: str, app_secret: str):
        self.app_key = app_key
        self.app_secret = app_secret
        self.base_url = "https://api.1688.com/router/rest"
    
    def generate_md5_sign(self, params: dict) -> str:
        """
        生成MD5签名（适用于大多数1688接口）
        """
        # 过滤空值和sign参数
        filtered = {k: v for k, v in params.items() 
                   if v is not None and k != 'sign'}
        
        # 按key升序排列
        sorted_items = sorted(filtered.items(), key=lambda x: x[0])
        
        # 拼接字符串
        sign_content = self.app_secret
        for key, value in sorted_items:
            sign_content += f"{key}{value}"
        sign_content += self.app_secret
        
        # MD5加密并转大写
        return hashlib.md5(sign_content.encode('utf-8')).hexdigest().upper()
    
    def generate_hmac_sign(self, params: dict) -> str:
        """
        生成HMAC-SHA1签名（部分接口使用）
        """
        filtered = {k: v for k, v in params.items() 
                   if v is not None and k != 'sign'}
        sorted_items = sorted(filtered.items(), key=lambda x: x[0])
        
        sign_content = self.app_secret
        for key, value in sorted_items:
            sign_content += f"{key}{value}"
        sign_content += self.app_secret
        
        # HMAC-SHA1加密
        sign = hmac.new(
            self.app_secret.encode('utf-8'),
            sign_content.encode('utf-8'),
            hashlib.sha1
        ).hexdigest().upper()
        return sign
    
    def build_base_params(self) -> dict:
        """构建基础请求参数"""
        return {
            'app_key': self.app_key,
            'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
            'format': 'json',
            'v': '2.0',
            'sign_method': 'md5'
        }
        

四、关键词搜索商品API实战

4.1 接口概述

关键词搜索是1688最核心的找货方式，支持按关键词、价格区间、类目、排序等多维度筛选。

接口地址：https://api.1688.com/router/rest
请求方式：POST
数据格式：application/x-www-form-urlencoded

4.2 核心参数说明

4.3 完整调用代码 测试接口获取key

import requests
import json
import time
from typing import List, Dict, Optional

class AlibabaKeywordSearch:
    """1688关键词搜索商品API封装"""
    
    def __init__(self, app_key: str, app_secret: str):
        self.signer = AlibabaApiSigner(app_key, app_secret)
        self.api_url = "https://api.1688.com/router/rest"
    
    def search_products(
        self,
        keyword: str,
        page: int = 1,
        page_size: int = 20,
        price_start: Optional[float] = None,
        price_end: Optional[float] = None,
        category_id: Optional[int] = None,
        sort: str = "default"
    ) -> Dict:
        """
        关键词搜索商品
        
        :param keyword: 搜索关键词，如"无线蓝牙耳机"
        :param page: 页码
        :param page_size: 每页数量（建议不超过50）
        :param price_start: 最低价格
        :param price_end: 最高价格
        :param category_id: 类目ID过滤
        :param sort: 排序方式
        :return: 搜索结果字典
        """
        # 构建基础参数
        params = self.signer.build_base_params()
        params['method'] = 'com.alibaba.product.search'  # 或 alibaba.wholesale.goods.search
        
        # 业务参数
        params['q'] = keyword
        params['page'] = page
        params['pageSize'] = page_size
        
        if price_start is not None:
            params['priceStart'] = price_start
        if price_end is not None:
            params['priceEnd'] = price_end
        if category_id is not None:
            params['categoryId'] = category_id
        if sort != "default":
            params['sort'] = sort
        
        # 生成签名
        params['sign'] = self.signer.generate_md5_sign(params)
        
        # 发送请求
        try:
            response = requests.post(
                self.api_url,
                data=params,
                headers={'Content-Type': 'application/x-www-form-urlencoded'},
                timeout=30
            )
            response.raise_for_status()
            return response.json()
        except requests.RequestException as e:
            print(f"请求异常: {e}")
            return {"error": str(e)}
    
    def parse_search_results(self, response: Dict) -> List[Dict]:
        """
        解析搜索结果，提取关键字段
        """
        results = []
        
        # 根据实际返回结构解析（不同接口返回结构可能不同）
        if 'result' in response:
            products = response['result'].get('result', [])
        elif 'productSearchResponse' in response:
            products = response['productSearchResponse'].get('products', [])
        else:
            # 适配不同返回结构
            products = response.get('products', [])
        
        for product in products:
            item = {
                'product_id': product.get('productId') or product.get('offerId'),
                'title': product.get('subject') or product.get('title'),
                'price': product.get('price') or product.get('salePrice'),
                'image_url': product.get('imageUrl') or product.get('mainImage'),
                'supplier_name': product.get('companyName') or product.get('supplier'),
                'min_order': product.get('minOrderQuantity', 1),
                'sale_count': product.get('saleCount', 0),
                'detail_url': product.get('detailUrl', ''),
                'location': product.get('location', '')
            }
            results.append(item)
        
        return results
    
    def batch_search(
        self,
        keywords: List[str],
        pages: int = 3,
        delay: float = 1.5
    ) -> Dict[str, List[Dict]]:
        """
        批量关键词搜索（带限流控制）
        
        :param keywords: 关键词列表
        :param pages: 每个关键词搜索页数
        :param delay: 请求间隔（秒）
        :return: {keyword: [products]}
        """
        all_results = {}
        
        for keyword in keywords:
            print(f"🔍 正在搜索: {keyword}")
            keyword_results = []
            
            for page in range(1, pages + 1):
                response = self.search_products(
                    keyword=keyword,
                    page=page,
                    page_size=50
                )
                
                products = self.parse_search_results(response)
                if not products:
                    break
                    
                keyword_results.extend(products)
                print(f"  第{page}页获取 {len(products)} 条数据")
                
                # 限流：避免触发QPS限制
                if page < pages:
                    time.sleep(delay)
            
            all_results[keyword] = keyword_results
            print(f"✅ {keyword} 共获取 {len(keyword_results)} 条数据\n")
        
        return all_results


# ========== 实战演示 ==========
if __name__ == '__main__':
    APP_KEY = 'your_app_key'
    APP_SECRET = 'your_app_secret'
    
    searcher = AlibabaKeywordSearch(APP_KEY, APP_SECRET)
    
    # 单关键词搜索
    print("=" * 60)
    print("【单关键词搜索示例】")
    print("=" * 60)
    
    result = searcher.search_products(
        keyword="无线蓝牙耳机",
        page=1,
        page_size=10,
        price_start=10,
        price_end=100,
        sort="price_asc"
    )
    
    products = searcher.parse_search_results(result)
    for i, p in enumerate(products[:5], 1):
        print(f"{i}. {p['title'][:30]}... | ¥{p['price']} | 销量:{p['sale_count']}")
    
    # 批量关键词搜索
    print("\n" + "=" * 60)
    print("【批量关键词搜索示例】")
    print("=" * 60)
    
    keywords = ["T恤批发", "手机壳", "数据线", "保温杯"]
    batch_results = searcher.batch_search(keywords, pages=2, delay=2)
    
    for kw, items in batch_results.items():
        print(f"📦 {kw}: {len(items)} 条商品")
        

五、图搜（以图搜货/拍立淘）API实战

5.1 接口概述

1688图搜API（拍立淘/以图搜货）是官方图像搜品接口，支持通过图片URL或Base64编码输入图片，秒级返回同款/相似商品的结构化数据。

核心能力：

• 上传商品图片 → 返回同款/相似商品
• 返回商品ID、标题、价格、供应商等结构化数据
• 支持图片URL或Base64两种输入方式

5.2 核心参数说明

5.3 完整调用代码

import base64
import requests
from typing import Optional

class AlibabaImageSearch:
    """1688图搜（拍立淘）API封装"""
    
    def __init__(self, app_key: str, app_secret: str):
        self.signer = AlibabaApiSigner(app_key, app_secret)
        self.api_url = "https://api.1688.com/router/rest"
    
    def search_by_url(
        self,
        image_url: str,
        category_id: Optional[int] = None,
        keyword: Optional[str] = None
    ) -> dict:
        """
        通过图片URL搜索商品
        
        :param image_url: 图片的网络地址
        :param category_id: 指定类目ID（可选）
        :param keyword: 辅助关键词（可选）
        :return: 搜索结果
        """
        params = self.signer.build_base_params()
        params['method'] = 'alibaba.image.search'
        params['image_url'] = image_url
        
        if category_id:
            params['category_id'] = category_id
        if keyword:
            params['keyword'] = keyword
        
        params['sign'] = self.signer.generate_md5_sign(params)
        
        try:
            response = requests.post(
                self.api_url,
                data=params,
                timeout=30
            )
            return response.json()
        except Exception as e:
            return {"error": str(e)}
    
    def search_by_file(
        self,
        image_path: str,
        category_id: Optional[int] = None,
        keyword: Optional[str] = None
    ) -> dict:
        """
        通过本地图片文件搜索商品（Base64编码上传）
        
        :param image_path: 本地图片路径
        :param category_id: 指定类目ID
        :param keyword: 辅助关键词
        :return: 搜索结果
        """
        # 读取图片并Base64编码
        try:
            with open(image_path, 'rb') as f:
                image_bytes = f.read()
                image_base64 = base64.b64encode(image_bytes).decode('utf-8')
        except Exception as e:
            return {"error": f"图片读取失败: {e}"}
        
        params = self.signer.build_base_params()
        params['method'] = 'alibaba.image.search'
        params['image_content'] = image_base64
        
        if category_id:
            params['category_id'] = category_id
        if keyword:
            params['keyword'] = keyword
        
        params['sign'] = self.signer.generate_md5_sign(params)
        
        try:
            response = requests.post(
                self.api_url,
                data=params,
                timeout=60  # 图搜可能需要更长时间
            )
            return response.json()
        except Exception as e:
            return {"error": str(e)}
    
    def parse_image_results(self, response: dict) -> list:
        """
        解析图搜结果
        """
        results = []
        
        # 适配不同返回结构
        if 'imageSearchResponse' in response:
            items = response['imageSearchResponse'].get('items', [])
        elif 'result' in response:
            items = response['result'].get('items', [])
        else:
            items = response.get('items', [])
        
        for item in items:
            result = {
                'product_id': item.get('productId') or item.get('offerId'),
                'title': item.get('subject') or item.get('title'),
                'price': item.get('price'),
                'image_url': item.get('imageUrl') or item.get('mainImage'),
                'similarity': item.get('similarity', 0),  # 相似度分数
                'supplier': item.get('companyName') or item.get('supplier'),
                'detail_url': item.get('detailUrl', ''),
                'min_order': item.get('minOrderQuantity', 1)
            }
            results.append(result)
        
        return results


# ========== 图搜实战演示 ==========
if __name__ == '__main__':
    APP_KEY = 'your_app_key'
    APP_SECRET = 'your_app_secret'
    
    image_searcher = AlibabaImageSearch(APP_KEY, APP_SECRET)
    
    print("=" * 60)
    print("【图搜实战：通过图片URL搜索】")
    print("=" * 60)
    
    # 方式1：通过图片URL搜索
    image_url = "https://example.com/your-product-image.jpg"
    result = image_searcher.search_by_url(
        image_url=image_url,
        category_id=1512,  # 可选：限定在手机数码类目
        keyword="蓝牙耳机"
    )
    
    items = image_searcher.parse_image_results(result)
    print(f"找到 {len(items)} 个相似商品：")
    for i, item in enumerate(items[:5], 1):
        print(f"{i}. {item['title'][:35]}... | ¥{item['price']} | 相似度:{item['similarity']}%")
    
    # 方式2：通过本地文件搜索
    print("\n" + "=" * 60)
    print("【图搜实战：通过本地图片文件搜索】")
    print("=" * 60)
    
    # result = image_searcher.search_by_file(
    #     image_path="/path/to/your/image.jpg",
    #     keyword="手机壳"
    # )
    # items = image_searcher.parse_image_results(result)
    

六、高级实战：关键词+图搜组合策略

6.1 组合搜索场景

在实际业务中，单一搜索方式往往不够精准。以下是几种高效的组合策略

6.2 组合搜索实战代码

class AlibabaSearchCombo:
    """1688组合搜索策略：关键词+图搜联动"""
    
    def __init__(self, app_key: str, app_secret: str):
        self.keyword_searcher = AlibabaKeywordSearch(app_key, app_secret)
        self.image_searcher = AlibabaImageSearch(app_key, app_secret)
    
    def find_source_factory(self, product_title: str) -> list:
        """
        找同款货源：先关键词搜索，再用图搜验证
        
        :param product_title: 商品标题
        :return: 源头工厂列表
        """
        print(f"🔍 步骤1：关键词搜索 '{product_title}'")
        
        # 第一步：关键词搜索获取候选商品
        kw_result = self.keyword_searcher.search_products(
            keyword=product_title,
            page=1,
            page_size=10,
            sort="sale"  # 按销量排序
        )
        candidates = self.keyword_searcher.parse_search_results(kw_result)
        
        if not candidates:
            return []
        
        # 取第一个商品的图片进行图搜
        first_product = candidates[0]
        image_url = first_product.get('image_url')
        
        if not image_url:
            return candidates
        
        print(f"🖼️ 步骤2：用图搜验证 '{first_product['title'][:20]}...'")
        
        # 第二步：图搜找同款
        img_result = self.image_searcher.search_by_url(
            image_url=image_url,
            keyword=product_title.split()[0]  # 取第一个词作为辅助关键词
        )
        similar_items = self.image_searcher.parse_image_results(img_result)
        
        # 合并结果，去重
        all_items = candidates + similar_items
        seen_ids = set()
        unique_items = []
        
        for item in all_items:
            pid = item.get('product_id')
            if pid and pid not in seen_ids:
                seen_ids.add(pid)
                unique_items.append(item)
        
        # 按价格升序排列（找最便宜的源头）
        unique_items.sort(key=lambda x: float(x.get('price', 0) or 0))
        
        return unique_items
    
    def price_comparison(self, image_url: str, keyword: str) -> dict:
        """
        价格监控：图搜锁定商品 + 关键词搜索比价
        
        :param image_url: 商品图片URL
        :param keyword: 商品关键词
        :return: 比价结果
        """
        # 图搜获取同款
        img_result = self.image_searcher.search_by_url(image_url, keyword=keyword)
        image_items = self.image_searcher.parse_image_results(img_result)
        
        # 关键词搜索获取同类
        kw_result = self.keyword_searcher.search_products(
            keyword=keyword,
            page=1,
            page_size=50
        )
        keyword_items = self.keyword_searcher.parse_search_results(kw_result)
        
        # 价格分析
        all_prices = []
        for item in image_items + keyword_items:
            price = item.get('price')
            if price and isinstance(price, (int, float)):
                all_prices.append(float(price))
        
        if not all_prices:
            return {"error": "未获取到价格数据"}
        
        return {
            "image_search_count": len(image_items),
            "keyword_search_count": len(keyword_items),
            "min_price": min(all_prices),
            "max_price": max(all_prices),
            "avg_price": round(sum(all_prices) / len(all_prices), 2),
            "median_price": round(sorted(all_prices)[len(all_prices)//2], 2),
            "cheapest_item": min(image_items + keyword_items, 
                                key=lambda x: float(x.get('price', float('inf')) or float('inf')))
        }


# ========== 组合搜索实战 ==========
if __name__ == '__main__':
    APP_KEY = 'your_app_key'
    APP_SECRET = 'your_app_secret'
    
    combo = AlibabaSearchCombo(APP_KEY, APP_SECRET)
    
    # 实战1：找同款货源
    print("=" * 60)
    print("【实战1：找同款货源】")
    print("=" * 60)
    
    sources = combo.find_source_factory("无线蓝牙耳机 降噪")
    print(f"找到 {len(sources)} 个源头：")
    for i, s in enumerate(sources[:5], 1):
        print(f"{i}. {s['title'][:30]}... | ¥{s['price']} | {s['supplier']}")
    
    # 实战2：价格监控
    print("\n" + "=" * 60)
    print("【实战2：价格监控】")
    print("=" * 60)
    
    # price_analysis = combo.price_comparison(
    #     image_url="https://example.com/product.jpg",
    #     keyword="蓝牙耳机"
    # )
    # print(f"最低价格: ¥{price_analysis['min_price']}")
    # print(f"平均价格: ¥{price_analysis['avg_price']}")
    # print(f"最便宜的商品: {price_analysis['cheapest_item']['title']}")
    

七、性能优化与效率翻倍技巧

7.1 分页与批量处理

class BatchProcessor:
    """批量处理优化"""
    
    def __init__(self, searcher: AlibabaKeywordSearch):
        self.searcher = searcher
    
    def fetch_all_pages(
        self,
        keyword: str,
        max_pages: int = 10,
        page_size: int = 50,
        delay: float = 1.2
    ) -> list:
        """
        获取关键词的所有分页数据
        
        :param max_pages: 最大页数
        :param delay: 请求间隔（遵守QPS限制）
        """
        all_products = []
        
        for page in range(1, max_pages + 1):
            result = self.searcher.search_products(
                keyword=keyword,
                page=page,
                page_size=page_size
            )
            products = self.searcher.parse_search_results(result)
            
            if not products:
                break
            
            all_products.extend(products)
            print(f"  第{page}页: {len(products)}条，累计:{len(all_products)}条")
            
            # 智能延时：根据返回数据量调整
            time.sleep(delay)
        
        return all_products
        

7.2 缓存策略

import json
from functools import lru_cache

class CachedSearcher(AlibabaKeywordSearch):
    """带缓存的搜索器"""
    
    def __init__(self, app_key: str, app_secret: str, cache_file: str = "search_cache.json"):
        super().__init__(app_key, app_secret)
        self.cache_file = cache_file
        self._load_cache()
    
    def _load_cache(self):
        try:
            with open(self.cache_file, 'r', encoding='utf-8') as f:
                self.cache = json.load(f)
        except FileNotFoundError:
            self.cache = {}
    
    def _save_cache(self):
        with open(self.cache_file, 'w', encoding='utf-8') as f:
            json.dump(self.cache, f, ensure_ascii=False, indent=2)
    
    def search_products(self, **kwargs):
        cache_key = json.dumps(kwargs, sort_keys=True)
        
        if cache_key in self.cache:
            print("📦 命中缓存")
            return self.cache[cache_key]
        
        result = super().search_products(**kwargs)
        self.cache[cache_key] = result
        self._save_cache()
        return result
        

7.3 并发请求（注意QPS限制）

from concurrent.futures import ThreadPoolExecutor, as_completed

def parallel_search(searcher, keywords: list, max_workers: int = 3) -> dict:
    """
    并发搜索多个关键词（需确保不超过QPS限制）
    
    :param max_workers: 并发数，建议不超过3
    """
    results = {}
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        future_to_kw = {
            executor.submit(searcher.search_products, kw, page=1, page_size=20): kw 
            for kw in keywords
        }
        
        for future in as_completed(future_to_kw):
            kw = future_to_kw[future]
            try:
                result = future.result()
                results[kw] = searcher.parse_search_results(result)
            except Exception as e:
                results[kw] = {"error": str(e)}
    
    return results
    

八、错误处理与常见问题

8.1 常见错误码

8.2 完整错误处理封装

class ApiErrorHandler:
    """API错误处理工具"""
    
    @staticmethod
    def handle_response(response: dict) -> dict:
        """
        统一处理API响应
        """
        if 'error_response' in response:
            error = response['error_response']
            code = error.get('code', 'unknown')
            msg = error.get('msg', '未知错误')
            
            error_map = {
                'isv.invalid-signature': "签名错误，请检查App Secret和签名算法",
                'isv.permission-api-forbidden': "API权限不足，请在开放平台申请权限",
                'isp.call-limited': "调用频率超限，请降低请求频率",
                'isv.invalid-parameter': f"参数错误: {msg}",
            }
            
            return {
                'success': False,
                'error_code': code,
                'error_msg': error_map.get(code, msg),
                'raw': response
            }
        
        return {'success': True, 'data': response}
        

九、总结