五金门店70万商品标签系统架构方案

# 规则匹配引擎伪代码

TAG_RULES = {
    # 场景标签规则
    "scene": {
        "engineering": {
            "keywords": ["工程", "建筑", "施工", "项目", "工地", "钢筋", "水泥", "模板"],
            "categories": [101, 102, 103],  # 对应分类ID
            "price_range": (50, 100000)  # 价格区间
        },
        "household": {
            "keywords": ["家用", "家庭", "维修", "日常", "DIY"],
            "categories": [201, 202, 203],
            "price_range": (0, 500)
        },
        "industrial": {
            "keywords": ["工业", "工厂", "机械", "设备", "电机", "轴承"],
            "categories": [301, 302, 303],
            "price_range": (100, 100000)
        }
    },

    # 难度等级规则
    "difficulty": {
        "professional": {
            "keywords": ["电焊", "切割", "大型", "重型", "专业设备"],
            "price_min": 500
        },
        "normal": {
            "keywords": ["扳手", "螺丝刀", "锤", "锯"],
            "price_range": (5, 500)
        },
        "beginner": {
            "keywords": ["钉子", "胶带", "粘钩", "小型"],
            "price_max": 50
        }
    },

    # 价格区间规则
    "price_range": {
        "price_a": (0, 5),
        "price_b": (5, 20),
        "price_c": (20, 100),
        "price_d": (100, 999999)
    },

    # 专业领域规则
    "profession": {
        "electrician": {
            "keywords": ["电线", "电缆", "开关", "插座", "配电", "空开", "漏保"]
        },
        "plumber": {
            "keywords": ["水管", "阀门", "龙头", "生料带", "接头", "PVC", "PPR"]
        },
        "painter": {
            "keywords": ["油漆", "涂料", "乳胶漆", "色浆", "滚筒", "刷子", "砂纸"]
        }
    }
}

def match_rules(product):
    """规则匹配标注"""
    matched_tags = []

    for dimension, rules in TAG_RULES.items():
        for tag_code, rule in rules.items():
            score = 0

            # 关键词匹配打分
            if rule.get("keywords"):
                for kw in rule["keywords"]:
                    if kw in product["name"] or kw in product.get("spec", ""):
                        score += 1

            # 分类匹配
            if rule.get("categories"):
                if product.get("category_id") in rule["categories"]:
                    score += 3

            # 价格区间匹配
            price = product.get("retail_price", 0)
            if "price_range" in rule:
                pmin, pmax = rule["price_range"]
                if pmin <= price < pmax:
                    score += 2
            if "price_min" in rule and price >= rule["price_min"]:
                score += 2
            if "price_max" in rule and price <= rule["price_max"]:
                score += 2

            # 阈值判断
            if score >= 2:
                matched_tags.append({
                    "dimension": dimension,
                    "tag_code": tag_code,
                    "score": score,
                    "source": "rule"
                })

    return matched_tags

# AI语义标注方案

import openai
from sentence_transformers import SentenceTransformer
import numpy as np

class AISemanticTagger:
    def __init__(self):
        # 本地Embedding模型（推荐）
        self.model = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2')

        # 或者使用OpenAI API
        # self.openai_client = openai.OpenAI()

        # 预定义标签的向量表示
        self.tag_embeddings = self._build_tag_embeddings()

    def _build_tag_embeddings(self):
        """构建标签向量库"""
        tag_texts = []
        tag_info = []

        # 为每个标签构建描述文本
        for dimension, tags in TAG_DEFINITIONS.items():
            for tag in tags:
                # 构建描述性文本
                desc = f"{dimension['name']}-{tag['name']}: {tag.get('description', '')}"
                tag_texts.append(desc)
                tag_info.append({
                    "dimension": dimension["code"],
                    "tag_code": tag["code"],
                    "name": tag["name"]
                })

        # 批量计算向量
        embeddings = self.model.encode(tag_texts, batch_size=64)
        return {"embeddings": embeddings, "tags": tag_info}

    def tag_product(self, product):
        """为单个商品生成标签"""
        # 构建商品描述
        product_text = f"{product['name']} {product.get('spec', '')} {product.get('brand', '')}"

        # 计算商品向量
        product_vec = self.model.encode([product_text])[0]

        # 计算与所有标签的相似度
        similarities = np.dot(
            self.tag_embeddings["embeddings"], 
            product_vec
        )

        # 选取每个维度最相关的标签
        result = {}
        for i, sim in enumerate(similarities):
            tag = self.tag_embeddings["tags"][i]
            dim = tag["dimension"]

            if dim not in result or sim > result[dim]["score"]:
                result[dim] = {
                    "tag_code": tag["tag_code"],
                    "score": float(sim),
                    "confidence": float(sim)
                }

        return result

    def batch_tag(self, products, batch_size=1000):
        """批量标注"""
        results = []
        total = len(products)

        for i in range(0, total, batch_size):
            batch = products[i:i+batch_size]

            # 批量构建商品描述
            texts = [f"{p['name']} {p.get('spec', '')} {p.get('brand', '')}" 
                     for p in batch]

            # 批量计算向量
            product_vecs = self.model.encode(texts, batch_size=64)

            # 批量计算相似度
            for j, product_vec in enumerate(product_vecs):
                similarities = np.dot(
                    self.tag_embeddings["embeddings"], 
                    product_vec
                )

                # 解析每个维度的top结果
                product_result = {"product_id": batch[j]["id"], "tags": []}
                dim_scores = {}

                for k, sim in enumerate(similarities):
                    tag = self.tag_embeddings["tags"][k]
                    dim = tag["dimension"]

                    if dim not in dim_scores or sim > dim_scores[dim]["score"]:
                        dim_scores[dim] = {"tag_code": tag["tag_code"], "score": float(sim)}

                # 转换为标签列表
                for dim, info in dim_scores.items():
                    if info["score"] > 0.5:  # 置信度阈值
                        product_result["tags"].append({
                            "dimension": dim,
                            "tag_code": info["tag_code"],
                            "source": "ai",
                            "confidence": info["score"]
                        })

                results.append(product_result)

            print(f"进度: {min(i+batch_size, total)}/{total}")

        return results

# 增量标注触发器（伪代码）

def on_new_product(product_id):
    """新商品入库自动标注"""
    product = db.get_product(product_id)

    # 1. 规则匹配（同步，实时）
    rule_tags = match_rules(product)

    # 2. AI标注（异步，延迟执行）
    queue.enqueue("ai_tag_task", {
        "product_id": product_id,
        "priority": "normal"
    })

    # 3. 记录标注结果
    save_tags(product_id, rule_tags)

    # 4. 同步到ES
    sync_to_elasticsearch(product_id)

def scheduled_full_recalculation():
    """定期全量重算（每周一次）"""
    # 1. 暂停增量标注
    pause_incremental_tagging()

    # 2. 重新运行AI标注
    run_batch_ai_tagging(
        product_ids=get_all_product_ids(),
        batch_size=5000,
        priority="low"
    )

    # 3. 重新计算权重
    recalculate_weights()

    # 4. 全量同步ES
    full_sync_to_elasticsearch()

    # 5. 恢复增量标注
    resume_incremental_tagging()

# 分批处理配置
BATCH_CONFIG = {
    "rule_matching": {
        "batch_size": 10000,
        "concurrent_batches": 8,
        "estimated_time": "3-5分钟",
        "total_products": 700000
    },

    "ai_semantic": {
        "batch_size": 1000,
        "concurrent_batches": 2,
        "estimated_time": "5-8小时",
        "embedding_model": "paraphrase-multilingual-MiniLM-L12-v2",
        "gpu_memory": "4GB"
    }
}

def process_batch_rule_matching(product_ids, batch_size=10000):
    """规则匹配批量处理"""
    total = len(product_ids)
    results = []

    for i in range(0, total, batch_size):
        batch_ids = product_ids[i:i+batch_size]
        batch_products = get_products(batch_ids)

        # 并行处理
        batch_results = parallel_map(
            match_rules, 
            batch_products, 
            workers=8
        )

        results.extend(batch_results)

        # 每批处理完写入数据库
        batch_insert_product_tags(results)

        # 更新任务进度
        update_task_progress(min(i+batch_size, total), total)

    return results

def process_batch_ai_semantic(product_ids, batch_size=1000):
    """AI语义批量处理（GPU加速）"""
    total = len(product_ids)
    results = []

    for i in range(0, total, batch_size):
        batch_ids = product_ids[i:i+batch_size]
        batch_products = get_products(batch_ids)

        # 调用AI标注
        batch_results = ai_tagger.batch_tag(batch_products, batch_size=1000)

        results.extend(batch_results)

        # 写入数据库（低置信度标记待复核）
        batch_insert_product_tags(results, set_review_flag=True)

        print(f"AI标注进度: {min(i+batch_size, total)}/{total}")

    return results

# 数据同步策略

class DataSyncManager:
    """MySQL与ES数据同步管理"""

    def __init__(self):
        self.mysql_client = MySQLClient()
        self.es_client = ElasticsearchClient()
        self.redis_client = RedisClient()

    def sync_single_product(self, product_id):
        """单个商品变更同步"""
        # 1. 获取商品完整信息
        product = self.mysql_client.get_product_with_tags(product_id)

        # 2. 构建ES文档
        es_doc = self._build_es_document(product)

        # 3. 更新ES
        self.es_client.index("products", es_doc, id=product_id)

        # 4. 清除缓存
        self.redis_client.delete(f"product:{product_id}")

    def sync_batch_products(self, product_ids, batch_size=5000):
        """批量同步"""
        for i in range(0, len(product_ids), batch_size):
            batch_ids = product_ids[i:i+batch_size]

            # 批量获取
            products = self.mysql_client.get_products_with_tags(batch_ids)

            # 批量构建ES文档
            actions = []
            for product in products:
                doc = self._build_es_document(product)
                actions.append({
                    "index": {"_index": "products", "_id": product["id"]}
                })
                actions.append(doc)

            # 批量写入ES
            self.es_client.bulk(actions)

            # 更新同步状态
            self.mysql_client.mark_es_synced(batch_ids)

            print(f"ES同步进度: {min(i+batch_size, len(product_ids))}/{len(product_ids)}")

    def full_sync(self):
        """全量同步（建议凌晨执行）"""
        # 1. 创建全量同步任务
        task_id = self.mysql_client.create_sync_task("full")

        # 2. 切换ES别名（零停机）
        self.es_client.switch_alias("products", "products_v2", "products_v1")

        # 3. 全量重建索引
        self.rebuild_index()

        # 4. 切换完成
        self.es_client.switch_alias("products", "products_v1", "products_v2")

        # 5. 删除旧索引
        self.es_client.delete_index("products_v2")

    def incremental_sync(self, interval_seconds=60):
        """增量同步（定时任务）"""
        while True:
            # 获取最近更新的商品
            recent_products = self.mysql_client.get_recent_updated(
                since_minutes=interval_seconds/60
            )

            if recent_products:
                self.sync_batch_products(recent_products)

            time.sleep(interval_seconds)