微店关键词搜索接口核心突破：动态权重算法与语义引擎的实战落地

class WeidianSearchAPI:
    def __init__(self, app_key: str, app_secret: str, shop_id: str):
        self.app_key = app_key
        self.app_secret = app_secret
        self.shop_id = shop_id
        self.api_url = "https://api.weidian.com/v2/search/items"
        self.session = self._init_session()  # 连接池优化
        self.weights = {"keyword_match": 0.4, "sales": 0.2, "rating": 0.15, 
                       "price": 0.1, "stock": 0.05, "user_preference": 0.1}  # 权重配置
    def _generate_signature(self, params: Dict) -> str:
        """按微店规范生成签名，防止请求篡改"""
        sorted_params = sorted(params.items(), key=lambda x: x[0])
        sign_str = f"{self.app_secret}" + "".join([f"{k}{v}" for k, v in sorted_params if v]) + f"{self.app_secret}"
        return hashlib.md5(sign_str.encode('utf-8')).hexdigest().upper()
    def basic_search(self, keyword: str, **kwargs) -> Tuple[List[Dict], int]:
        """基础搜索实现，支持多条件筛选与错误处理"""
        params = {"app_key": self.app_key, "shop_id": self.shop_id, "keyword": keyword,
                  "timestamp": int(time.time() * 1000), "format": "json"}
        params.update({k: v for k, v in kwargs.items() if k in ["page", "size", "sort", "category_id"]})
        params["sign"] = self._generate_signature(params)
        try:
            response = self.session.get(self.api_url, params=params, timeout=(5, 15))
            response.raise_for_status()
            result = response.json()
            if result.get("status") != 0:
                logger.error(f"接口错误[{result.get('status')}]: {result.get('message')}")
                return [], 0
            data = result.get("data", {})
            return data.get("items", []), data.get("total", 0)
        except requests.exceptions.RequestException as e:
            logger.error(f"请求异常: {str(e)}")
            return [], 0

def _preprocess_keyword(self, keyword: str) -> str:
    """去除噪声词，提升匹配精度"""
    keyword = re.sub(r'[^\u4e00-\u9fa5a-zA-Z0-9]', ' ', keyword)  # 过滤特殊字符
    return " ".join([w for w in jieba.lcut(keyword) if w not in self.stopwords])
def _expand_keywords(self, keyword: str) -> List[str]:
    """基于同义词表扩展关键词，如“手机”→["手机","智能机","移动电话"]"""
    expanded = [keyword]
    for word in keyword.split():
        expanded.extend(self.synonym_dict.get(word, []))
    return list(set(expanded))[:3]  # 限制扩展数量，平衡性能与召回率

def _classify_search_intent(self, keyword: str) -> str:
    if any(cat in keyword for cat in ["衣服", "手机", "食品"]):
        return "category"  # 品类搜索，优先展示全品类商品
    elif any(feat in keyword for feat in ["特价", "新款", "正品"]):
        return "feature"  # 特征搜索，强化价格/新品权重
    return "product"  # 产品搜索，侧重关键词匹配度

def _rank_items_by_semantic_similarity(self, items: List[Dict], keyword: str, intent: str, user_id: Optional[str]) -> List[Dict]:
    """多维度评分排序，综合匹配度、热度、偏好等因子"""
    scored_items = []
    for item in items:
        scores = {
            "keyword_match": self._calculate_match_score(item, keyword),  # TF-IDF相似度计算
            "sales": self._normalize_score(item.get("sales", 0), 0, 1000),  # 销量标准化
            "rating": self._normalize_score(item.get("rating", 3), 1, 5),  # 评分标准化
            "price": self._calculate_price_score(item, intent, user_id),  # 价格偏好适配
            "stock": self._normalize_score(item.get("stock", 0), 0, 100),  # 库存充足度
            "user_preference": self._calculate_preference_score(item, user_id)  # 个性化偏好
        }
        total_score = sum(scores[k] * self.weights[k] for k in scores)
        scored_items.append({"item": item, "total_score": total_score})
    return [x["item"] for x in sorted(scored_items, key=lambda x: x["total_score"], reverse=True)]

def _update_user_preferences(self, user_id: str) -> None:
    """每10次搜索更新一次偏好模型，涵盖价格区间、品类偏好"""
    clicked_items = self._get_user_clicked_items(user_id)
    if not clicked_items:
        return
    # 分析价格偏好（四分位区间）
    prices = [item["price"] for item in clicked_items]
    self.user_preferences[user_id]["price_preference"] = (np.percentile(prices, 25), np.percentile(prices, 75))
    # 分析品类偏好
    category_counts = defaultdict(int)
    for item in clicked_items:
        category_counts[item["category_id"]] += 1
    total = sum(category_counts.values())
    self.user_preferences[user_id]["category_preferences"] = {k: v/total for k, v in category_counts.items()}