索引隔离。不同的tenant_id用不同的索引

tangwang
1 parent 775db2b0
Showing 6 changed files with 92 additions and 49 deletions Show diff stats
api/routes/search.py
docs/亚马逊到店匠格式转换分析.md
indexer/bulk_indexing_service.py
indexer/incremental_service.py
indexer/mapping_generator.py
search/searcher.py
@@ -326,18 +326,36 @@ async def instant_search(
 @router.get("/{doc_id}", response_model=DocumentResponse)
-async def get_document(doc_id: str):
+async def get_document(doc_id: str, http_request: Request):
     """
     Get a single document by ID.
+    
+    Requires tenant_id in header (X-Tenant-ID) or query parameter (tenant_id).
     """
     try:
+        # Extract tenant_id (required)
+        tenant_id = http_request.headers.get('X-Tenant-ID')
+        if not tenant_id:
+            # Try to get from query string
+            from urllib.parse import parse_qs
+            query_string = http_request.url.query
+            if query_string:
+                params = parse_qs(query_string)
+                tenant_id = params.get('tenant_id', [None])[0]
+        
+        if not tenant_id:
+            raise HTTPException(
+                status_code=400,
+                detail="tenant_id is required. Provide it via header 'X-Tenant-ID' or query parameter 'tenant_id'"
+            )
+        
         from api.app import get_searcher
         searcher = get_searcher()
-        doc = searcher.get_document(doc_id)
+        doc = searcher.get_document(tenant_id=tenant_id, doc_id=doc_id)
         if doc is None:
-            raise HTTPException(status_code=404, detail=f"Document {doc_id} not found")
+            raise HTTPException(status_code=404, detail=f"Document {doc_id} not found for tenant {tenant_id}")
         return DocumentResponse(id=doc_id, source=doc)
@@ -368,3 +368,4 @@ python scripts/amazon_xlsx_to_shoplazza_xlsx.py \
+
@@ -10,7 +10,7 @@ from sqlalchemy import Engine
 from utils.es_client import ESClient
 from indexer.spu_transformer import SPUTransformer
 from indexer.bulk_indexer import BulkIndexer
-from indexer.mapping_generator import load_mapping, delete_index_if_exists, DEFAULT_INDEX_NAME
+from indexer.mapping_generator import load_mapping, delete_index_if_exists, get_tenant_index_name
 from indexer.indexer_logger import (
     get_indexer_logger, log_index_request, log_index_result, log_bulk_index_batch
 )
@@ -33,13 +33,16 @@ class BulkIndexingService:
         """
         self.db_engine = db_engine
         self.es_client = es_client
-        self.index_name = DEFAULT_INDEX_NAME
+        # Index name is now generated dynamically per tenant, no longer stored here
     def bulk_index(self, tenant_id: str, recreate_index: bool = False, batch_size: int = 500) -> Dict[str, Any]:
         """执行全量索引"""
         import time
         start_time = time.time()
+        # Generate tenant-specific index name
+        index_name = get_tenant_index_name(tenant_id)
+        
         # 记录请求开始
         log_index_request(
             indexer_logger,
@@ -48,7 +51,7 @@ class BulkIndexingService:
             request_params={
                 'recreate_index': recreate_index,
                 'batch_size': batch_size,
-                'index_name': self.index_name
+                'index_name': index_name
             }
         )
@@ -61,45 +64,45 @@ class BulkIndexingService:
                     'index_type': 'bulk',
                     'tenant_id': tenant_id,
                     'operation': 'load_mapping',
-                    'index_name': self.index_name
+                    'index_name': index_name
                 }
             )
             mapping = load_mapping()
             # 2. 处理索引（删除并重建或创建）
             if recreate_index:
-                logger.info(f"[BulkIndexing] Recreating index: {self.index_name}")
+                logger.info(f"[BulkIndexing] Recreating index: {index_name}")
                 indexer_logger.info(
-                    f"Recreating index: {self.index_name}",
+                    f"Recreating index: {index_name}",
                     extra={
                         'index_type': 'bulk',
                         'tenant_id': tenant_id,
                         'operation': 'recreate_index',
-                        'index_name': self.index_name
+                        'index_name': index_name
                     }
                 )
-                if self.es_client.index_exists(self.index_name):
-                    if delete_index_if_exists(self.es_client, self.index_name):
-                        logger.info(f"[BulkIndexing] Deleted existing index: {self.index_name}")
+                if self.es_client.index_exists(index_name):
+                    if delete_index_if_exists(self.es_client, index_name):
+                        logger.info(f"[BulkIndexing] Deleted existing index: {index_name}")
                     else:
-                        raise Exception(f"Failed to delete index: {self.index_name}")
+                        raise Exception(f"Failed to delete index: {index_name}")
-            if not self.es_client.index_exists(self.index_name):
-                logger.info(f"[BulkIndexing] Creating index: {self.index_name}")
+            if not self.es_client.index_exists(index_name):
+                logger.info(f"[BulkIndexing] Creating index: {index_name}")
                 indexer_logger.info(
-                    f"Creating index: {self.index_name}",
+                    f"Creating index: {index_name}",
                     extra={
                         'index_type': 'bulk',
                         'tenant_id': tenant_id,
                         'operation': 'create_index',
-                        'index_name': self.index_name
+                        'index_name': index_name
                     }
                 )
-                if not self.es_client.create_index(self.index_name, mapping):
-                    raise Exception(f"Failed to create index: {self.index_name}")
-                logger.info(f"[BulkIndexing] Created index: {self.index_name}")
+                if not self.es_client.create_index(index_name, mapping):
+                    raise Exception(f"Failed to create index: {index_name}")
+                logger.info(f"[BulkIndexing] Created index: {index_name}")
             else:
-                logger.info(f"[BulkIndexing] Index already exists: {self.index_name}")
+                logger.info(f"[BulkIndexing] Index already exists: {index_name}")
             # 3. 转换数据
             logger.info(f"[BulkIndexing] Transforming data for tenant_id={tenant_id}")
@@ -109,7 +112,7 @@ class BulkIndexingService:
                     'index_type': 'bulk',
                     'tenant_id': tenant_id,
                     'operation': 'transform_data',
-                    'index_name': self.index_name
+                    'index_name': index_name
                 }
             )
             transformer = SPUTransformer(self.db_engine, tenant_id)
@@ -126,7 +129,7 @@ class BulkIndexingService:
                     success_count=0,
                     failed_count=0,
                     elapsed_time=elapsed_time,
-                    index_name=self.index_name
+                    index_name=index_name
                 )
                 return {
                     "success": True,
@@ -135,7 +138,7 @@ class BulkIndexingService:
                     "failed": 0,
                     "elapsed_time": elapsed_time,
                     "message": "No documents to index",
-                    "index_name": self.index_name,
+                    "index_name": index_name,
                     "tenant_id": tenant_id
                 }
@@ -147,13 +150,13 @@ class BulkIndexingService:
                     'tenant_id': tenant_id,
                     'operation': 'transform_complete',
                     'total_count': len(documents),
-                    'index_name': self.index_name
+                    'index_name': index_name
                 }
             )
             # 4. 批量导入
             logger.info(f"[BulkIndexing] Indexing {len(documents)} documents (batch_size={batch_size})")
-            indexer = BulkIndexer(self.es_client, self.index_name, batch_size=batch_size, max_retries=3)
+            indexer = BulkIndexer(self.es_client, index_name, batch_size=batch_size, max_retries=3)
             results = indexer.index_documents(
                 documents,
                 id_field="spu_id",
@@ -171,7 +174,7 @@ class BulkIndexingService:
                 success_count=results['success'],
                 failed_count=results['failed'],
                 elapsed_time=elapsed_time,
-                index_name=self.index_name,
+                index_name=index_name,
                 errors=results.get('errors', [])
             )
@@ -187,7 +190,7 @@ class BulkIndexingService:
                 "indexed": results['success'],
                 "failed": results['failed'],
                 "elapsed_time": elapsed_time,
-                "index_name": self.index_name,
+                "index_name": index_name,
                 "tenant_id": tenant_id
             }
@@ -203,7 +206,7 @@ class BulkIndexingService:
                     'operation': 'request_failed',
                     'error': error_msg,
                     'elapsed_time': elapsed_time,
-                    'index_name': self.index_name
+                    'index_name': index_name
                 },
                 exc_info=True
             )
@@ -9,7 +9,7 @@ import numpy as np
 from sqlalchemy import text, bindparam
 from indexer.indexing_utils import load_category_mapping, create_document_transformer
 from indexer.bulk_indexer import BulkIndexer
-from indexer.mapping_generator import DEFAULT_INDEX_NAME
+from indexer.mapping_generator import get_tenant_index_name
 from indexer.indexer_logger import (
     get_indexer_logger, log_index_request, log_index_result, log_spu_processing
 )
@@ -393,7 +393,7 @@ class IncrementalIndexerService:
         es_client,
         tenant_id: str,
         spu_ids: List[str],
-        index_name: str = DEFAULT_INDEX_NAME,
+        index_name: str = None,
         batch_size: int = 100,
         delete_spu_ids: List[str] = None
     ) -> Dict[str, Any]:
@@ -408,13 +408,16 @@ class IncrementalIndexerService:
             es_client: Elasticsearch客户端
             tenant_id: 租户ID
             spu_ids: SPU ID列表（要索引的）
-            index_name: 索引名称
+            index_name: 索引名称（可选，如果不提供则根据tenant_id自动生成）
             batch_size: 批量写入ES的批次大小
             delete_spu_ids: 显式指定要删除的SPU ID列表（可选）
         Returns:
             包含成功/失败列表的字典，以及删除结果
         """
+        # Generate tenant-specific index name if not provided
+        if index_name is None:
+            index_name = get_tenant_index_name(tenant_id)
         # 去重但保持顺序（避免重复DB/翻译/embedding/写ES）
         if spu_ids:
             spu_ids = list(dict.fromkeys(spu_ids))
@@ -11,13 +11,26 @@ from pathlib import Path
 logger = logging.getLogger(__name__)
-# Default index name
+# Default index name (deprecated, use get_tenant_index_name instead)
 DEFAULT_INDEX_NAME = "search_products"
 # Default mapping file path
 DEFAULT_MAPPING_FILE = Path(__file__).parent.parent / "mappings" / "search_products.json"
+def get_tenant_index_name(tenant_id: str) -> str:
+    """
+    Generate index name for a tenant.
+    
+    Args:
+        tenant_id: Tenant ID
+        
+    Returns:
+        Index name in format: search_products_tenant_{tenant_id}
+    """
+    return f"search_products_tenant_{tenant_id}"
+
+
 def load_mapping(mapping_file: str = None) -> Dict[str, Any]:
     """
     Load Elasticsearch mapping from JSON file.
@@ -20,6 +20,7 @@ from config.utils import get_match_fields_for_index
 from context.request_context import RequestContext, RequestContextStage, create_request_context
 from api.models import FacetResult, FacetValue, FacetConfig
 from api.result_formatter import ResultFormatter
+from indexer.mapping_generator import get_tenant_index_name
 logger = logging.getLogger(__name__)
@@ -93,7 +94,7 @@ class Searcher:
         """
         self.es_client = es_client
         self.config = config
-        self.index_name = config.es_index_name
+        # Index name is now generated dynamically per tenant, no longer stored here
         self.query_parser = query_parser or QueryParser(config)
         # Initialize components
@@ -107,8 +108,9 @@ class Searcher:
         self.source_fields = config.query_config.source_fields or []
         # Query builder - simplified single-layer architecture
+        # index_name is no longer needed in query builder since we use tenant-specific indices
         self.query_builder = ESQueryBuilder(
-            index_name=self.index_name,
+            index_name="",  # Not used, kept for backward compatibility
             match_fields=self.match_fields,
             text_embedding_field=self.text_embedding_field,
             image_embedding_field=self.image_embedding_field,
@@ -271,10 +273,10 @@ class Searcher:
         # Step 3: Query building
         context.start_stage(RequestContextStage.QUERY_BUILDING)
         try:
-            # Add tenant_id to filters (required)
-            if filters is None:
-                filters = {}
-            filters['tenant_id'] = tenant_id
+            # Generate tenant-specific index name
+            index_name = get_tenant_index_name(tenant_id)
+            
+            # No longer need to add tenant_id to filters since each tenant has its own index
             es_query = self.query_builder.build_query(
                 query_text=parsed_query.rewritten_query or parsed_query.normalized_query,
@@ -332,8 +334,9 @@ class Searcher:
         # Step 4: Elasticsearch search
         context.start_stage(RequestContextStage.ELASTICSEARCH_SEARCH)
         try:
+            # Use tenant-specific index name
             es_response = self.es_client.search(
-                index_name=self.index_name,
+                index_name=index_name,
                 body=body_for_es,
                 size=size,
                 from_=from_
@@ -496,10 +499,10 @@ class Searcher:
         if image_vector is None:
             raise ValueError(f"Failed to encode image: {image_url}")
-        # Add tenant_id to filters (required)
-        if filters is None:
-            filters = {}
-        filters['tenant_id'] = tenant_id
+        # Generate tenant-specific index name
+        index_name = get_tenant_index_name(tenant_id)
+        
+        # No longer need to add tenant_id to filters since each tenant has its own index
         # Build KNN query
         es_query = {
@@ -529,7 +532,7 @@ class Searcher:
         # Execute search
         es_response = self.es_client.search(
-            index_name=self.index_name,
+            index_name=index_name,
             body=es_query,
             size=size
         )
@@ -576,24 +579,26 @@ class Searcher:
         """
         return self.query_builder.get_domain_summary()
-    def get_document(self, doc_id: str) -> Optional[Dict[str, Any]]:
+    def get_document(self, tenant_id: str, doc_id: str) -> Optional[Dict[str, Any]]:
         """
         Get single document by ID.
         Args:
+            tenant_id: Tenant ID (required to determine which index to query)
             doc_id: Document ID
         Returns:
             Document or None if not found
         """
         try:
+            index_name = get_tenant_index_name(tenant_id)
             response = self.es_client.client.get(
-                index=self.index_name,
+                index=index_name,
                 id=doc_id
             )
             return response.get('_source')
         except Exception as e:
-            logger.error(f"Failed to get document {doc_id}: {e}", exc_info=True)
+            logger.error(f"Failed to get document {doc_id} from tenant {tenant_id}: {e}", exc_info=True)
             return None
     def _standardize_facets(