1688 图片搜图找货接口开发实战：从图像特征提取到商品匹配全流程

import base64from PIL import Imageimport iodef process_image(image_path, max_size=(800, 800), quality=85):
    """
    预处理图片并转换为Base64编码
    :param image_path: 图片路径
    :param max_size: 最大尺寸（宽, 高）
    :param quality: 压缩质量（1-100）
    :return: Base64编码字符串
    """
    try:
        # 打开图片
        with Image.open(image_path) as img:
            # 转换为RGB模式（处理透明背景）
            if img.mode in ('RGBA', 'LA'):
                background = Image.new(img.mode[:-1], img.size, (255, 255, 255))
                background.paste(img, img.split()[-1])
                img = background            
            # 按比例缩放
            img.thumbnail(max_size)
            
            # 保存到内存缓冲区
            buffer = io.BytesIO()
            img.save(buffer, format='JPEG', quality=quality)
            
            # 转换为Base64编码
            base64_str = base64.b64encode(buffer.getvalue()).decode('utf-8')
            
            return base64_str            
    except Exception as e:
        print(f"图片处理失败: {str(e)}")
        return None

import timeimport hashlibimport urllib.parseimport requestsimport jsonfrom typing import Dict, Optional, Anydef generate_1688_sign(params, app_secret):
    """生成1688接口签名"""
    # 1. 按参数名ASCII升序排序
    sorted_params = sorted(params.items(), key=lambda x: x[0])
    
    # 2. 拼接为key=value&key=value格式
    query_string = urllib.parse.urlencode(sorted_params)
    
    # 3. 拼接appsecret并MD5加密
    sign_str = f"{query_string}{app_secret}"
    sign = hashlib.md5(sign_str.encode('utf-8')).hexdigest().upper()
    
    return signclass AlibabaImageSearchAPI:
    def __init__(self, app_key: str, app_secret: str):
        self.app_key = app_key
        self.app_secret = app_secret
        self.api_url = "https://gw.open.1688.com/openapi/param2/2.0/"
        
    def search_by_image(self, 
                       image_base64: str = None,
                       image_url: str = None,
                       page: int = 1,
                       page_size: int = 20,** kwargs) -> Optional[Dict[str, Any]]:
        """
        以图搜货接口调用
        :param image_base64: 图片Base64编码
        :param image_url: 图片URL（与image_base64二选一）
        :param page: 页码
        :param page_size: 每页条数
        :param kwargs: 其他筛选参数
        :return: 搜索结果字典
        """
        # 校验图片参数
        if not image_base64 and not image_url:
            print("必须提供图片Base64编码或图片URL")
            return None
            
        # 1. 构建基础参数
        params = {
            "method": "alibaba.image.search",
            "app_key": self.app_key,
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
            "format": "json",
            "v": "2.0",
            "page": page,
            "page_size": page_size,
            **kwargs        }
        
        # 2. 添加图片参数
        if image_base64:
            params["image"] = image_base64        else:
            params["image_url"] = image_url        
        # 3. 生成签名
        params["sign"] = generate_1688_sign(params, self.app_secret)
        
        try:
            # 4. 发送请求
            response = requests.get(
                self.api_url,
                params=params,
                timeout=30  # 图片搜索耗时较长，设置较长超时
            )
            
            # 5. 处理响应
            response.raise_for_status()
            result = json.loads(response.text)
            
            # 6. 错误处理
            if "error_response" in result:
                error = result["error_response"]
                print(f"接口错误: {error.get('msg')} (错误码: {error.get('code')})")
                return None
                
            return result.get("alibaba_image_search_response", {}).get("result", {})
            
        except requests.exceptions.RequestException as e:
            print(f"HTTP请求异常: {str(e)}")
            return None
        except json.JSONDecodeError as e:
            print(f"JSON解析错误: {str(e)}")
            return None

import pandas as pddef parse_image_search_result(raw_result: Dict) -> tuple[Optional[pd.DataFrame], Optional[Dict]]:
    """
    解析图片搜索结果
    :param raw_result: 接口返回的原始数据
    :return: 商品DataFrame和统计信息
    """
    if not raw_result or "products" not in raw_result:
        return None, None
        
    # 解析商品列表
    products = []
    for item in raw_result["products"]:
        supplier = item.get("supplier", {})
        product_info = {
            "product_id": item.get("product_id"),
            "title": item.get("title"),
            "main_image": item.get("main_image"),
            "price": item.get("price"),
            "min_buy": item.get("min_buy", 1),
            "similarity": item.get("similarity", 0),
            "supplier_id": supplier.get("user_id"),
            "company_name": supplier.get("company_name"),
            "credit_level": supplier.get("credit_level", 0)
        }
        products.append(product_info)
    
    # 转换为DataFrame
    df = pd.DataFrame(products)
    
    # 解析统计信息
    stats = {
        "total": raw_result.get("total", 0),
        "page": raw_result.get("page", 1),
        "page_size": raw_result.get("page_size", 20),
        "total_page": (raw_result.get("total", 0) + raw_result.get("page_size", 20) - 1) 
                      // raw_result.get("page_size", 20)
    }
    
    return df, statsdef filter_best_suppliers(df: pd.DataFrame, min_credit: int = 4, min_similarity: float = 0.8) -> pd.DataFrame:
    """
    筛选优质供应商商品
    :param df: 商品DataFrame
    :param min_credit: 最低信用等级
    :param min_similarity: 最低相似度
    :return: 筛选后的DataFrame
    """
    if df.empty:
        return df        
    # 转换数据类型
    df["similarity"] = df["similarity"].astype(float)
    df["credit_level"] = df["credit_level"].astype(int)
    df["price"] = df["price"].astype(float)
    df["min_buy"] = df["min_buy"].astype(int)
    
    # 筛选条件
    filtered = df[
        (df["credit_level"] >= min_credit) &
        (df["similarity"] >= min_similarity)
    ]
    
    # 按相似度和价格排序
    return filtered.sort_values(
        by=["similarity", "price"], 
        ascending=[False, True]
    ).reset_index(drop=True)

if __name__ == "__main__":
    # 配置应用信息（替换为实际值）
    APP_KEY = "your_app_key_here"
    APP_SECRET = "your_app_secret_here"
    
    # 初始化API客户端
    image_search_api = AlibabaImageSearchAPI(APP_KEY, APP_SECRET)
    
    # 图片处理（可以是本地图片或网络图片URL）
    image_path = "test_product.jpg"  # 本地测试图片
    image_base64 = process_image(image_path)
    
    # 或者使用图片URL
    # image_url = "https://example.com/product.jpg"
    
    if image_base64:
        # 调用图片搜索接口
        print("正在进行以图搜货...")
        search_result = image_search_api.search_by_image(
            image_base64=image_base64,
            page=1,
            page_size=20,
            price_from=10,
            price_to=100,
            is_wholesale=1,  # 只看支持批发的商品
            sort=0  # 按相似度排序
        )
        
        # 解析结果
        if search_result:
            products_df, stats = parse_image_search_result(search_result)
            
            if products_df is not None and not products_df.empty:
                print(f"\n搜索完成，共找到 {stats['total']} 件匹配商品")
                
                # 筛选优质供应商商品
                best_products = filter_best_suppliers(products_df)
                print(f"筛选出 {len(best_products)} 件优质商品（信用等级≥4，相似度≥0.8）")
                
                # 显示前5条结果
                print("\n前5条匹配结果:")
                print(best_products[["title", "price", "min_buy", "similarity", "company_name", "credit_level"]].head())
                
                # 保存结果到CSV
                products_df.to_csv("image_search_results.csv", index=False, encoding="utf-8-sig")
                print("\n完整结果已保存至 image_search_results.csv")
    else:
        print("图片处理失败，无法进行搜索")

def advanced_image_processing(image_path):
    """高级图像预处理，提升搜索准确率"""
    try:
        with Image.open(image_path) as img:
            # 1. 裁剪：去除边缘无关区域（假设商品在中心）
            width, height = img.size
            crop_size = min(width, height)
            left = (width - crop_size) // 2
            top = (height - crop_size) // 2
            img = img.crop((left, top, left + crop_size, top + crop_size))
            
            # 2. 增强对比度
            from PIL import ImageEnhance
            enhancer = ImageEnhance.Contrast(img)
            img = enhancer.enhance(1.2)  # 增加20%对比度
            
            # 3. 转为Base64
            buffer = io.BytesIO()
            img.save(buffer, format='JPEG', quality=90)
            return base64.b64encode(buffer.getvalue()).decode('utf-8')
            
    except Exception as e:
        print(f"高级图像处理失败: {str(e)}")
        return process_image(image_path)  # 降级处理

import redisimport picklefrom datetime import timedeltaclass CachedImageSearchAPI(AlibabaImageSearchAPI):
    def __init__(self, app_key, app_secret, redis_host="localhost", redis_port=6379):
        super().__init__(app_key, app_secret)
        self.redis = redis.Redis(host=redis_host, port=redis_port, db=0)
        
    def get_image_hash(self, image_base64):
        """生成图片内容的哈希值作为缓存键"""
        return hashlib.md5(image_base64.encode('utf-8')).hexdigest()
        
    def search_with_cache(self, image_base64=None, image_url=None, expire=3600, **kwargs):
        """带缓存的图片搜索"""
        # 生成缓存键
        if image_base64:
            cache_key = f"1688_image_search:{self.get_image_hash(image_base64)}"
        elif image_url:
            cache_key = f"1688_image_search_url:{hashlib.md5(image_url.encode()).hexdigest()}"
        else:
            return None
            
        # 尝试从缓存获取
        cached_data = self.redis.get(cache_key)
        if cached_data:
            return pickle.loads(cached_data)
            
        # 调用接口
        result = self.search_by_image(image_base64, image_url,** kwargs)
        
        # 存入缓存（图片搜索结果相对稳定，可设置较长缓存时间）
        if result:
            self.redis.setex(cache_key, timedelta(seconds=expire), pickle.dumps(result))
            
        return result

from concurrent.futures import ThreadPoolExecutor, as_completeddef batch_image_search(api_client, image_paths, max_workers=3):
    """批量图片搜索"""
    results = {}
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        # 提交所有搜索任务
        futures = {}
        for path in image_paths:
            img_base64 = process_image(path)
            if img_base64:
                future = executor.submit(
                    api_client.search_with_cache,
                    image_base64=img_base64,
                    page=1,
                    page_size=10
                )
                futures[future] = path        
        # 处理结果
        for future in as_completed(futures):
            path = futures[future]
            try:
                result = future.result()
                if result:
                    results[path] = result                    print(f"图片 {path} 搜索完成，找到 {result.get('total', 0)} 件商品")
            except Exception as e:
                print(f"图片 {path} 搜索失败: {str(e)}")
    
    return results