1688 拍立淘接口实战：从图像优化、工厂排序到供应链匹配（附可跑代码）

import base64from PIL import Imageimport ioimport numpy as npdef preprocess_image(image_path: str, target_size: tuple = (800, 800)) -> str:    """    图像预处理：统一分辨率、压缩大小、生成标准Base64编码    :param image_path: 本地图像路径    :param target_size: 目标分辨率（1688推荐800*800，兼顾精度与速度）    :return: 标准Base64编码字符串    """    try:        # 1. 读取图像并统一分辨率（避免因尺寸差异导致识别偏差）        with Image.open(image_path) as img:            # 保持宽高比缩放，避免拉伸变形            img.thumbnail(target_size, Image.Resampling.LANCZOS)            # 补白至目标尺寸（1688要求图像为正方形，否则裁剪边缘）            background = Image.new('RGB', target_size, (255, 255, 255))            offset = ((target_size[0] - img.size[0]) // 2, (target_size[1] - img.size[1]) // 2)            background.paste(img, offset)                        # 2. 压缩图像大小（控制在500KB内，避免接口超时）            img_byte_arr = io.BytesIO()            # 根据图像类型调整压缩质量（jpg用80%，png用无损）            if image_path.lower().endswith('.png'):                background.save(img_byte_arr, format='PNG', optimize=True)            else:                background.save(img_byte_arr, format='JPEG', quality=80)            img_byte_arr = img_byte_arr.getvalue()                        # 3. 生成Base64编码（关键：去掉换行符，否则1688接口报1001错误）            base64_str = base64.b64encode(img_byte_arr).decode('utf-8').replace('\n', '')            return base64_str    except Exception as e:        print(f"图像预处理失败：{str(e)}（常见原因：图像损坏、格式不支持）")        return ""# 测试示例if __name__ == "__main__":    base64_img = preprocess_image("test_shirt.jpg")    print(f"预处理后Base64长度：{len(base64_img)}（建议500KB内，对应Base64长度约680000字符）")

import timeimport hashlibimport requestsimport jsonfrom typing import Dict, List, Optionalclass Ali1688ImageSearchAPI:    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/portals.open/api/alibaba.image.search.offer.match"        self.session = self._init_session()    def _init_session(self) -> requests.Session:        """初始化会话：图像接口耗时久，设3次重试+长超时"""        session = requests.Session()        adapter = requests.adapters.HTTPAdapter(            pool_connections=10, pool_maxsize=50, max_retries=3        )        session.mount('https://', adapter)        return session    def _generate_sign(self, params: Dict) -> str:        """生成1688签名：参数排序+URL编码，中文不编码必错"""        # 1. 过滤空值并按ASCII升序排序        valid_params = {k: v for k, v in params.items() if v is not None}        sorted_params = sorted(valid_params.items(), key=lambda x: x[0])        # 2. 拼接参数串（每个值需URL编码，避免特殊字符影响签名）        sign_str = '&'.join([f"{k}={requests.utils.quote(str(v), safe='')}" for k, v in sorted_params])        # 3. 首尾加app_secret，MD5加密转大写        sign_str = self.app_secret + sign_str + self.app_secret        return hashlib.md5(sign_str.encode('utf-8')).hexdigest().upper()    def search_offer_by_image(self, base64_img: str, similarity: float = 0.75, page_num: int = 1) -> Optional[Dict]:        """        图像匹配商品：核心接口调用逻辑        :param base64_img: 预处理后的图像Base64编码        :param similarity: 相似度阈值（0.7-0.8最优）        :param page_num: 页码（最大20页）        :return: 结构化识别结果        """        if len(base64_img) == 0:            print("图像Base64编码为空，无法调用接口")            return None        # 1. 构建请求参数        params = {            "method": "alibaba.image.search.offer.match",            "app_key": self.app_key,            "timestamp": str(int(time.time() * 1000)),  # 13位毫秒级时间戳            "format": "json",            "v": "2.0",            "sign_method": "md5",            "image": base64_img,            "similarity": str(similarity),            "pageNum": str(page_num),            "pageSize": "20",  # 最大20，不可调整            "needSupplyInfo": "true"  # 返回供应链字段（起订量、批发价等）        }        params["sign"] = self._generate_sign(params)        try:            # 2. 发送请求（图像接口耗时较长，超时设15秒）            response = self.session.get(self.api_url, params=params, timeout=(5, 15))            result = response.json()            # 3. 处理接口错误            if "error_response" in result:                err_code = result["error_response"]["code"]                err_msg = result["error_response"]["msg"]                print(f"接口调用失败：{err_msg}（错误码：{err_code}）")                # 常见错误处理建议                if err_code == 1001:                    print("可能原因：Base64编码含换行符、图像格式不支持")                elif err_code == 429:                    print("可能原因：QPS超限，建议控制在2次/秒内")                return None            # 4. 解析结构化结果（处理供应链字段格式差异）            raw_result = result["alibaba_image_search_offer_match_response"]["result"]            return self._parse_search_result(raw_result)        except requests.exceptions.Timeout:            print("接口超时：建议压缩图像大小（控制在500KB内）")            return None        except json.JSONDecodeError:            print("返回数据解析失败：可能是图像过大导致响应不完整")            return None        except Exception as e:            print(f"未知错误：{str(e)}")            return None    def _parse_search_result(self, raw_result: Dict) -> Dict:        """解析识别结果：处理供应链字段的格式差异"""        parsed_offers = []        for raw_offer in raw_result.get("offerList", []):            # 处理批发价区间（部分返回"10.00-15.00"，部分返回单值）            price_range = raw_offer.get("priceRange", "0.00")            if "-" in price_range:                min_price = float(price_range.split("-")[0])                max_price = float(price_range.split("-")[1])            else:                min_price = max_price = float(price_range)            # 处理起订量（部分返回"10+"，需截取数字）            moq_str = raw_offer.get("moq", "0")            moq = int(moq_str.replace("+", "")) if moq_str else 0            parsed_offers.append({                "offerId": raw_offer.get("offerId", ""),                "title": raw_offer.get("title", ""),                "similarity": float(raw_offer.get("similarity", 0)),                "minPrice": min_price,                "maxPrice": max_price,                "moq": moq,                "supplierName": raw_offer.get("supplierName", ""),                "supplyType": raw_offer.get("supplyType", "")  # 现货/定制            })        return {            "totalCount": int(raw_result.get("totalCount", 0)),            "pageNum": int(raw_result.get("pageNum", 1)),            "totalPages": (int(raw_result.get("totalCount", 0)) + 19) // 20,            "offers": parsed_offers        }

def optimize_supply_chain_sort(offers: List[Dict], priority_weights: Dict = None) -> List[Dict]:    """    供应链匹配排序：结合相似度、起订量、供应类型的复合排序    :param offers: 原始识别结果列表    :param priority_weights: 权重配置（可根据业务调整）    :return: 排序后结果    """    # 默认权重：相似度(0.5) > 起订量(0.3) > 供应类型(0.2)    weights = priority_weights or {        "similarity": 0.5,        "moq": 0.3,        "supplyType": 0.2    }    # 计算每个商品的综合得分（0-100分）    for offer in offers:        # 1. 相似度得分（0-1 → 0-50分）        similarity_score = offer["similarity"] * 100 * weights["similarity"]        # 2. 起订量得分（起订量越小得分越高，0-30分）        max_moq = max([o["moq"] for o in offers]) if offers else 1000        moq_score = (1 - min(offer["moq"] / max_moq, 1)) * 100 * weights["moq"]        # 3. 供应类型得分（现货=20分，定制=10分，其他=5分）        if offer["supplyType"] == "现货":            supply_score = 100 * weights["supplyType"]        elif offer["supplyType"] == "定制":            supply_score = 50 * weights["supplyType"]        else:            supply_score = 25 * weights["supplyType"]        # 综合得分        offer["compositeScore"] = round(similarity_score + moq_score + supply_score, 2)    # 按综合得分降序排序（得分相同按相似度排序）    return sorted(offers, key=lambda x: (-x["compositeScore"], -x["similarity"]))# 测试示例if __name__ == "__main__":    # 初始化API客户端（替换为自己的app_key和app_secret）    api = Ali1688ImageSearchAPI("your_app_key", "your_app_secret")    # 预处理图像    base64_img = preprocess_image("test_shirt.jpg")    # 调用接口    search_result = api.search_offer_by_image(base64_img)    if search_result and search_result["offers"]:        # 优化排序        sorted_offers = optimize_supply_chain_sort(search_result["offers"])        # 输出结果        print("排序后Top3商品：")        for idx, offer in enumerate(sorted_offers[:3], 1):            print(f"第{idx}名：得分{offer['compositeScore']} | 标题：{offer['title']} | 起订量：{offer['moq']}件 | 相似度：{offer['similarity']}")