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

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)} 条商品")

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)

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

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