tangwang · tangwang
Showing 14 changed files Show diff stats
api/app.py
api/models.py
api/result_formatter.py
api/routes/search.py
docs/ES常用表达式.md
indexer/mapping_generator.py
indexer/spu_transformer.py
query/query_parser.py
scripts/recreate_and_import.py
search/es_query_builder.py
search/multilang_query_builder.py
search/query_config.py
search/searcher.py
utils/es_client.py
@@ -40,14 +40,13 @@ limiter = Limiter(key_func=get_remote_address)
 # Add parent directory to path
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
-from config import ConfigLoader, SearchConfig
 from config.env_config import ES_CONFIG
 from utils import ESClient
 from search import Searcher
+from search.query_config import DEFAULT_INDEX_NAME
 from query import QueryParser
 # Global instances
-_config: Optional[SearchConfig] = None
 _es_client: Optional[ESClient] = None
 _searcher: Optional[Searcher] = None
 _query_parser: Optional[QueryParser] = None
@@ -60,20 +59,11 @@ def init_service(es_host: str = &quot;http://localhost:9200&quot;):
     Args:
         es_host: Elasticsearch host URL
     """
-    global _config, _es_client, _searcher, _query_parser
+    global _es_client, _searcher, _query_parser
     start_time = time.time()
     logger.info("Initializing search service (multi-tenant)")
-    # Load and validate configuration
-    logger.info("Loading configuration...")
-    config_loader = ConfigLoader("config/config.yaml")
-    _config = config_loader.load_config()
-    errors = config_loader.validate_config(_config)
-    if errors:
-        raise ValueError(f"Configuration validation failed: {errors}")
-    logger.info(f"Configuration loaded: {_config.es_index_name}")
-
     # Get ES credentials
     es_username = os.getenv('ES_USERNAME') or ES_CONFIG.get('username')
     es_password = os.getenv('ES_PASSWORD') or ES_CONFIG.get('password')
@@ -91,20 +81,15 @@ def init_service(es_host: str = &quot;http://localhost:9200&quot;):
     # Initialize components
     logger.info("Initializing query parser...")
-    _query_parser = QueryParser(_config)
+    _query_parser = QueryParser()
     logger.info("Initializing searcher...")
-    _searcher = Searcher(_config, _es_client, _query_parser)
+    _searcher = Searcher(_es_client, _query_parser, index_name=DEFAULT_INDEX_NAME)
     elapsed = time.time() - start_time
-    logger.info(f"Search service ready! (took {elapsed:.2f}s)")
+    logger.info(f"Search service ready! (took {elapsed:.2f}s) | Index: {DEFAULT_INDEX_NAME}")
-def get_config() -> SearchConfig:
-    """Get search engine configuration."""
-    if _config is None:
-        raise RuntimeError("Service not initialized")
-    return _config
 def get_es_client() -> ESClient:
@@ -243,8 +228,8 @@ async def health_check(request: Request):
     """Health check endpoint."""
     try:
         # Check if services are initialized
-        get_config()
         get_es_client()
+        get_searcher()
         return {
             "status": "healthy",
@@ -69,6 +69,10 @@ class SearchRequest(BaseModel):
     query: str = Field(..., description="搜索查询字符串，支持布尔表达式（AND, OR, RANK, ANDNOT）")
     size: int = Field(10, ge=1, le=100, description="返回结果数量")
     from_: int = Field(0, ge=0, alias="from", description="分页偏移量")
+    language: Literal["zh", "en"] = Field(
+        "zh",
+        description="响应语言：'zh'（中文）或 'en'（英文），用于选择 title/description/vendor 等多语言字段"
+    )
     # 过滤器 - 精确匹配和多值匹配
     filters: Optional[Dict[str, Union[str, int, bool, List[Union[str, int]]]]] = Field(
@@ -175,28 +179,53 @@ class FacetResult(BaseModel):
 class SkuResult(BaseModel):
     """SKU 结果"""
     sku_id: str = Field(..., description="SKU ID")
-    title: Optional[str] = Field(None, description="SKU标题")
+    # 与 ES nested skus 结构对齐
     price: Optional[float] = Field(None, description="价格")
     compare_at_price: Optional[float] = Field(None, description="原价")
-    sku: Optional[str] = Field(None, description="SKU编码")
+    sku_code: Optional[str] = Field(None, description="SKU编码")
     stock: int = Field(0, description="库存数量")
-    options: Optional[Dict[str, Any]] = Field(None, description="选项（颜色、尺寸等）")
+    weight: Optional[float] = Field(None, description="重量")
+    weight_unit: Optional[str] = Field(None, description="重量单位")
+    option1_value: Optional[str] = Field(None, description="选项1取值（如颜色）")
+    option2_value: Optional[str] = Field(None, description="选项2取值（如尺码）")
+    option3_value: Optional[str] = Field(None, description="选项3取值")
+    image_src: Optional[str] = Field(None, description="SKU图片地址")
 class SpuResult(BaseModel):
     """SPU 搜索结果"""
     spu_id: str = Field(..., description="SPU ID")
     title: Optional[str] = Field(None, description="商品标题")
+    brief: Optional[str] = Field(None, description="商品短描述")
     handle: Optional[str] = Field(None, description="商品handle")
     description: Optional[str] = Field(None, description="商品描述")
     vendor: Optional[str] = Field(None, description="供应商/品牌")
-    category: Optional[str] = Field(None, description="类目")
-    tags: Optional[str] = Field(None, description="标签")
+    category: Optional[str] = Field(None, description="类目（兼容字段，等同于category_name）")
+    category_path: Optional[str] = Field(None, description="类目路径（多级，用于面包屑）")
+    category_name: Optional[str] = Field(None, description="类目名称（展示用）")
+    category_id: Optional[str] = Field(None, description="类目ID")
+    category_level: Optional[int] = Field(None, description="类目层级")
+    category1_name: Optional[str] = Field(None, description="一级类目名称")
+    category2_name: Optional[str] = Field(None, description="二级类目名称")
+    category3_name: Optional[str] = Field(None, description="三级类目名称")
+    tags: Optional[List[str]] = Field(None, description="标签列表")
     price: Optional[float] = Field(None, description="价格（min_price）")
     compare_at_price: Optional[float] = Field(None, description="原价")
     currency: str = Field("USD", description="货币单位")
     image_url: Optional[str] = Field(None, description="主图URL")
     in_stock: bool = Field(True, description="是否有库存")
+    # SKU 扁平化信息
+    sku_prices: Optional[List[float]] = Field(None, description="所有SKU价格列表")
+    sku_weights: Optional[List[int]] = Field(None, description="所有SKU重量列表")
+    sku_weight_units: Optional[List[str]] = Field(None, description="所有SKU重量单位列表")
+    total_inventory: Optional[int] = Field(None, description="总库存")
+    option1_name: Optional[str] = Field(None, description="选项1名称（如颜色）")
+    option2_name: Optional[str] = Field(None, description="选项2名称（如尺码）")
+    option3_name: Optional[str] = Field(None, description="选项3名称")
+    specifications: Optional[List[Dict[str, Any]]] = Field(
+        None,
+        description="规格列表（与 ES specifications 字段对应）"
+    )
     skus: List[SkuResult] = Field(default_factory=list, description="SKU列表")
     relevance_score: float = Field(..., ge=0.0, description="相关性分数（ES原始分数）")
@@ -12,7 +12,8 @@ class ResultFormatter:
     @staticmethod
     def format_search_results(
         es_hits: List[Dict[str, Any]],
-        max_score: float = 1.0
+        max_score: float = 1.0,
+        language: str = "zh"
     ) -> List[SpuResult]:
         """
         Convert ES hits to SpuResult list.
@@ -25,6 +26,18 @@ class ResultFormatter:
             List of SpuResult objects
         """
         results = []
+        lang = (language or "zh").lower()
+        if lang not in ("zh", "en"):
+            lang = "en"
+
+        def pick_lang_field(src: Dict[str, Any], base: str) -> Optional[str]:
+            """从 *_zh / *_en 字段中按语言选择一个值，若目标语言缺失则回退到另一种。"""
+            zh_val = src.get(f"{base}_zh")
+            en_val = src.get(f"{base}_en")
+            if lang == "zh":
+                return zh_val or en_val
+            else:
+                return en_val or zh_val
         for hit in es_hits:
             source = hit.get('_source', {})
@@ -40,6 +53,14 @@ class ResultFormatter:
                 except (ValueError, TypeError):
                     relevance_score = 0.0
+            # Multi-language fields
+            title = pick_lang_field(source, "title")
+            brief = pick_lang_field(source, "brief")
+            description = pick_lang_field(source, "description")
+            vendor = pick_lang_field(source, "vendor")
+            category_path = pick_lang_field(source, "category_path")
+            category_name = pick_lang_field(source, "category_name")
+
             # Extract SKUs
             skus = []
             skus_data = source.get('skus', [])
@@ -62,17 +83,33 @@ class ResultFormatter:
             # Build SpuResult
             spu = SpuResult(
                 spu_id=str(source.get('spu_id', '')),
-                title=source.get('title'),
+                title=title,
+                brief=brief,
                 handle=source.get('handle'),
-                description=source.get('description'),
-                vendor=source.get('vendor'),
-                category=source.get('category'),
+                description=description,
+                vendor=vendor,
+                category=category_name,
+                category_path=category_path,
+                category_name=category_name,
+                category_id=source.get('category_id'),
+                category_level=source.get('category_level'),
+                category1_name=source.get('category1_name'),
+                category2_name=source.get('category2_name'),
+                category3_name=source.get('category3_name'),
                 tags=source.get('tags'),
                 price=source.get('min_price'),
                 compare_at_price=source.get('compare_at_price'),
                 currency="USD",  # Default currency
                 image_url=source.get('image_url'),
                 in_stock=in_stock,
+                sku_prices=source.get('sku_prices'),
+                sku_weights=source.get('sku_weights'),
+                sku_weight_units=source.get('sku_weight_units'),
+                total_inventory=source.get('total_inventory'),
+                option1_name=source.get('option1_name'),
+                option2_name=source.get('option2_name'),
+                option3_name=source.get('option3_name'),
+                specifications=source.get('specifications'),
                 skus=skus,
                 relevance_score=relevance_score
             )
@@ -89,6 +126,11 @@ class ResultFormatter:
         """
         Format ES aggregations to FacetResult list.
+        支持：
+        1. 普通terms聚合
+        2. range聚合
+        3. specifications嵌套聚合（按name分组，然后按value聚合）
+
         Args:
             es_aggregations: ES aggregations response
             facet_configs: Facet configurations (optional)
@@ -100,6 +142,38 @@ class ResultFormatter:
         for field_name, agg_data in es_aggregations.items():
             display_field = field_name[:-6] if field_name.endswith("_facet") else field_name
+            
+            # 处理specifications嵌套分面
+            if field_name == "specifications_facet" and 'by_name' in agg_data:
+                # specifications嵌套聚合：按name分组，每个name下有value_counts
+                by_name_agg = agg_data['by_name']
+                if 'buckets' in by_name_agg:
+                    for name_bucket in by_name_agg['buckets']:
+                        name = name_bucket['key']
+                        value_counts = name_bucket.get('value_counts', {})
+                        
+                        values = []
+                        if 'buckets' in value_counts:
+                            for value_bucket in value_counts['buckets']:
+                                value = FacetValue(
+                                    value=value_bucket['key'],
+                                    label=str(value_bucket['key']),
+                                    count=value_bucket['doc_count'],
+                                    selected=False
+                                )
+                                values.append(value)
+                        
+                        # 为每个name创建一个分面结果
+                        facet = FacetResult(
+                            field=f"specifications.{name}",
+                            label=str(name),  # 使用name作为label，如"颜色"、"尺寸"
+                            type="terms",
+                            values=values,
+                            total_count=name_bucket['doc_count']
+                        )
+                        facets.append(facet)
+                continue
+            
             # Handle terms aggregation
             if 'buckets' in agg_data:
                 values = []
@@ -94,7 +94,8 @@ async def search(request: SearchRequest, http_request: Request):
             context=context,
             sort_by=request.sort_by,
             sort_order=request.sort_order,
-            debug=request.debug
+            debug=request.debug,
+            language=request.language,
         )
         # Include performance summary in response
@@ -7,3 +7,14 @@ GET /search_products/_search
   }
 }
+
+curl -u 'essa:4hOaLaf41y2VuI8y'   -X GET 'http://localhost:9200/search_products/_search?pretty'   -H 'Content-Type: application/json'   -d '{
+    "size": 5,
+    "query": {
+      "bool": {
+        "filter": [
+          { "term": { "tenant_id": "162" } }
+        ]
+      }
+    }
+  }'
 \ No newline at end of file
@@ -51,23 +51,26 @@ def create_index_if_not_exists(es_client, index_name: str, mapping: Dict[str, An
     Create Elasticsearch index if it doesn't exist.
     Args:
-        es_client: Elasticsearch client instance
+        es_client: ESClient instance
         index_name: Name of the index to create
         mapping: Index mapping configuration. If None, loads from default file.
     Returns:
         True if index was created, False if it already exists
     """
-    if es_client.indices.exists(index=index_name):
+    if es_client.index_exists(index_name):
         logger.info(f"Index '{index_name}' already exists")
         return False
     if mapping is None:
         mapping = load_mapping()
-    es_client.indices.create(index=index_name, body=mapping)
-    logger.info(f"Index '{index_name}' created successfully")
-    return True
+    if es_client.create_index(index_name, mapping):
+        logger.info(f"Index '{index_name}' created successfully")
+        return True
+    else:
+        logger.error(f"Failed to create index '{index_name}'")
+        return False
 def delete_index_if_exists(es_client, index_name: str) -> bool:
@@ -75,19 +78,22 @@ def delete_index_if_exists(es_client, index_name: str) -&gt; bool:
     Delete Elasticsearch index if it exists.
     Args:
-        es_client: Elasticsearch client instance
+        es_client: ESClient instance
         index_name: Name of the index to delete
     Returns:
         True if index was deleted, False if it didn't exist
     """
-    if not es_client.indices.exists(index=index_name):
+    if not es_client.index_exists(index_name):
         logger.warning(f"Index '{index_name}' does not exist")
         return False
-    es_client.indices.delete(index=index_name)
-    logger.info(f"Index '{index_name}' deleted successfully")
-    return True
+    if es_client.delete_index(index_name):
+        logger.info(f"Index '{index_name}' deleted successfully")
+        return True
+    else:
+        logger.error(f"Failed to delete index '{index_name}'")
+        return False
 def update_mapping(es_client, index_name: str, new_fields: Dict[str, Any]) -> bool:
@@ -95,18 +101,21 @@ def update_mapping(es_client, index_name: str, new_fields: Dict[str, Any]) -&gt; bo
     Update mapping for existing index (only adding new fields).
     Args:
-        es_client: Elasticsearch client instance
+        es_client: ESClient instance
         index_name: Name of the index
         new_fields: New field mappings to add
     Returns:
         True if successful
     """
-    if not es_client.indices.exists(index=index_name):
+    if not es_client.index_exists(index_name):
         logger.error(f"Index '{index_name}' does not exist")
         return False
     mapping = {"properties": new_fields}
-    es_client.indices.put_mapping(index=index_name, body=mapping)
-    logger.info(f"Mapping updated for index '{index_name}'")
-    return True
+    if es_client.update_mapping(index_name, mapping):
+        logger.info(f"Mapping updated for index '{index_name}'")
+        return True
+    else:
+        logger.error(f"Failed to update mapping for index '{index_name}'")
+        return False
@@ -124,7 +124,7 @@ class SPUTransformer:
         query = text("""
             SELECT 
                 id, spu_id, shop_id, shoplazza_id, shoplazza_product_id,
-                position, name, values, tenant_id,
+                position, name, `values`, tenant_id,
                 creator, create_time, updater, update_time, deleted
             FROM shoplazza_product_option
             WHERE tenant_id = :tenant_id AND deleted = 0
@@ -8,8 +8,14 @@ from typing import Dict, List, Optional, Any
 import numpy as np
 import logging
-from config import SearchConfig, QueryConfig
 from embeddings import BgeEncoder
+from search.query_config import (
+    ENABLE_TEXT_EMBEDDING,
+    ENABLE_TRANSLATION,
+    REWRITE_DICTIONARY,
+    TRANSLATION_API_KEY,
+    TRANSLATION_SERVICE
+)
 from .language_detector import LanguageDetector
 from .translator import Translator
 from .query_rewriter import QueryRewriter, QueryNormalizer
@@ -64,7 +70,6 @@ class QueryParser:
     def __init__(
         self,
-        config: SearchConfig,
         text_encoder: Optional[BgeEncoder] = None,
         translator: Optional[Translator] = None
     ):
@@ -72,24 +77,21 @@ class QueryParser:
         Initialize query parser.
         Args:
-            config: Search configuration
             text_encoder: Text embedding encoder (lazy loaded if not provided)
             translator: Translator instance (lazy loaded if not provided)
         """
-        self.config = config
-        self.query_config = config.query_config
         self._text_encoder = text_encoder
         self._translator = translator
         # Initialize components
         self.normalizer = QueryNormalizer()
         self.language_detector = LanguageDetector()
-        self.rewriter = QueryRewriter(self.query_config.rewrite_dictionary)
+        self.rewriter = QueryRewriter(REWRITE_DICTIONARY)
     @property
     def text_encoder(self) -> BgeEncoder:
         """Lazy load text encoder."""
-        if self._text_encoder is None and self.query_config.enable_text_embedding:
+        if self._text_encoder is None and ENABLE_TEXT_EMBEDDING:
             logger.info("Initializing text encoder (lazy load)...")
             self._text_encoder = BgeEncoder()
         return self._text_encoder
@@ -97,13 +99,13 @@ class QueryParser:
     @property
     def translator(self) -> Translator:
         """Lazy load translator."""
-        if self._translator is None and self.query_config.enable_translation:
+        if self._translator is None and ENABLE_TRANSLATION:
             logger.info("Initializing translator (lazy load)...")
             self._translator = Translator(
-                api_key=self.query_config.translation_api_key,
+                api_key=TRANSLATION_API_KEY,
                 use_cache=True,
-                glossary_id=getattr(self.query_config, 'translation_glossary_id', None),
-                translation_context=getattr(self.query_config, 'translation_context', 'e-commerce product search')
+                glossary_id=None,  # Can be added to query_config if needed
+                translation_context='e-commerce product search'
             )
         return self._translator
@@ -154,7 +156,7 @@ class QueryParser:
         # Stage 2: Query rewriting
         rewritten = None
-        if self.query_config.enable_query_rewrite:
+        if REWRITE_DICTIONARY:  # Enable rewrite if dictionary exists
             rewritten = self.rewriter.rewrite(query_text)
             if rewritten != query_text:
                 log_info(f"查询重写 | '{query_text}' -> '{rewritten}'")
@@ -171,26 +173,11 @@ class QueryParser:
         # Stage 4: Translation
         translations = {}
-        if self.query_config.enable_translation:
+        if ENABLE_TRANSLATION:
             try:
                 # Determine target languages for translation
-                # If domain has language_field_mapping, only translate to languages in the mapping
-                # Otherwise, use all supported languages
-                target_langs_for_translation = self.query_config.supported_languages
-
-                # Check if domain has language_field_mapping
-                domain_config = next(
-                    (idx for idx in self.config.indexes if idx.name == domain),
-                    None
-                )
-                if domain_config and domain_config.language_field_mapping:
-                    # Only translate to languages that exist in the mapping
-                    available_languages = set(domain_config.language_field_mapping.keys())
-                    target_langs_for_translation = [
-                        lang for lang in self.query_config.supported_languages
-                        if lang in available_languages
-                    ]
-                    log_debug(f"域 '{domain}' 有语言字段映射，将翻译到: {target_langs_for_translation}")
+                # Simplified: always translate to Chinese and English
+                target_langs_for_translation = ['zh', 'en']
                 target_langs = self.translator.get_translation_needs(
                     detected_lang,
@@ -200,7 +187,7 @@ class QueryParser:
                 if target_langs:
                     log_info(f"开始翻译 | 源语言: {detected_lang} | 目标语言: {target_langs}")
                     # Use e-commerce context for better disambiguation
-                    translation_context = getattr(self.query_config, 'translation_context', 'e-commerce product search')
+                    translation_context = 'e-commerce product search'
                     translations = self.translator.translate_multi(
                         query_text,
                         target_langs,
@@ -223,7 +210,7 @@ class QueryParser:
         # Stage 5: Text embedding
         query_vector = None
         if (generate_vector and
-            self.query_config.enable_text_embedding and
+            ENABLE_TEXT_EMBEDDING and
             domain == "default"):  # Only generate vector for default domain
             try:
                 log_debug("开始生成查询向量")
@@ -0,0 +1,184 @@
+#!/usr/bin/env python3
+"""
+重建索引并导入数据的脚本。
+
+清除旧索引，使用新的mapping重建索引，然后导入数据。
+"""
+
+import sys
+import os
+import argparse
+from pathlib import Path
+
+# Add parent directory to path
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+from utils.db_connector import create_db_connection
+from utils.es_client import ESClient
+from indexer.mapping_generator import load_mapping, delete_index_if_exists, DEFAULT_INDEX_NAME
+from indexer.spu_transformer import SPUTransformer
+from indexer.bulk_indexer import BulkIndexer
+
+
+def main():
+    parser = argparse.ArgumentParser(description='重建ES索引并导入数据')
+    
+    # Database connection
+    parser.add_argument('--db-host', help='MySQL host (或使用环境变量 DB_HOST)')
+    parser.add_argument('--db-port', type=int, help='MySQL port (或使用环境变量 DB_PORT, 默认: 3306)')
+    parser.add_argument('--db-database', help='MySQL database (或使用环境变量 DB_DATABASE)')
+    parser.add_argument('--db-username', help='MySQL username (或使用环境变量 DB_USERNAME)')
+    parser.add_argument('--db-password', help='MySQL password (或使用环境变量 DB_PASSWORD)')
+    
+    # Tenant and ES
+    parser.add_argument('--tenant-id', required=True, help='Tenant ID (必需)')
+    parser.add_argument('--es-host', help='Elasticsearch host (或使用环境变量 ES_HOST, 默认: http://localhost:9200)')
+    
+    # Options
+    parser.add_argument('--batch-size', type=int, default=500, help='批量导入大小 (默认: 500)')
+    parser.add_argument('--skip-delete', action='store_true', help='跳过删除旧索引步骤')
+    
+    args = parser.parse_args()
+
+    print("=" * 60)
+    print("重建ES索引并导入数据")
+    print("=" * 60)
+
+    # 加载mapping
+    print("\n[1/4] 加载mapping配置...")
+    try:
+        mapping = load_mapping()
+        print(f"✓ 成功加载mapping配置")
+    except Exception as e:
+        print(f"✗ 加载mapping失败: {e}")
+        return 1
+
+    index_name = DEFAULT_INDEX_NAME
+    print(f"索引名称: {index_name}")
+
+    # 连接Elasticsearch
+    print("\n[2/4] 连接Elasticsearch...")
+    es_host = args.es_host or os.environ.get('ES_HOST', 'http://localhost:9200')
+    es_username = os.environ.get('ES_USERNAME')
+    es_password = os.environ.get('ES_PASSWORD')
+    
+    print(f"ES地址: {es_host}")
+    if es_username:
+        print(f"ES用户名: {es_username}")
+    
+    try:
+        if es_username and es_password:
+            es_client = ESClient(hosts=[es_host], username=es_username, password=es_password)
+        else:
+            es_client = ESClient(hosts=[es_host])
+        
+        if not es_client.ping():
+            print(f"✗ 无法连接到Elasticsearch: {es_host}")
+            return 1
+        print("✓ Elasticsearch连接成功")
+    except Exception as e:
+        print(f"✗ 连接Elasticsearch失败: {e}")
+        return 1
+
+    # 删除旧索引
+    if not args.skip_delete:
+        print("\n[3/4] 删除旧索引...")
+        if es_client.index_exists(index_name):
+            print(f"发现已存在的索引: {index_name}")
+            if delete_index_if_exists(es_client, index_name):
+                print(f"✓ 成功删除索引: {index_name}")
+            else:
+                print(f"✗ 删除索引失败: {index_name}")
+                return 1
+        else:
+            print(f"索引不存在，跳过删除: {index_name}")
+    else:
+        print("\n[3/4] 跳过删除旧索引步骤")
+
+    # 创建新索引
+    print("\n[4/4] 创建新索引...")
+    try:
+        if es_client.index_exists(index_name):
+            print(f"✓ 索引已存在: {index_name}，跳过创建")
+        else:
+            print(f"创建索引: {index_name}")
+            if es_client.create_index(index_name, mapping):
+                print(f"✓ 成功创建索引: {index_name}")
+            else:
+                print(f"✗ 创建索引失败: {index_name}")
+                return 1
+    except Exception as e:
+        print(f"✗ 创建索引失败: {e}")
+        import traceback
+        traceback.print_exc()
+        return 1
+
+    # 连接MySQL
+    print("\n[5/5] 连接MySQL...")
+    db_host = args.db_host or os.environ.get('DB_HOST')
+    db_port = args.db_port or int(os.environ.get('DB_PORT', 3306))
+    db_database = args.db_database or os.environ.get('DB_DATABASE')
+    db_username = args.db_username or os.environ.get('DB_USERNAME')
+    db_password = args.db_password or os.environ.get('DB_PASSWORD')
+
+    if not all([db_host, db_database, db_username, db_password]):
+        print("✗ MySQL连接参数不完整")
+        print("请提供 --db-host, --db-database, --db-username, --db-password")
+        print("或设置环境变量: DB_HOST, DB_DATABASE, DB_USERNAME, DB_PASSWORD")
+        return 1
+
+    print(f"MySQL: {db_host}:{db_port}/{db_database}")
+    try:
+        db_engine = create_db_connection(
+            host=db_host,
+            port=db_port,
+            database=db_database,
+            username=db_username,
+            password=db_password
+        )
+        print("✓ MySQL连接成功")
+    except Exception as e:
+        print(f"✗ 连接MySQL失败: {e}")
+        return 1
+
+    # 导入数据
+    print("\n[6/6] 导入数据...")
+    print(f"Tenant ID: {args.tenant_id}")
+    print(f"批量大小: {args.batch_size}")
+    
+    try:
+        transformer = SPUTransformer(db_engine, args.tenant_id)
+        print("正在转换数据...")
+        documents = transformer.transform_batch()
+        print(f"✓ 转换完成: {len(documents)} 个文档")
+        
+        if not documents:
+            print("⚠ 没有数据需要导入")
+            return 0
+
+        print(f"正在导入数据到ES (批量大小: {args.batch_size})...")
+        indexer = BulkIndexer(es_client, index_name, batch_size=args.batch_size)
+        results = indexer.index_documents(documents, id_field="spu_id", show_progress=True)
+        
+        print(f"\n{'='*60}")
+        print("导入完成!")
+        print(f"{'='*60}")
+        print(f"成功: {results['success']}")
+        print(f"失败: {results['failed']}")
+        print(f"耗时: {results.get('elapsed_time', 0):.2f}秒")
+        
+        if results['failed'] > 0:
+            print(f"\n⚠ 警告: {results['failed']} 个文档导入失败")
+            return 1
+        
+        return 0
+    except Exception as e:
+        print(f"✗ 导入数据失败: {e}")
+        import traceback
+        traceback.print_exc()
+        return 1
+
+
+if __name__ == '__main__':
+    sys.exit(main())
+
@@ -330,10 +330,15 @@ class ESQueryBuilder:
         """
         构建分面聚合。
+        支持：
+        1. 分类分面：category1_name, category2_name, category3_name, category_name
+        2. specifications分面：嵌套聚合，按name聚合，然后按value聚合
+        
         Args:
             facet_configs: 分面配置列表（标准格式）：
                 - str: 字段名，使用默认 terms 配置
                 - FacetConfig: 详细的分面配置对象
+                - 特殊值 "specifications": 构建specifications嵌套分面
         Returns:
             ES aggregations 字典
@@ -344,6 +349,34 @@ class ESQueryBuilder:
         aggs = {}
         for config in facet_configs:
+            # 特殊处理：specifications嵌套分面
+            if isinstance(config, str) and config == "specifications":
+                # 构建specifications嵌套分面（按name聚合，然后按value聚合）
+                aggs["specifications_facet"] = {
+                    "nested": {
+                        "path": "specifications"
+                    },
+                    "aggs": {
+                        "by_name": {
+                            "terms": {
+                                "field": "specifications.name",
+                                "size": 20,
+                                "order": {"_count": "desc"}
+                            },
+                            "aggs": {
+                                "value_counts": {
+                                    "terms": {
+                                        "field": "specifications.value",
+                                        "size": 10,
+                                        "order": {"_count": "desc"}
+                                    }
+                                }
+                            }
+                        }
+                    }
+                }
+                continue
+            
             # 简单模式：只有字段名（字符串）
             if isinstance(config, str):
                 field = config
@@ -11,9 +11,9 @@ import numpy as np
 import logging
 import re
-from config import SearchConfig, IndexConfig
 from query import ParsedQuery
 from .es_query_builder import ESQueryBuilder
+from .query_config import DEFAULT_MATCH_FIELDS, DOMAIN_FIELDS, FUNCTION_SCORE_CONFIG
 logger = logging.getLogger(__name__)
@@ -30,8 +30,8 @@ class MultiLanguageQueryBuilder(ESQueryBuilder):
     def __init__(
         self,
-        config: SearchConfig,
         index_name: str,
+        match_fields: Optional[List[str]] = None,
         text_embedding_field: Optional[str] = None,
         image_embedding_field: Optional[str] = None,
         source_fields: Optional[List[str]] = None
@@ -40,53 +40,32 @@ class MultiLanguageQueryBuilder(ESQueryBuilder):
         Initialize multi-language query builder.
         Args:
-            config: Search configuration
             index_name: ES index name
+            match_fields: Fields to search for text matching (default: from query_config)
             text_embedding_field: Field name for text embeddings
             image_embedding_field: Field name for image embeddings
             source_fields: Fields to return in search results (_source includes)
         """
-        self.config = config
-        self.function_score_config = config.function_score
+        self.function_score_config = FUNCTION_SCORE_CONFIG
-        # For default domain, use all fields as fallback
-        default_fields = self._get_domain_fields("default")
+        # Use provided match_fields or default
+        if match_fields is None:
+            match_fields = DEFAULT_MATCH_FIELDS
         super().__init__(
             index_name=index_name,
-            match_fields=default_fields,
+            match_fields=match_fields,
             text_embedding_field=text_embedding_field,
             image_embedding_field=image_embedding_field,
             source_fields=source_fields
         )
-        # Build domain configurations
-        self.domain_configs = self._build_domain_configs()
-
-    def _build_domain_configs(self) -> Dict[str, IndexConfig]:
-        """Build mapping of domain name to IndexConfig."""
-        return {index.name: index for index in self.config.indexes}
+        # Build domain configurations from query_config
+        self.domain_configs = DOMAIN_FIELDS
     def _get_domain_fields(self, domain_name: str) -> List[str]:
         """Get fields for a specific domain with boost notation."""
-        for index in self.config.indexes:
-            if index.name == domain_name:
-                result = []
-                for field_name in index.fields:
-                    field = self._get_field_by_name(field_name)
-                    if field and field.boost != 1.0:
-                        result.append(f"{field_name}^{field.boost}")
-                    else:
-                        result.append(field_name)
-                return result
-        return []
-
-    def _get_field_by_name(self, field_name: str):
-        """Get field configuration by name."""
-        for field in self.config.fields:
-            if field.name == field_name:
-                return field
-        return None
+        return self.domain_configs.get(domain_name, DEFAULT_MATCH_FIELDS)
     def build_multilang_query(
         self,
@@ -103,7 +82,7 @@ class MultiLanguageQueryBuilder(ESQueryBuilder):
         min_score: Optional[float] = None
     ) -> Dict[str, Any]:
         """
-        Build ES query with multi-language support (重构版).
+        Build ES query with multi-language support (简化版).
         Args:
             parsed_query: Parsed query with language info and translations
@@ -120,19 +99,18 @@ class MultiLanguageQueryBuilder(ESQueryBuilder):
         Returns:
             ES query DSL dictionary
         """
-        domain = parsed_query.domain
-        domain_config = self.domain_configs.get(domain)
-
-        if not domain_config:
-            # Fallback to default domain
-            domain = "default"
-            domain_config = self.domain_configs.get("default")
-
-        if not domain_config:
-            # Use original behavior
+        # 1. 根据域选择匹配字段（默认域使用 DEFAULT_MATCH_FIELDS）
+        domain = parsed_query.domain or "default"
+        domain_fields = self.domain_configs.get(domain) or DEFAULT_MATCH_FIELDS
+
+        # 2. 临时切换 match_fields，复用基类 build_query 逻辑
+        original_match_fields = self.match_fields
+        self.match_fields = domain_fields
+        try:
             return super().build_query(
-                query_text=parsed_query.rewritten_query,
+                query_text=parsed_query.rewritten_query or parsed_query.normalized_query,
                 query_vector=query_vector,
+                query_node=query_node,
                 filters=filters,
                 range_filters=range_filters,
                 size=size,
@@ -142,95 +120,9 @@ class MultiLanguageQueryBuilder(ESQueryBuilder):
                 knn_num_candidates=knn_num_candidates,
                 min_score=min_score
             )
-
-        logger.debug(f"Building query for domain: {domain}, language: {parsed_query.detected_language}")
-
-        # Build query clause with multi-language support
-        if query_node and isinstance(query_node, tuple) and len(query_node) > 0:
-            # Handle boolean query from tuple (AST, score)
-            ast_node = query_node[0]
-            query_clause = self._build_boolean_query_from_tuple(ast_node)
-            logger.debug(f"Using boolean query")
-        elif query_node and hasattr(query_node, 'operator') and query_node.operator != 'TERM':
-            # Handle boolean query using base class method
-            query_clause = self._build_boolean_query(query_node)
-            logger.debug(f"Using boolean query")
-        else:
-            # Handle text query with multi-language support
-            query_clause = self._build_multilang_text_query(parsed_query, domain_config)
-
-        # 构建内层bool: 文本和KNN二选一
-        inner_bool_should = [query_clause]
-
-        # 如果启用KNN，添加到should
-        if enable_knn and query_vector is not None and self.text_embedding_field:
-            knn_query = {
-                "knn": {
-                    "field": self.text_embedding_field,
-                    "query_vector": query_vector.tolist(),
-                    "k": knn_k,
-                    "num_candidates": knn_num_candidates
-                }
-            }
-            inner_bool_should.append(knn_query)
-            logger.info(f"KNN query added: field={self.text_embedding_field}, k={knn_k}")
-        else:
-            # Debug why KNN is not added
-            reasons = []
-            if not enable_knn:
-                reasons.append("enable_knn=False")
-            if query_vector is None:
-                reasons.append("query_vector is None")
-            if not self.text_embedding_field:
-                reasons.append(f"text_embedding_field is not set (current: {self.text_embedding_field})")
-            logger.debug(f"KNN query NOT added. Reasons: {', '.join(reasons) if reasons else 'unknown'}")
-
-        # 构建内层bool结构
-        inner_bool = {
-            "bool": {
-                "should": inner_bool_should,
-                "minimum_should_match": 1
-            }
-        }
-
-        # 构建外层bool: 包含filter
-        filter_clauses = self._build_filters(filters, range_filters) if (filters or range_filters) else []
-
-        outer_bool = {
-            "bool": {
-                "must": [inner_bool]
-            }
-        }
-
-        if filter_clauses:
-            outer_bool["bool"]["filter"] = filter_clauses
-
-        # 包裹function_score（从配置读取score_mode和boost_mode）
-        function_score_query = {
-            "function_score": {
-                "query": outer_bool,
-                "functions": self._build_score_functions(),
-                "score_mode": self.function_score_config.score_mode if self.function_score_config else "sum",
-                "boost_mode": self.function_score_config.boost_mode if self.function_score_config else "multiply"
-            }
-        }
-
-        es_query = {
-            "size": size,
-            "from": from_,
-            "query": function_score_query
-        }
-
-        # Add _source filtering if source_fields are configured
-        if self.source_fields:
-            es_query["_source"] = {
-                "includes": self.source_fields
-            }
-
-        if min_score is not None:
-            es_query["min_score"] = min_score
-
-        return es_query
+        finally:
+            # 恢复原始配置，避免影响后续查询
+            self.match_fields = original_match_fields
     def _build_score_functions(self) -> List[Dict[str, Any]]:
         """
@@ -291,7 +183,7 @@ class MultiLanguageQueryBuilder(ESQueryBuilder):
     def _build_multilang_text_query(
         self,
         parsed_query: ParsedQuery,
-        domain_config: IndexConfig
+        domain_config: Dict[str, Any]
     ) -> Dict[str, Any]:
         """
         Build text query with multi-language field routing.
@@ -37,17 +37,32 @@ DOMAIN_FIELDS: Dict[str, List[str]] = {
 }
 # Source fields to return in search results
+# 注意：为了在后端做多语言选择，_zh / _en 字段仍然需要从 ES 取出，
+# 但不会原样透出给前端，而是统一映射到 title / description / vendor 等字段。
 SOURCE_FIELDS = [
+    # 基本标识
     "tenant_id",
     "spu_id",
+    "create_time",
+    "update_time",
+
+    # 多语言文本字段（仅用于后端选择，不直接返回给前端）
     "title_zh",
+    "title_en",
     "brief_zh",
+    "brief_en",
     "description_zh",
+    "description_en",
     "vendor_zh",
-    "tags",
-    "image_url",
+    "vendor_en",
     "category_path_zh",
+    "category_path_en",
     "category_name_zh",
+    "category_name_en",
+
+    # 语言无关字段（直接返回给前端）
+    "tags",
+    "image_url",
     "category_id",
     "category_name",
     "category_level",
@@ -60,11 +75,12 @@ SOURCE_FIELDS = [
     "min_price",
     "max_price",
     "compare_at_price",
+    "sku_prices",
+    "sku_weights",
+    "sku_weight_units",
     "total_inventory",
-    "create_time",
-    "update_time",
     "skus",
-    "specifications"
+    "specifications",
 ]
 # Query processing settings
@@ -112,3 +128,13 @@ def load_rewrite_dictionary() -&gt; Dict[str, str]:
 REWRITE_DICTIONARY = load_rewrite_dictionary()
+# Default facets for faceted search
+# 分类分面：使用category1_name, category2_name, category3_name
+# specifications分面：使用嵌套聚合，按name分组，然后按value聚合
+DEFAULT_FACETS = [
+    "category1_name",  # 一级分类
+    "category2_name",  # 二级分类
+    "category3_name",  # 三级分类
+    "specifications"  # 规格分面（特殊处理：嵌套聚合）
+]
+
@@ -8,15 +8,23 @@ from typing import Dict, Any, List, Optional, Union
 import time
 import logging
-from config import SearchConfig
 from utils.es_client import ESClient
 from query import QueryParser, ParsedQuery
-from indexer import MappingGenerator
 from embeddings import CLIPImageEncoder
 from .boolean_parser import BooleanParser, QueryNode
 from .es_query_builder import ESQueryBuilder
 from .multilang_query_builder import MultiLanguageQueryBuilder
 from .rerank_engine import RerankEngine
+from .query_config import (
+    DEFAULT_INDEX_NAME,
+    DEFAULT_MATCH_FIELDS,
+    TEXT_EMBEDDING_FIELD,
+    IMAGE_EMBEDDING_FIELD,
+    SOURCE_FIELDS,
+    ENABLE_TRANSLATION,
+    ENABLE_TEXT_EMBEDDING,
+    RANKING_EXPRESSION
+)
 from context.request_context import RequestContext, RequestContextStage, create_request_context
 from api.models import FacetResult, FacetValue
 from api.result_formatter import ResultFormatter
@@ -79,39 +87,38 @@ class Searcher:
     def __init__(
         self,
-        config: SearchConfig,
         es_client: ESClient,
-        query_parser: Optional[QueryParser] = None
+        query_parser: Optional[QueryParser] = None,
+        index_name: str = DEFAULT_INDEX_NAME
     ):
         """
         Initialize searcher.
         Args:
-            config: Search configuration
             es_client: Elasticsearch client
             query_parser: Query parser (created if not provided)
+            index_name: ES index name (default: search_products)
         """
-        self.config = config
         self.es_client = es_client
-        self.query_parser = query_parser or QueryParser(config)
+        self.index_name = index_name
+        self.query_parser = query_parser or QueryParser()
         # Initialize components
         self.boolean_parser = BooleanParser()
-        self.rerank_engine = RerankEngine(config.ranking.expression, enabled=False)
+        self.rerank_engine = RerankEngine(RANKING_EXPRESSION, enabled=False)
-        # Get mapping info
-        mapping_gen = MappingGenerator(config)
-        self.match_fields = mapping_gen.get_match_fields_for_domain("default")
-        self.text_embedding_field = mapping_gen.get_text_embedding_field()
-        self.image_embedding_field = mapping_gen.get_image_embedding_field()
+        # Use constants from query_config
+        self.match_fields = DEFAULT_MATCH_FIELDS
+        self.text_embedding_field = TEXT_EMBEDDING_FIELD
+        self.image_embedding_field = IMAGE_EMBEDDING_FIELD
         # Query builder - use multi-language version
         self.query_builder = MultiLanguageQueryBuilder(
-            config=config,
-            index_name=config.es_index_name,
+            index_name=index_name,
+            match_fields=self.match_fields,
             text_embedding_field=self.text_embedding_field,
             image_embedding_field=self.image_embedding_field,
-            source_fields=config.query_config.source_fields
+            source_fields=SOURCE_FIELDS
         )
     def search(
@@ -127,7 +134,8 @@ class Searcher:
         context: Optional[RequestContext] = None,
         sort_by: Optional[str] = None,
         sort_order: Optional[str] = "desc",
-        debug: bool = False
+        debug: bool = False,
+        language: str = "zh",
     ) -> SearchResult:
         """
         Execute search query (外部友好格式).
@@ -154,8 +162,8 @@ class Searcher:
             context = create_request_context()
         # Always use config defaults (these are backend configuration, not user parameters)
-        enable_translation = self.config.query_config.enable_translation
-        enable_embedding = self.config.query_config.enable_text_embedding
+        enable_translation = ENABLE_TRANSLATION
+        enable_embedding = ENABLE_TEXT_EMBEDDING
         enable_rerank = False  # Temporarily disabled
         # Start timing
@@ -278,14 +286,6 @@ class Searcher:
                 min_score=min_score
             )
-            # Add SPU collapse if configured
-            if self.config.spu_config.enabled:
-                es_query = self.query_builder.add_spu_collapse(
-                    es_query,
-                    self.config.spu_config.spu_field,
-                    self.config.spu_config.inner_hits_size
-                )
-
             # Add facets for faceted search
             if facets:
                 facet_aggs = self.query_builder.build_facets(facets)
@@ -329,7 +329,7 @@ class Searcher:
         context.start_stage(RequestContextStage.ELASTICSEARCH_SEARCH)
         try:
             es_response = self.es_client.search(
-                index_name=self.config.es_index_name,
+                index_name=self.index_name,
                 body=body_for_es,
                 size=size,
                 from_=from_
@@ -374,7 +374,11 @@ class Searcher:
             max_score = es_response.get('hits', {}).get('max_score') or 0.0
             # Format results using ResultFormatter
-            formatted_results = ResultFormatter.format_search_results(es_hits, max_score)
+            formatted_results = ResultFormatter.format_search_results(
+                es_hits,
+                max_score,
+                language=language
+            )
             # Format facets
             standardized_facets = None
@@ -503,9 +507,9 @@ class Searcher:
         }
         # Add _source filtering if source_fields are configured
-        if self.config.query_config.source_fields:
+        if SOURCE_FIELDS:
             es_query["_source"] = {
-                "includes": self.config.query_config.source_fields
+                "includes": SOURCE_FIELDS
             }
         if filters or range_filters:
@@ -519,7 +523,7 @@ class Searcher:
         # Execute search
         es_response = self.es_client.search(
-            index_name=self.config.es_index_name,
+            index_name=self.index_name,
             body=es_query,
             size=size
         )
@@ -573,7 +577,7 @@ class Searcher:
         """
         try:
             response = self.es_client.client.get(
-                index=self.config.es_index_name,
+                index=self.index_name,
                 id=doc_id
             )
             return response.get('_source')
@@ -657,10 +661,11 @@ class Searcher:
     def _get_field_label(self, field: str) -> str:
         """获取字段的显示标签"""
-        # 从配置中获取字段标签
-        for field_config in self.config.fields:
-            if field_config.name == field:
-                # 尝试获取 label 属性
-                return getattr(field_config, 'label', field)
-        # 如果没有配置，返回字段名
-        return field
+        # 字段标签映射（简化版，不再从配置读取）
+        field_labels = {
+            "category1_name": "一级分类",
+            "category2_name": "二级分类",
+            "category3_name": "三级分类",
+            "specifications": "规格"
+        }
+        return field_labels.get(field, field)
@@ -172,6 +172,18 @@ class ESClient:
         Returns:
             Search results
         """
+        # Safety guard: collapse is no longer needed (index is already SPU-level).
+        # If any caller accidentally adds a collapse clause (e.g. on product_id),
+        # strip it here to avoid 400 errors like:
+        # "no mapping found for `product_id` in order to collapse on"
+        if isinstance(body, dict) and "collapse" in body:
+            logger.warning(
+                "Removing unsupported 'collapse' clause from ES query body: %s",
+                body.get("collapse")
+            )
+            body = dict(body)  # shallow copy to avoid mutating caller
+            body.pop("collapse", None)
+
         try:
             return self.client.search(
                 index=index_name,