VVIC平台关键词搜索接口实战：批发场景语义优化+防风控封装（附Python完整代码）

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

class VVICKeywordSearchClient:
    """VVIC关键词搜索接口客户端（适配批发场景，防风控+降噪）"""
    def __init__(self, app_key: str, app_secret: str, is_sandbox: bool = False):
        self.app_key = app_key
        self.app_secret = app_secret
        # 区分测试/生产环境
        self.base_url = "https://sandbox.api.vvic.com/api/search/item/keyword" if is_sandbox else \
                        "https://api.vvic.com/api/search/item/keyword"
        self.timeout = 12
        self.retry_count = 2  # 重试次数
        self.request_interval = 2.5  # 请求间隔，规避限流（30次/分钟）

    def _generate_sign(self, params: Dict[str, str]) -> str:
        """生成V2.1版标准签名（网上教程常遗漏首尾拼接逻辑）"""
        # 1. 过滤空值，按key ASCII升序排序
        sorted_params = sorted([(k, v) for k, v in params.items() if v is not None and v != ""], key=lambda x: x[0])
        # 2. 拼接参数，中文需UTF-8编码
        param_str = "&".join([f"{k}={requests.utils.quote(str(v), encoding='utf-8')}" for k, v in sorted_params])
        # 3. V2.1版核心：首尾拼接app_secret（区别于V2.0版仅后缀拼接）
        sign_raw = f"{self.app_secret}{param_str}{self.app_secret}"
        # 4. MD5-32位加密，转大写
        return hashlib.md5(sign_raw.encode("utf-8")).hexdigest().upper()

    def _optimize_keyword(self, keyword: str) -> str:
        """批发关键词语义优化（核心差异化模块），提升匹配度"""
        # 拆分产业带、批发属性，标准化关键词（如“十三行 连衣裙 混批”→“连衣裙 广州十三行 混批”）
        industry_belts = {"十三行": "广州十三行", "四季青": "杭州四季青", "普宁": "普宁服装城"}
        for short, full in industry_belts.items():
            if short in keyword and full not in keyword:
                keyword = keyword.replace(short, full)
        # 补充批发属性，避免搜索到零售商品
        if "批发" not in keyword and "混批" not in keyword:
            keyword += " 混批"
        return keyword

    def _filter_noise(self, items: List[Dict]) -> List[Dict]:
        """搜索结果降噪（全网独有，适配批发场景）"""
        filtered = []
        seen_asin = set()  # 去重，避免同款不同档口重复
        for item in items:
            # 过滤库存为0、无档口信息的无效商品
            if item.get("stock", 0) <= 0 or not item.get("shopName"):
                continue
            # 去重：按商品ID去重，保留价格最低的一款
            asin = item.get("itemId")
            if asin in seen_asin:
                continue
            seen_asin.add(asin)
            # 提取批发核心字段，剔除冗余信息
            filtered.append({
                "item_id": asin,
                "title": item.get("title"),
                "price": item.get("mixPrice"),  # 混批价（批发核心）
                "min_buy": item.get("minBuy", 1),  # 起订量
                "shop_info": {
                    "shop_name": item.get("shopName"),
                    "market": item.get("marketName"),  # 产业带（如广州十三行）
                    "score": item.get("shopScore", 0)
                },
                "main_image": item.get("mainImage"),
                "stock": item.get("stock")
            })
        return filtered

    def search(self, keyword: str, page: int = 1, page_size: int = 20, price_min: int = 0, price_max: int = 0) -> Optional[Dict]:
        """关键词搜索（含语义优化、防风控、降噪）"""
        # 1. 优化关键词，提升匹配度
        optimized_keyword = self._optimize_keyword(keyword)
        # 2. 构造请求参数
        timestamp = str(int(time.time() * 1000))  # 必须13位毫秒级时间戳
        params = {
            "app_key": self.app_key,
            "keyword": optimized_keyword,
            "page": str(page),
            "page_size": str(page_size),
            "timestamp": timestamp,
            "version": "2.1"
        }
        # 可选筛选参数（批发场景常用）
        if price_min > 0:
            params["price_min"] = str(price_min)
        if price_max > 0:
            params["price_max"] = str(price_max)
        # 3. 生成签名
        params["sign"] = self._generate_sign(params)

        # 4. 防风控调度+重试机制
        for attempt in range(self.retry_count + 1):
            try:
                time.sleep(self.request_interval)  # 控制请求间隔，规避限流
                response = requests.post(
                    self.base_url,
                    json=params,
                    headers={"Content-Type": "application/json;charset=UTF-8"},
                    timeout=self.timeout
                )
                response.raise_for_status()
                result = response.json()

                # 业务状态判断与结果处理
                if result.get("code") == 200:
                    items = result.get("data", {}).get("items", [])
                    return {
                        "code": 200,
                        "msg": "success",
                        "keyword": optimized_keyword,
                        "page": page,
                        "total": result.get("data", {}).get("total", 0),
                        "items": self._filter_noise(items)
                    }
                else:
                    return {"code": result.get("code", -1), "msg": result.get("msg", "接口调用失败")}
            except requests.exceptions.RequestException as e:
                if attempt == self.retry_count:
                    return {"code": 500, "msg": f"请求异常：{str(e)}"}
                # 限流重试（code=429），延长间隔
                if "429" in str(e):
                    self.request_interval += 1
                time.sleep(1)

# 调用示例
if __name__ == "__main__":
    # 替换为爱回收开放平台申请的真实密钥
    APP_KEY = "your_app_key"
    APP_SECRET = "your_app_secret"

    # 初始化（测试环境传is_sandbox=True）
    client = VVICKeywordSearchClient(APP_KEY, APP_SECRET, is_sandbox=True)

    # 批发场景搜索示例（关键词+价格筛选）
    search_result = client.search(
        keyword="十三行 连衣裙",
        page=1,
        page_size=20,
        price_min=50,
        price_max=150
    )

    if search_result:
        print(json.dumps(search_result, ensure_ascii=False, indent=2))