苏宁搜索接口深析：全品类智能分轨如何解决 O2O 电商的搜索痛点？

import re
import redis
import numpy as np
from sklearn.feature_extraction.text import TfidfVectorizer
class SuningCategoryRouter:
    def __init__(self):
        # 初始化Redis（缓存品类词典与库存数据）
        self.redis = redis.Redis(host="localhost", port=6379, db=8)
        # 构建全品类参数词典（核心差异化组件）
        self.category_vocab = self._build_category_vocab()
        # 品类识别模型
        self.vectorizer = TfidfVectorizer()
        self._init_category_model()
    def _build_category_vocab(self) -> dict:
        """按苏宁核心品类构建参数与场景词典"""
        return {
            "3C数码": {
                "params": {"尺寸": ["寸", "mm"], "分辨率": ["4K", "1080P"], "内存": ["GB", "内存"]},
                "scenes": ["快充", "5G", "游戏性能"]
            },
            "生鲜食品": {
                "params": {"重量": ["kg", "斤"], "保质期": ["天", "月"]},
                "scenes": ["当日达", "有机", "冷链"]
            },
            "大家电": {
                "params": {"功率": ["W", "千瓦"], "能效": ["一级", "二级"]},
                "scenes": ["以旧换新", "上门安装"]
            }
        }
    def _init_category_model(self):
        """预训练品类识别模型"""
        category_samples = [
            "55寸 4K HDR 智能电视", "12GB+256GB 5G 快充手机",  # 3C数码
            "2kg 有机草莓 当日达", "10斤 东北大米 保质期6个月",  # 生鲜食品
            "1.5匹 一级能效 空调 上门安装", "8kg 变频 洗衣机"     # 大家电
        ]
        self.vectorizer.fit(category_samples)
    def _category_recognition(self, query: str) -> str:
        """基于TF-IDF的品类识别（核心分轨入口）"""
        query_vec = self.vectorizer.transform([query])
        max_score = 0
        matched_category = "通用"
        for category, data in self.category_vocab.items():
            # 计算关键词与品类特征的匹配度
            feature_words = sum(data["params"].values(), []) + data["scenes"]
            feature_str = " ".join(feature_words)
            feature_vec = self.vectorizer.transform([feature_str])
            score = np.dot(query_vec.toarray(), feature_vec.toarray().T)[0][0]
            if score > max_score:
                max_score = score
                matched_category = category
        return matched_category
    def advanced_route_process(self, query: str) -> dict:
        """全品类分轨处理：输出品类+参数+场景结构化结果"""
        category = self._category_recognition(query)
        # 提取对应品类的参数与场景
        vocab = self.category_vocab.get(category, {"params": {}, "scenes": []})
        # 参数提取（以3C数码为例）
        params = {}
        for param, synonyms in vocab["params"].items():
            pattern = rf"(\d+[a-zA-Z%]?|[\u4e00-\u9fa5]+)({param}|{'|'.join(synonyms)})"
            match = re.search(pattern, query)
            if match:
                params[param] = match.group(1)
        # 场景识别
        scenes = [scene for scene in vocab["scenes"] if scene in query]
        return {
            "query": query,
            "category": category,
            "technical_params": params,
            "o2o_scenes": scenes  # 如"当日达""门店自提"
        }

class O2OInventoryLinker:
    def __init__(self):
        self.redis = redis.Redis(host="localhost", port=6379, db=9)
        self.inventory_topic = "suning:o2o:inventory:change"  # Kafka主题
    def sync_store_inventory(self, product_id: str, store_id: str, stock: int):
        """门店库存实时同步（基于CDC+Kafka）"""
        # 1. 缓存更新（设置10分钟过期，配合实时变更刷新）
        cache_key = f"o2o:stock:{product_id}:{store_id}"
        self.redis.setex(cache_key, 600, stock)
        # 2. 发送变更消息（供搜索服务消费更新索引）
        from kafka import KafkaProducer
        producer = KafkaProducer(bootstrap_servers="localhost:9092")
        producer.send(
            self.inventory_topic,
            key=product_id.encode(),
            value=json.dumps({
                "product_id": product_id,
                "store_id": store_id,
                "stock": stock,
                "update_time": int(time.time())
            }).encode()
        )
        producer.flush()
    def check_inventory_availability(self, product_id: str, user_location: str) -> dict:
        """搜索结果库存校验：匹配最近门店库存"""
        # 1. 根据用户位置匹配3km内门店
        nearby_stores = self._get_nearby_stores(user_location, 3)
        # 2. 批量查询门店库存
        cache_keys = [f"o2o:stock:{product_id}:{store['id']}" for store in nearby_stores]
        stock_list = self.redis.mget(cache_keys)
        # 3. 筛选有库存的门店并排序（优先距离近）
        available = []
        for i, stock in enumerate(stock_list):
            if stock and int(stock) > 0:
                available.append({
                    "store_id": nearby_stores[i]["id"],
                    "store_name": nearby_stores[i]["name"],
                    "distance": nearby_stores[i]["distance"],
                    "stock": int(stock),
                    "pickup_time": "1小时内" if nearby_stores[i]["distance"] < 1 else "2-4小时"
                })
        return {"available_stores": available, "has_stock": len(available) > 0}