破解微店引流难题！商品全量接口的精准匹配引擎与推荐系统实现指南

def standardize_product_data(raw_data: dict) -> dict:
    """将微店接口原始数据标准化"""
    # 基础信息标准化
    base_info = {
        "product_id": raw_data.get("product_id"),
        "title": raw_data.get("title"),
        "price": float(raw_data.get("price", 0)),
        "stock": int(raw_data.get("stock", 0)),
        "category": get_standard_category(raw_data.get("category_name")),  # 统一类目名称
        "tags": normalize_tags(raw_data.get("tags", []))  # 标签去重、标准化（如"保湿"合并"补水"）
    }
    # 新增"放心选"维度（对接微店货源市场标准）
    trust_info = {
        "quality_grade": raw_data.get("quality_grade", "C"),  # 微店质量评级
        "return_support": raw_data.get("return_policy") == "支持7天无理由",
        "supplier_credit": get_supplier_credit(raw_data.get("supplier_id"))  # 供应商信用分
    }
    # 补充营销维度
    marketing_info = {
        "is_on_sale": raw_data.get("status") == "上架",
        "sale_count": int(raw_data.get("sales", 0)),
        "share_rate": calculate_share_rate(raw_data.get("product_id"))  # 社交分享转化率
    }
    return {**base_info, **trust_info, **marketing_info}

class ProductMatcher:
    def __init__(self, standardized_products: list):
        # 构建倒排索引（关键词→商品ID列表）
        self.inverted_index = self._build_inverted_index(standardized_products)
        # 构建属性索引（属性键值对→商品ID列表）
        self.attribute_index = self._build_attribute_index(standardized_products)
        # 加载意图词典（微店高频需求场景）
        self.intent_dict = self._load_intent_dict()
    def _build_inverted_index(self, products: list) -> dict:
        """构建关键词倒排索引，支持同义词扩展"""
        index = defaultdict(list)
        for product in products:
            # 提取标题、标签中的关键词
            keywords = extract_keywords(product["title"] + " " + " ".join(product["tags"]))
            # 同义词扩展（如"保湿"→["保湿","补水"]）
            expanded_keywords = expand_synonyms(keywords)
            for kw in expanded_keywords:
                index[kw].append(product["product_id"])
        return index
    def match(self, user_query: str, filters: dict = None) -> list:
        """
        双模式匹配：先意图解析，再关键词+属性过滤
        user_query: 用户自然语言查询（如"敏感肌能用的保湿面膜"）
        filters: 额外筛选条件（如{"price_max": 200, "quality_grade": "A"}）
        """
        # 1. 意图解析（提取核心需求和属性）
        intent = self._parse_intent(user_query)  # 如{"category": "面膜", "attribute": "敏感肌友好", "function": "保湿"}
        # 2. 关键词匹配（基于意图核心词）
        core_kw = intent.get("category", "") + " " + intent.get("function", "")
        candidate_ids = self._keyword_match(core_kw)
        # 3. 属性过滤（结合意图属性和用户筛选条件）
        final_ids = self._attribute_filter(candidate_ids, {**intent.get("attribute", {}), **(filters or {})})
        # 4. 返回匹配结果（按相关性排序）
        return self._get_sorted_products(final_ids, user_query)

class SmartRecommender:
    def __init__(self, product_data: list, user_behavior: list):
        self.products = {p["product_id"]: p for p in product_data}
        self.user_behavior = user_behavior  # 包含浏览、收藏、购买、分享等行为
        # 预计算商品相似度矩阵（基于协同过滤）
        self.product_similarity = self._calculate_similarity()
    def recommend(self, user_id: str, scene: str) -> list:
        """
        场景化推荐：支持首页、详情页、活动页等不同场景
        scene: 推荐场景（"home"→首页，"product_detail"→详情页关联，"activity"→活动搭配）
        """
        # 1. 获取用户画像（基于历史行为）
        user_profile = self._build_user_profile(user_id)  # 如{"偏好类目": ["面膜","精华"], "价格敏感度": "中", "分享倾向": True}
        # 2. 场景化推荐策略
        if scene == "home":
            # 首页：热门+偏好+放心选权重
            candidates = self._hybrid_recommend(
                user_profile, 
                hot_weight=0.3,  # 热门商品权重
                preference_weight=0.5,  # 个人偏好权重
                trust_weight=0.2  # 放心选权重
            )
        elif scene == "product_detail":
            # 详情页：关联商品（基于相似度）
            current_product_id = self._get_current_viewed_product(user_id)
            candidates = self._similar_product_recommend(current_product_id)
        elif scene == "activity":
            # 活动页：搭配商品（基于历史搭配数据）
            activity_type = self._get_activity_type()
            candidates = self._collocation_recommend(activity_type, user_profile)
        # 3. 过滤规则（去重、排除滞销、库存不足）
        filtered_candidates = self._filter_candidates(candidates, user_id)
        # 4. 返回Top10推荐
        return [self.products[pid] for pid in filtered_candidates[:10]]

┌─────────────┐  优先命中  ┌─────────────┐
│  本地缓存   │◄─────────►│  Redis缓存  │
│（内存，30s）│           │（10分钟）   │
└─────────────┘           └─────────────┘

import requests
def get_access_token(shop_id: str, app_key: str, app_secret: str) -> str:
    resp = requests.post(
        "https://open.weidian.com/api/v2/token/get",
        json={
            "shop_id": shop_id,
            "app_key": app_key,
            "app_secret": app_secret,
            "grant_type": "client_credentials"
        },
        timeout=5
    )
    return resp.json()["access_token"]