from PIL import Imageimport ioimport base64def preprocess_image(image_path: str, max_size_kb: int = 500, target_size: tuple = (800, 800)) -> str:
    """
    图片预处理：压缩、裁剪、转Base64
    """
    with Image.open(image_path) as img:
        # 1. 转RGB（去除透明通道）
        if img.mode in ('RGBA', 'P'):
            img = img.convert('RGB')
        
        # 2. 等比缩放，最大边不超过 target_size
        img.thumbnail(target_size, Image.Resampling.LANCZOS)
        
        # 3. 压缩至目标大小
        quality = 95
        while True:
            buffer = io.BytesIO()
            img.save(buffer, format='JPEG', quality=quality, optimize=True)
            size_kb = buffer.tell() / 1024
            
            if size_kb <= max_size_kb or quality <= 30:
                break
            quality -= 5
        
        # 4. 转Base64，去除换行和空格
        base64_str = base64.b64encode(buffer.getvalue()).decode('utf-8')
        return base64_str.replace("\n", "").replace(" ", "")

import hashlibdef generate_1688_sign(params: dict, app_secret: str) -> str:
    """
    生成1688 MD5签名
    """
    # 过滤空值和 sign 本身
    filtered = {k: v for k, v in params.items() if v is not None and k != 'sign'}
    # 按 key ASCII 升序排序
    sorted_params = sorted(filtered.items(), key=lambda x: x[0])
    # 拼接成 key=value&key=value，值需URL编码
    import urllib.parse
    param_str = '&'.join([f"{k}={urllib.parse.quote(str(v), safe='')}" for k, v in sorted_params])
    # 首尾拼接 app_secret
    sign_str = f"{app_secret}{param_str}{app_secret}"
    # MD5 大写
    return hashlib.md5(sign_str.encode('utf-8')).hexdigest().upper()

import requestsimport hashlibimport timeimport base64import urllib.parsefrom typing import Dict, List, Optionalclass Ali1688ImageSearch:
    """
    1688 拍立淘/图搜接口封装
    """
    
    def __init__(self, app_key: str, app_secret: str):
        self.app_key = app_key
        self.app_secret = app_secret        # 网关地址以官方最新文档为准
        self.gateway = "https://gw.open.1688.com/openapi/param2/2/portals.open/api/alibaba.image.search.offer.match"
    
    def _sign(self, params: Dict) -> str:
        """MD5 签名"""
        filtered = {k: v for k, v in params.items() if v is not None and k != 'sign'}
        sorted_params = sorted(filtered.items(), key=lambda x: x[0])
        param_str = '&'.join([f"{k}={urllib.parse.quote(str(v), safe='')}" 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 search_by_image(self, 
                        image_input: str, 
                        is_base64: bool = True,
                        match_type: str = "exact",
                        sort: str = "_sale",
                        page_size: int = 20) -> Optional[Dict]:
        """
        以图搜货
        
        :param image_input: Base64字符串 或 图片URL
        :param is_base64: True表示Base64，False表示URL
        :param match_type: exact=同款, similar=相似
        :param sort: _sale=销量, _price=价格
        :param page_size: 每页条数，最大50
        """
        # 组装参数
        params = {
            "method": "alibaba.image.search.offer.match",
            "app_key": self.app_key,
            "timestamp": str(int(time.time() * 1000)),
            "format": "json",
            "v": "2.0",
            "sign_method": "md5",
            "image": image_input,
            "imageType": "base64" if is_base64 else "url",
            "matchType": match_type,
            "sortType": sort,
            "pageNum": "1",
            "pageSize": str(min(page_size, 50)),
            "needSupplyInfo": "true"  # 返回供应链字段
        }
        params["sign"] = self._sign(params)
        
        try:
            resp = requests.post(self.gateway, data=params, timeout=(5, 15))
            resp.raise_for_status()
            result = resp.json()
            
            if "error_response" in result:
                err = result["error_response"]
                raise Exception(f"接口错误 [{err.get('code')}]: {err.get('msg')}")
            
            return result.get("alibaba_image_search_offer_match_response", {}).get("result", {})
            
        except requests.exceptions.Timeout:
            print("接口超时：建议压缩图片至500KB以内")
            return None
        except Exception as e:
            print(f"请求失败: {e}")
            return None
    
    def extract_offers(self, result: Dict) -> List[Dict]:
        """解析并标准化返回结果"""
        offers = []
        raw_offers = result.get("offerList", [])
        
        for raw in raw_offers:
            # 处理批发价区间（如 "10.00-15.00" 或单值）
            price_range = raw.get("priceRange", "0.00")
            if "-" in str(price_range):
                min_price, max_price = map(float, str(price_range).split("-"))
            else:
                min_price = max_price = float(price_range)
            
            # 处理起订量（如 "10+"）
            moq_str = str(raw.get("moq", "0")).replace("+", "")
            moq = int(moq_str) if moq_str.isdigit() else 0
            
            offers.append({
                "offerId": raw.get("offerId"),
                "title": raw.get("title", ""),
                "similarity": float(raw.get("similarity", 0)),  # 相似度 0~1
                "minPrice": min_price,
                "maxPrice": max_price,
                "moq": moq,
                "supplierName": raw.get("supplierName", ""),
                "supplyType": raw.get("supplyType", ""),  # 现货/定制
                "mainImage": raw.get("mainImage", ""),
                "detailUrl": raw.get("detailUrl", "")
            })
        
        return offers# 使用示例if __name__ == "__main__":
    client = Ali1688ImageSearch("your_app_key", "your_app_secret")
    
    # 方式1：本地图片转Base64
    with open("product.jpg", "rb") as f:
        img_b64 = base64.b64encode(f.read()).decode('utf-8').replace("\n", "")
    
    result = client.search_by_image(img_b64, is_base64=True, match_type="exact")
    
    if result:
        offers = client.extract_offers(result)
        print(f"共找到 {len(offers)} 个匹配商品")
        
        # 按相似度过滤（≥0.85视为同款）
        same_style = [o for o in offers if o["similarity"] >= 0.85]
        print(f"同款商品 {len(same_style)} 个")
        
        for item in same_style[:5]:
            print(f"• {item['title']} | ¥{item['minPrice']}-{item['maxPrice']} | MOQ:{item['moq']} | 相似度:{item['similarity']:.0%}")

{
    "alibaba_image_search_offer_match_response": {
        "result": {
            "totalCount": 156,
            "pageNum": 1,
            "pageSize": 20,
            "offerList": [
                {
                    "offerId": "123456789012345",
                    "title": "2026新款 磁吸无线充电宝 10000mAh 超薄便携",
                    "mainImage": "https://cbu01.alicdn.com/...",
                    "similarity": 0.92,
                    "priceRange": "32.00-45.00",
                    "moq": "50",
                    "supplyType": "现货",
                    "supplierName": "深圳市XX电子有限公司",
                    "supplierId": "b2b-123456",
                    "detailUrl": "https://detail.1688.com/offer/123456789012345.html",
                    "bookedCount": 1860,
                    "isFactory": true
                }
            ]
        }
    }}

def optimize_supply_chain_sort(offers: List[Dict], weights: Dict = None) -> List[Dict]:
    """
    供应链匹配复合排序
    默认权重：相似度(50%) > 起订量(30%) > 供应类型(20%)
    """
    weights = weights or {"similarity": 0.5, "moq": 0.3, "supply_type": 0.2}
    
    # 计算最大 MOQ 用于归一化
    max_moq = max([o["moq"] for o in offers]) if offers else 1000
    
    for offer in offers:
        # 相似度得分（0~50分）
        sim_score = offer["similarity"] * 100 * weights["similarity"]
        
        # 起订量得分（MOQ越小得分越高，0~30分）
        moq_score = (1 - min(offer["moq"] / max_moq, 1)) * 100 * weights["moq"]
        
        # 供应类型得分（现货=20分，定制=10分）
        if offer["supplyType"] == "现货":
            type_score = 100 * weights["supply_type"]
        elif offer["supplyType"] == "定制":
            type_score = 50 * weights["supply_type"]
        else:
            type_score = 25 * weights["supply_type"]
        
        offer["compositeScore"] = round(sim_score + moq_score + type_score, 2)
    
    # 按综合得分降序，得分相同按相似度降序
    return sorted(offers, key=lambda x: (-x["compositeScore"], -x["similarity"]))

海外平台热销商品/展会样品/竞品图片
              │
              ▼
    ┌─────────────────────┐
    │  图片预处理          │
    │  (压缩→去底→转Base64) │
    └─────────────────────┘
              │
              ▼
    ┌─────────────────────┐
    │  1688 图搜接口       │
    │  (alibaba.image.search)│
    └─────────────────────┘
              │
              ▼
    ┌─────────────────────┐
    │  结果过滤            │
    │  similarity ≥ 0.85   │
    │  按 compositeScore 排序 │
    └─────────────────────┘
              │
              ▼
    ┌─────────────────────┐
    │  商品详情接口        │
    │  (alibaba.product.get) │
    │  获取 SKU/MOQ/库存/运费 │
    └─────────────────────┘
              │
              ▼
    ┌─────────────────────┐
    │  供应商背调          │
    │  诚信通年限/工厂认证/评价 │
    └─────────────────────┘
              │
              ▼
    ┌─────────────────────┐
    │  采购决策            │
    │  测款→起量→锁产能    │
    └─────────────────────┘