api/routes/indexer.py

- 新增批量索引接口: POST /indexer/bulk - 全量索引功能 - SPU接口改进: POST /indexer/spus - 支持批量获取SPU文档（最多100个）新增全量索引服务 indexer/bulk_indexing_service.py docs/搜索API对接指南.md - 新增索引接口文档: 详细的批量索引和SPU索引接口说明 - 请求示例: 提供完整的curl命令示例

api/routes/indexer.py
- 新增批量索引接口: POST /indexer/bulk - 全量索引功能 - SPU接口改进: POST /indexer/spus - 支持批量获取SPU文档（最多100个）新增全量索引服务 indexer/bulk_indexing_service.py docs/搜索API对接指南.md - 新增索引接口文档: 详细的批量索引和SPU索引接口说明 - 请求示例: 提供完整的curl命令示例
tangwang
1 parent 0064e946
Showing 8 changed files with 526 additions and 494 deletions Show diff stats
COMMIT_MESSAGE.md
api/app.py
api/routes/indexer.py
docs/搜索API对接指南.md
indexer/bulk_indexing_service.py
indexer/incremental_service.py
indexer/indexing_utils.py
indexer/spu_transformer.py
@@ -1,154 +0,0 @@
-# 本次修改总结
-
-## 最终状态
-
-### 1. 增量数据获取服务
-
-**新增文件**：
-- `indexer/incremental_service.py`: 增量索引服务，提供单个SPU数据获取
-- `api/routes/indexer.py`: 增量索引API路由
-- `indexer/test_indexing.py`: 索引功能测试脚本
-
-**功能**：
-- 提供 `GET /indexer/spu/{spu_id}?tenant_id={tenant_id}` 接口，返回单个SPU的ES文档数据
-- 服务启动时预加载分类映射（全局共享），提高性能
-- 支持按需加载租户配置和搜索配置
-
-### 2. 公共文档转换器
-
-**新增文件**：
-- `indexer/document_transformer.py`: SPU文档转换器，提取全量和增量共用的转换逻辑
-
-**功能**：
-- 统一了全量索引（SPUTransformer）和增量索引（IncrementalIndexerService）的文档转换逻辑
-- 消除了约300行重复代码
-- 支持根据租户配置进行语言处理和翻译
-
-### 3. 租户配置系统
-
-**配置位置**：
-- 租户配置合并到统一配置文件 `config/config.yaml` 的 `tenant_config` 部分
-- 删除了独立的 `config/tenant_config.json` 文件
-
-**配置结构**：
-```yaml
-tenant_config:
-  default:
-    primary_language: "zh"
-    translate_to_en: true
-    translate_to_zh: false
-  tenants:
-    "162":
-      primary_language: "zh"
-      translate_to_en: false  # 翻译关闭
-      translate_to_zh: false
-```
-
-**功能**：
-- 每个租户可配置主语言和翻译选项
-- 租户162配置为翻译关闭（用于测试）
-- 未配置的租户使用默认配置
-
-### 4. 翻译功能集成
-
-**翻译模块增强**：
-- `query/translator.py`: 支持提示词参数，作为DeepL API的`context`参数传递
-- 修复了重复的executor初始化代码
-- 统一使用logger替代print语句
-
-**翻译提示词配置**：
-- 在 `config/config.yaml` 的 `translation_prompts` 部分配置
-- 支持中英文提示词：
-  - `product_title_zh/en`: 商品标题翻译提示词
-  - `query_zh/en`: 查询翻译提示词
-  - `default_zh/en`: 默认翻译用词
-
-**翻译模式**：
-- **索引场景**：同步翻译，等待结果返回，使用缓存避免重复翻译
-- **查询场景**：异步翻译，立即返回缓存结果，后台翻译缺失项
-
-**DeepL Context参数**：
-- 提示词作为DeepL API的`context`参数传递（不参与翻译，仅提供上下文）
-- Context中的字符不计入DeepL计费
-
-### 5. 代码重构
-
-**消除冗余**：
-- 提取公共转换逻辑到 `SPUDocumentTransformer`
-- `SPUTransformer` 和 `IncrementalIndexerService` 都使用公共转换器
-- 移除了重复的 `_transform_spu_to_doc` 和 `_transform_sku_row` 方法
-
-**架构优化**：
-- 全量和增量索引共用同一转换逻辑
-- 分类映射在服务启动时预加载（全局共享）
-- 租户配置按需加载（支持热更新）
-
-### 6. 测试
-
-**测试文件位置**（遵循模块化原则）：
-- `indexer/test_indexing.py`: 索引功能测试（全量、增量、租户配置、文档转换器）
-- `query/test_translation.py`: 翻译功能测试（同步、异步、缓存、Context参数）
-
-### 7. 文档更新
-
-- `docs/索引数据接口文档.md`: 更新了租户配置说明，从独立JSON文件改为统一配置文件
-- `docs/翻译功能测试说明.md`: 新增翻译功能测试说明文档
-
-## 修改的文件
-
-### 新增文件
-- `indexer/incremental_service.py`
-- `indexer/document_transformer.py`
-- `indexer/test_indexing.py`
-- `api/routes/indexer.py`
-- `query/test_translation.py`
-- `config/tenant_config_loader.py` (重构，从JSON改为YAML)
-- `docs/翻译功能测试说明.md`
-
-### 修改文件
-- `config/config.yaml`: 添加租户配置和翻译提示词配置
-- `config/config_loader.py`: 支持租户配置加载
-- `config/tenant_config_loader.py`: 从统一配置文件加载租户配置
-- `indexer/spu_transformer.py`: 使用公共转换器，集成翻译服务
-- `indexer/incremental_service.py`: 使用公共转换器，集成翻译服务
-- `query/translator.py`: 支持提示词作为context参数，修复冗余代码
-- `query/query_parser.py`: 使用翻译提示词
-- `api/app.py`: 注册增量索引路由，初始化增量服务
-- `docs/索引数据接口文档.md`: 更新租户配置说明
-
-### 删除文件
-- `config/tenant_config.json`: 合并到统一配置文件
-
-## 测试验证
-
-### 租户162测试（翻译关闭）
-- 全量索引：验证翻译功能关闭，title_en为None
-- 增量索引：验证翻译功能关闭，title_en为None
-- 文档转换器：验证根据租户配置正确处理翻译
-
-### 其他租户测试（翻译开启）
-- 验证翻译功能正常工作
-- 验证提示词正确使用
-
-## 架构设计
-
-### 数据流
-```
-MySQL数据
-    ↓
-SPUTransformer / IncrementalIndexerService (数据加载层)
-    ↓
-SPUDocumentTransformer (公共转换层)
-    ↓
-ES文档 (输出)
-```
-
-### 配置层次
-1. **索引配置** (`config/config.yaml`): 搜索行为配置
-2. **租户配置** (`config/config.yaml` 的 `tenant_config` 部分): 数据转换配置
-
-### 性能优化
-1. 公共数据预加载：分类映射在服务启动时一次性加载
-2. 配置按需加载：租户配置和搜索配置按需加载，支持热更新
-3. 翻译缓存：索引时使用缓存避免重复翻译
-
@@ -24,12 +24,15 @@ from slowapi.util import get_remote_address
 from slowapi.errors import RateLimitExceeded
 # Configure logging with better formatting
+import pathlib
+log_dir = pathlib.Path('logs')
+log_dir.mkdir(exist_ok=True)
 logging.basicConfig(
     level=logging.INFO,
     format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
     handlers=[
         logging.StreamHandler(),
-        logging.FileHandler('/tmp/search_engine_api.log', mode='a')
+        logging.FileHandler(log_dir / 'api.log', mode='a', encoding='utf-8')
     ]
 )
 logger = logging.getLogger(__name__)
@@ -54,6 +57,7 @@ _searcher: Optional[Searcher] = None
 _query_parser: Optional[QueryParser] = None
 _config = None
 _incremental_service: Optional[IncrementalIndexerService] = None
+_bulk_indexing_service = None
 def init_service(es_host: str = "http://localhost:9200"):
@@ -63,7 +67,7 @@ def init_service(es_host: str = &quot;http://localhost:9200&quot;):
     Args:
         es_host: Elasticsearch host URL
     """
-    global _es_client, _searcher, _query_parser, _config, _incremental_service
+    global _es_client, _searcher, _query_parser, _config, _incremental_service, _bulk_indexing_service
     start_time = time.time()
     logger.info("Initializing search service (multi-tenant)")
@@ -96,16 +100,20 @@ def init_service(es_host: str = &quot;http://localhost:9200&quot;):
     logger.info("Initializing searcher...")
     _searcher = Searcher(_es_client, _config, _query_parser)
-    # Initialize incremental indexer service (if DB config is available)
+    # Initialize indexing services (if DB config is available)
     try:
-        db_host = DB_CONFIG.get('host')
-        db_port = DB_CONFIG.get('port', 3306)
-        db_database = DB_CONFIG.get('database')
-        db_username = DB_CONFIG.get('username')
-        db_password = DB_CONFIG.get('password')
+        from utils.db_connector import create_db_connection
+        from indexer.incremental_service import IncrementalIndexerService
+        from indexer.bulk_indexing_service import BulkIndexingService
+        
+        db_host = os.getenv('DB_HOST')
+        db_port = int(os.getenv('DB_PORT', 3306))
+        db_database = os.getenv('DB_DATABASE')
+        db_username = os.getenv('DB_USERNAME')
+        db_password = os.getenv('DB_PASSWORD')
         if all([db_host, db_database, db_username, db_password]):
-            logger.info("Initializing incremental indexer service...")
+            logger.info("Initializing database connection for indexing services...")
             db_engine = create_db_connection(
                 host=db_host,
                 port=db_port,
@@ -113,15 +121,22 @@ def init_service(es_host: str = &quot;http://localhost:9200&quot;):
                 username=db_username,
                 password=db_password
             )
+            
+            # Initialize incremental service
             _incremental_service = IncrementalIndexerService(db_engine)
             logger.info("Incremental indexer service initialized")
+            
+            # Initialize bulk indexing service
+            _bulk_indexing_service = BulkIndexingService(db_engine, _es_client)
+            logger.info("Bulk indexing service initialized")
         else:
-            logger.warning("Database configuration incomplete, incremental indexer service will not be available")
+            logger.warning("Database config incomplete, indexing services will not be available")
             logger.warning("Required: DB_HOST, DB_DATABASE, DB_USERNAME, DB_PASSWORD")
     except Exception as e:
-        logger.warning(f"Failed to initialize incremental indexer service: {e}")
-        logger.warning("Incremental indexer endpoints will not be available")
+        logger.warning(f"Failed to initialize indexing services: {e}")
+        logger.warning("Indexing endpoints will not be available")
         _incremental_service = None
+        _bulk_indexing_service = None
     elapsed = time.time() - start_time
     logger.info(f"Search service ready! (took {elapsed:.2f}s) | Index: {_config.es_index_name}")
@@ -162,6 +177,11 @@ def get_incremental_service() -&gt; Optional[IncrementalIndexerService]:
     return _incremental_service
+def get_bulk_indexing_service():
+    """Get bulk indexing service instance."""
+    return _bulk_indexing_service
+
+
 # Create FastAPI app with enhanced configuration
 app = FastAPI(
     title="E-Commerce Search API",
@@ -207,15 +227,14 @@ app.add_middleware(
 async def startup_event():
     """Initialize service on startup."""
     es_host = os.getenv("ES_HOST", "http://localhost:9200")
-
     logger.info("Starting E-Commerce Search API (Multi-Tenant)")
     logger.info(f"Elasticsearch Host: {es_host}")
-
+    
     try:
         init_service(es_host=es_host)
         logger.info("Service initialized successfully")
     except Exception as e:
-        logger.error(f"Failed to initialize service: {e}")
+        logger.error(f"Failed to initialize service: {e}", exc_info=True)
         logger.warning("Service will start but may not function correctly")
 """
-增量索引API路由。
+索引API路由。
-提供单个SPU数据获取接口，用于增量更新ES索引。
+提供全量和增量索引接口，供外部Java程序调用。
 """
-from fastapi import APIRouter, HTTPException, Query, Request
-from typing import Optional
+from fastapi import APIRouter, HTTPException
+from typing import List
+from pydantic import BaseModel
 import logging
-from ..models import ErrorResponse
-
 logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/indexer", tags=["indexer"])
-@router.get("/spu/{spu_id}")
-async def get_spu_document(
-    spu_id: str,
-    tenant_id: str = Query(..., description="租户ID"),
-    request: Request = None
-):
-    """
-    获取单个SPU的ES文档数据（用于增量索引更新）。
+class BulkIndexRequest(BaseModel):
+    tenant_id: str
+    recreate_index: bool = False
+    batch_size: int = 500
+
-    功能说明：
-    - 根据 tenant_id 和 spu_id 查询MySQL数据库
-    - 返回该SPU的完整ES文档数据（JSON格式）
-    - 外部Java程序可以调用此接口获取数据后推送到ES
+class BatchSpuRequest(BaseModel):
+    tenant_id: str
+    spu_ids: List[str]
-    参数：
-    - spu_id: SPU ID（路径参数）
-    - tenant_id: 租户ID（查询参数，必需）
-    返回：
-    - 成功：返回ES文档JSON对象
-    - SPU不存在或已删除：返回404
-    - 其他错误：返回500
+@router.post("/bulk")
+async def bulk_index(request: BulkIndexRequest):
+    """全量索引接口"""
+    try:
+        from ..app import get_bulk_indexing_service
+        service = get_bulk_indexing_service()
+        if service is None:
+            raise HTTPException(status_code=503, detail="Bulk indexing service is not initialized")
+        return service.bulk_index(
+            tenant_id=request.tenant_id,
+            recreate_index=request.recreate_index,
+            batch_size=request.batch_size
+        )
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error(f"Error in bulk indexing for tenant_id={request.tenant_id}: {e}", exc_info=True)
+        raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")
-    示例请求：
-    ```
-    GET /indexer/spu/123?tenant_id=1
-    ```
-    示例响应：
-    ```json
-    {
-        "tenant_id": "1",
-        "spu_id": "123",
-        "title_zh": "商品标题",
-        "brief_zh": "商品简介",
-        "description_zh": "商品描述",
-        "vendor_zh": "供应商",
-        "tags": ["标签1", "标签2"],
-        "category_path_zh": "类目1/类目2/类目3",
-        "category1_name": "类目1",
-        "category2_name": "类目2",
-        "category3_name": "类目3",
-        "category_id": "100",
-        "category_level": 3,
-        "min_price": 99.99,
-        "max_price": 199.99,
-        "compare_at_price": 299.99,
-        "sales": 1000,
-        "total_inventory": 500,
-        "skus": [...],
-        "specifications": [...],
-        ...
-    }
-    ```
-    """
+@router.post("/spus")
+async def get_spu_documents(request: BatchSpuRequest):
+    """获取SPU文档接口（支持单个或批量）"""
     try:
         from ..app import get_incremental_service
-        
-        # 获取增量服务实例
+        if not request.spu_ids:
+            raise HTTPException(status_code=400, detail="spu_ids cannot be empty")
+        if len(request.spu_ids) > 100:
+            raise HTTPException(status_code=400, detail="Maximum 100 SPU IDs allowed per request")
         service = get_incremental_service()
         if service is None:
-            raise HTTPException(
-                status_code=503,
-                detail="Incremental indexer service is not initialized. Please check database connection."
-            )
-
-        # 获取SPU文档
-        doc = service.get_spu_document(tenant_id=tenant_id, spu_id=spu_id)
-        
-        if doc is None:
-            raise HTTPException(
-                status_code=404,
-                detail=f"SPU {spu_id} not found for tenant_id={tenant_id} or has been deleted"
-            )
-
-        return doc
-
+            raise HTTPException(status_code=503, detail="Incremental indexer service is not initialized")
+        success_list, failed_list = [], []
+        for spu_id in request.spu_ids:
+            try:
+                doc = service.get_spu_document(tenant_id=request.tenant_id, spu_id=spu_id)
+                (success_list if doc else failed_list).append({
+                    "spu_id": spu_id,
+                    "document": doc
+                } if doc else {
+                    "spu_id": spu_id,
+                    "error": "SPU not found or deleted"
+                })
+            except Exception as e:
+                failed_list.append({"spu_id": spu_id, "error": str(e)})
+        return {
+            "success": success_list,
+            "failed": failed_list,
+            "total": len(request.spu_ids),
+            "success_count": len(success_list),
+            "failed_count": len(failed_list)
+        }
     except HTTPException:
         raise
     except Exception as e:
-        logger.error(f"Error getting SPU document for tenant_id={tenant_id}, spu_id={spu_id}: {e}", exc_info=True)
+        logger.error(f"Error getting SPU documents for tenant_id={request.tenant_id}: {e}", exc_info=True)
         raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")
 @router.get("/health")
 async def indexer_health_check():
-    """
-    检查增量索引服务健康状态。
-
-    返回：
-    - 服务是否可用
-    - 数据库连接状态
-    - 预加载数据状态
-    """
+    """检查索引服务健康状态"""
     try:
         from ..app import get_incremental_service
-        
+        from sqlalchemy import text
         service = get_incremental_service()
         if service is None:
-            return {
-                "status": "unavailable",
-                "message": "Incremental indexer service is not initialized",
-                "database": "unknown",
-                "preloaded_data": {
-                    "category_mappings": 0
-                }
-            }
-
-        # 检查数据库连接
+            return {"status": "unavailable", "database": "unknown", "preloaded_data": {"category_mappings": 0}}
         try:
-            from sqlalchemy import text
             with service.db_engine.connect() as conn:
                 conn.execute(text("SELECT 1"))
             db_status = "connected"
         except Exception as e:
             db_status = f"disconnected: {str(e)}"
-
         return {
             "status": "available",
             "database": db_status,
-            "preloaded_data": {
-                "category_mappings": len(service.category_id_to_name)
-            }
+            "preloaded_data": {"category_mappings": len(service.category_id_to_name)}
         }
-
     except Exception as e:
         logger.error(f"Error checking indexer health: {e}", exc_info=True)
-        return {
-            "status": "error",
-            "message": str(e)
-        }
+        return {"status": "error", "message": str(e)}
@@ -76,12 +76,191 @@ curl -X POST &quot;http://120.76.41.98:6002/search/&quot; \
 ## 接口概览
-| 接口 | HTTP Method | Endpoint | 
-|------|------|------|
-| 搜索 | POST | `/search/` | 
-| 搜索建议（框架，暂未实现） | GET | `/search/suggestions` | 
-| 获取文档 | GET | `/search/{doc_id}` | 
-| 健康检查 | GET | `/admin/health` | 
+| 接口 | HTTP Method | Endpoint | 说明 |
+|------|------|------|------|
+| 搜索 | POST | `/search/` | 执行搜索查询 |
+| 全量索引 | POST | `/indexer/bulk` | 全量索引接口 |
+| SPU索引 | POST | `/indexer/spus` | 获取SPU文档（支持单个或批量） |
+| 索引健康检查 | GET | `/indexer/health` | 检查索引服务状态 |
+| 搜索建议（框架，暂未实现） | GET | `/search/suggestions` | 搜索建议 |
+| 获取文档 | GET | `/search/{doc_id}` | 获取单个文档 |
+| 健康检查 | GET | `/admin/health` | 服务健康检查 |
+
+---
+
+## 索引接口
+
+### 全量索引接口
+
+- **端点**: `POST /indexer/bulk`
+- **描述**: 将指定租户的所有SPU数据导入到ES索引
+
+#### 请求参数
+
+```json
+{
+  "tenant_id": "162",
+  "recreate_index": false,
+  "batch_size": 500
+}
+```
+
+| 参数 | 类型 | 必填 | 默认值 | 说明 |
+|------|------|------|--------|------|
+| `tenant_id` | string | Y | - | 租户ID |
+| `recreate_index` | boolean | N | false | 是否重建索引（删除旧索引后创建新索引） |
+| `batch_size` | integer | N | 500 | 批量导入大小 |
+
+#### 响应格式
+
+**成功响应（200 OK）**:
+```json
+{
+  "success": true,
+  "total": 1000,
+  "indexed": 1000,
+  "failed": 0,
+  "elapsed_time": 12.34,
+  "index_name": "search_products",
+  "tenant_id": "162"
+}
+```
+
+**错误响应**:
+- `400 Bad Request`: 参数错误
+- `503 Service Unavailable`: 服务未初始化
+
+#### 请求示例
+
+**首次索引（重建索引）**:
+```bash
+curl -X POST "http://localhost:6002/indexer/bulk" \
+  -H "Content-Type: application/json" \
+  -d '{
+    "tenant_id": "162",
+    "recreate_index": true,
+    "batch_size": 500
+  }'
+```
+
+**查看日志**:
+```bash
+# 查看API日志（包含索引操作日志）
+tail -f logs/api.log
+
+# 或者查看所有日志文件
+tail -f logs/*.log
+```
+
+**增量更新（不重建索引）**:
+```bash
+curl -X POST "http://localhost:6002/indexer/bulk" \
+  -H "Content-Type: application/json" \
+  -d '{
+    "tenant_id": "162",
+    "recreate_index": false,
+    "batch_size": 500
+  }'
+```
+
+---
+
+### SPU索引接口
+
+- **端点**: `POST /indexer/spus`
+- **描述**: 获取SPU的ES文档数据（支持单个或批量）
+
+#### 请求参数
+
+```json
+{
+  "tenant_id": "162",
+  "spu_ids": ["123", "456", "789"]
+}
+```
+
+| 参数 | 类型 | 必填 | 说明 |
+|------|------|------|------|
+| `tenant_id` | string | Y | 租户ID |
+| `spu_ids` | array[string] | Y | SPU ID列表（1-100个） |
+
+#### 响应格式
+
+```json
+{
+  "success": [
+    {
+      "spu_id": "123",
+      "document": {
+        "tenant_id": "162",
+        "spu_id": "123",
+        "title_zh": "商品标题",
+        ...
+      }
+    },
+    {
+      "spu_id": "456",
+      "document": {...}
+    }
+  ],
+  "failed": [
+    {
+      "spu_id": "789",
+      "error": "SPU not found or deleted"
+    }
+  ],
+  "total": 3,
+  "success_count": 2,
+  "failed_count": 1
+}
+```
+
+#### 请求示例
+
+**单个SPU**:
+```bash
+curl -X POST "http://localhost:6002/indexer/spus" \
+  -H "Content-Type: application/json" \
+  -d '{
+    "tenant_id": "162",
+    "spu_ids": ["123"]
+  }'
+```
+
+**批量SPU**:
+```bash
+curl -X POST "http://localhost:6002/indexer/spus" \
+  -H "Content-Type: application/json" \
+  -d '{
+    "tenant_id": "162",
+    "spu_ids": ["123", "456", "789"]
+  }'
+```
+
+---
+
+### 索引健康检查接口
+
+- **端点**: `GET /indexer/health`
+- **描述**: 检查索引服务的健康状态
+
+#### 响应格式
+
+```json
+{
+  "status": "available",
+  "database": "connected",
+  "preloaded_data": {
+    "category_mappings": 150
+  }
+}
+```
+
+#### 请求示例
+
+```bash
+curl -X GET "http://localhost:6002/indexer/health"
+```
 ---
@@ -1163,8 +1342,10 @@ curl &quot;http://localhost:6002/search/12345&quot;
 | 分析器 | 语言 | 描述 |
 |--------|------|------|
-| `hanlp_index` | 中文 | 中文索引分析器（用于中文字段） |
-| `hanlp_standard` | 中文 | 中文查询分析器（用于中文字段） |
+| `index_ansj` | 中文 | 中文索引分析器（用于中文字段） |
+| `query_ansj` | 中文 | 中文查询分析器（用于中文字段） |
+| `hanlp_index`（暂不支持） | 中文 | 中文索引分析器（用于中文字段） |
+| `hanlp_standard`（暂不支持） | 中文 | 中文查询分析器（用于中文字段） |
 | `english` | 英文 | 标准英文分析器（用于英文字段） |
 | `lowercase` | - | 小写标准化器（用于keyword子字段） |
@@ -1180,4 +1361,3 @@ curl &quot;http://localhost:6002/search/12345&quot;
 | `date` | `date` | 日期时间 |
 | `nested` | `nested` | 嵌套对象（specifications, skus, image_embedding） |
 | `dense_vector` | `dense_vector` | 向量字段（title_embedding，仅用于搜索） |
-
@@ -0,0 +1,108 @@
+"""
+全量索引服务。
+
+提供全量索引功能，将指定租户的所有SPU数据导入到ES。
+"""
+
+import logging
+from typing import Dict, Any
+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
+
+logger = logging.getLogger(__name__)
+
+
+class BulkIndexingService:
+    """全量索引服务，提供批量导入功能。"""
+
+    def __init__(self, db_engine: Engine, es_client: ESClient):
+        """
+        初始化全量索引服务。
+
+        Args:
+            db_engine: SQLAlchemy database engine
+            es_client: Elasticsearch client
+        """
+        self.db_engine = db_engine
+        self.es_client = es_client
+        self.index_name = DEFAULT_INDEX_NAME
+
+    def bulk_index(self, tenant_id: str, recreate_index: bool = False, batch_size: int = 500) -> Dict[str, Any]:
+        """执行全量索引"""
+        import time
+        start_time = time.time()
+
+        try:
+            # 1. 加载mapping
+            logger.info(f"[BulkIndexing] Loading mapping for tenant_id={tenant_id}")
+            mapping = load_mapping()
+
+            # 2. 处理索引（删除并重建或创建）
+            if recreate_index:
+                logger.info(f"[BulkIndexing] Recreating index: {self.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}")
+                    else:
+                        raise Exception(f"Failed to delete index: {self.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.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}")
+            else:
+                logger.info(f"[BulkIndexing] Index already exists: {self.index_name}")
+
+            # 3. 转换数据
+            logger.info(f"[BulkIndexing] Transforming data for tenant_id={tenant_id}")
+            transformer = SPUTransformer(self.db_engine, tenant_id)
+            documents = transformer.transform_batch()
+
+            if not documents:
+                logger.warning(f"[BulkIndexing] No documents to index for tenant_id={tenant_id}")
+                return {
+                    "success": True,
+                    "total": 0,
+                    "indexed": 0,
+                    "failed": 0,
+                    "elapsed_time": time.time() - start_time,
+                    "message": "No documents to index"
+                }
+
+            logger.info(f"[BulkIndexing] Transformed {len(documents)} documents")
+
+            # 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)
+            results = indexer.index_documents(
+                documents,
+                id_field="spu_id",
+                show_progress=False  # API调用时不打印进度
+            )
+
+            elapsed_time = time.time() - start_time
+
+            logger.info(
+                f"[BulkIndexing] Completed for tenant_id={tenant_id}: "
+                f"indexed={results['success']}, failed={results['failed']}, "
+                f"elapsed={elapsed_time:.2f}s"
+            )
+
+            return {
+                "success": results['failed'] == 0,
+                "total": len(documents),
+                "indexed": results['success'],
+                "failed": results['failed'],
+                "elapsed_time": elapsed_time,
+                "index_name": self.index_name,
+                "tenant_id": tenant_id
+            }
+
+        except Exception as e:
+            logger.error(f"[BulkIndexing] Failed for tenant_id={tenant_id}: {e}", exc_info=True)
+            raise
+
-"""
-增量数据获取服务。
-
-提供单个SPU的数据获取接口，用于增量更新ES索引。
-公共数据（分类映射、配置等）在服务启动时预加载，以提高性能。
-"""
+"""增量数据获取服务"""
 import pandas as pd
-import numpy as np
 import logging
 from typing import Dict, Any, Optional
 from sqlalchemy import text
-from config import ConfigLoader
-from config.tenant_config_loader import get_tenant_config_loader
-from indexer.document_transformer import SPUDocumentTransformer
+from indexer.indexing_utils import load_category_mapping, create_document_transformer
 # Configure logger
 logger = logging.getLogger(__name__)
 class IncrementalIndexerService:
-    """增量索引服务，提供单个SPU数据获取功能。"""
+    """增量索引服务，提供SPU数据获取功能。"""
     def __init__(self, db_engine: Any):
-        """
-        初始化增量索引服务。
-
-        Args:
-            db_engine: SQLAlchemy database engine
-        """
+        """初始化增量索引服务"""
         self.db_engine = db_engine
         # 预加载分类映射（全局，所有租户共享）
-        self.category_id_to_name = self._load_category_mapping()
+        self.category_id_to_name = load_category_mapping(db_engine)
         logger.info(f"Preloaded {len(self.category_id_to_name)} category mappings")
-        
-        # 租户配置加载器（延迟加载，按需获取租户配置）
-        self.tenant_config_loader = get_tenant_config_loader()
-
-    def _load_category_mapping(self) -> Dict[str, str]:
-        """
-        加载分类ID到名称的映射（全局，所有租户共享）。
-        
-        Returns:
-            Dictionary mapping category_id to category_name
-        """
-        query = text("""
-            SELECT DISTINCT
-                category_id,
-                category
-            FROM shoplazza_product_spu
-            WHERE deleted = 0 AND category_id IS NOT NULL
-        """)
-        
-        mapping = {}
-        try:
-            with self.db_engine.connect() as conn:
-                result = conn.execute(query)
-                for row in result:
-                    category_id = str(int(row.category_id))
-                    category_name = row.category
-                    
-                    if not category_name or not category_name.strip():
-                        logger.warning(f"Category ID {category_id} has empty name, skipping")
-                        continue
-                    
-                    mapping[category_id] = category_name
-        except Exception as e:
-            logger.error(f"Failed to load category mapping: {e}", exc_info=True)
-        
-        return mapping
     def get_spu_document(self, tenant_id: str, spu_id: str) -> Optional[Dict[str, Any]]:
-        """
-        获取单个SPU的ES文档数据。
-
-        Args:
-            tenant_id: 租户ID
-            spu_id: SPU ID
-
-        Returns:
-            ES文档字典，如果SPU不存在或已删除则返回None
-        """
+        """获取SPU的ES文档数据"""
         try:
             # 加载SPU数据
             spu_row = self._load_single_spu(tenant_id, spu_id)
@@ -94,38 +36,10 @@ class IncrementalIndexerService:
             # 加载Option数据
             options_df = self._load_options_for_spu(tenant_id, spu_id)
-            # 获取租户配置
-            tenant_config = self.tenant_config_loader.get_tenant_config(tenant_id)
-            
-            # 加载搜索配置
-            translator = None
-            translation_prompts = {}
-            searchable_option_dimensions = ['option1', 'option2', 'option3']
-            try:
-                config_loader = ConfigLoader()
-                config = config_loader.load_config()
-                searchable_option_dimensions = config.spu_config.searchable_option_dimensions
-                
-                # Initialize translator if translation is enabled
-                if config.query_config.enable_translation:
-                    from query.translator import Translator
-                    translator = Translator(
-                        api_key=config.query_config.translation_api_key,
-                        use_cache=True,  # 索引时使用缓存避免重复翻译
-                        glossary_id=config.query_config.translation_glossary_id,
-                        translation_context=config.query_config.translation_context
-                    )
-                    translation_prompts = config.query_config.translation_prompts
-            except Exception as e:
-                logger.warning(f"Failed to load config, using default: {e}")
-            
             # 创建文档转换器
-            transformer = SPUDocumentTransformer(
+            transformer = create_document_transformer(
                 category_id_to_name=self.category_id_to_name,
-                searchable_option_dimensions=searchable_option_dimensions,
-                tenant_config=tenant_config,
-                translator=translator,
-                translation_prompts=translation_prompts
+                tenant_id=tenant_id
             )
             # 转换为ES文档
@@ -147,16 +61,7 @@ class IncrementalIndexerService:
             raise
     def _load_single_spu(self, tenant_id: str, spu_id: str) -> Optional[pd.Series]:
-        """
-        加载单个SPU数据。
-
-        Args:
-            tenant_id: 租户ID
-            spu_id: SPU ID
-
-        Returns:
-            SPU行数据，如果不存在则返回None
-        """
+        """加载单个SPU数据"""
         query = text("""
             SELECT 
                 id, shop_id, shoplazza_id, title, brief, description,
@@ -180,16 +85,7 @@ class IncrementalIndexerService:
         return df.iloc[0]
     def _load_skus_for_spu(self, tenant_id: str, spu_id: str) -> pd.DataFrame:
-        """
-        加载指定SPU的所有SKU数据。
-
-        Args:
-            tenant_id: 租户ID
-            spu_id: SPU ID
-
-        Returns:
-            SKU数据DataFrame
-        """
+        """加载指定SPU的所有SKU数据"""
         query = text("""
             SELECT 
                 id, spu_id, shop_id, shoplazza_id, shoplazza_product_id,
@@ -210,16 +106,7 @@ class IncrementalIndexerService:
         return df
     def _load_options_for_spu(self, tenant_id: str, spu_id: str) -> pd.DataFrame:
-        """
-        加载指定SPU的所有Option数据。
-
-        Args:
-            tenant_id: 租户ID
-            spu_id: SPU ID
-
-        Returns:
-            Option数据DataFrame
-        """
+        """加载指定SPU的所有Option数据"""
         query = text("""
             SELECT 
                 id, spu_id, shop_id, shoplazza_id, shoplazza_product_id,
@@ -0,0 +1,112 @@
+"""
+索引工具函数。
+
+提取公共逻辑，避免代码重复。
+"""
+
+import logging
+from typing import Dict, Any, Optional
+from sqlalchemy import Engine, text
+from config import ConfigLoader
+from config.tenant_config_loader import get_tenant_config_loader
+from indexer.document_transformer import SPUDocumentTransformer
+
+logger = logging.getLogger(__name__)
+
+
+def load_category_mapping(db_engine: Engine) -> Dict[str, str]:
+    """
+    加载分类ID到名称的映射（全局，所有租户共享）。
+    
+    Args:
+        db_engine: SQLAlchemy database engine
+        
+    Returns:
+        Dictionary mapping category_id to category_name
+    """
+    query = text("""
+        SELECT DISTINCT
+            category_id,
+            category
+        FROM shoplazza_product_spu
+        WHERE deleted = 0 AND category_id IS NOT NULL
+    """)
+    
+    mapping = {}
+    try:
+        with db_engine.connect() as conn:
+            result = conn.execute(query)
+            for row in result:
+                category_id = str(int(row.category_id))
+                category_name = row.category
+                
+                if not category_name or not category_name.strip():
+                    logger.warning(f"Category ID {category_id} has empty name, skipping")
+                    continue
+                
+                mapping[category_id] = category_name
+    except Exception as e:
+        logger.error(f"Failed to load category mapping: {e}", exc_info=True)
+    
+    return mapping
+
+
+def create_document_transformer(
+    category_id_to_name: Dict[str, str],
+    tenant_id: str,
+    searchable_option_dimensions: Optional[list] = None,
+    translator: Optional[Any] = None,
+    translation_prompts: Optional[Dict[str, str]] = None
+) -> SPUDocumentTransformer:
+    """
+    创建文档转换器（统一初始化逻辑）。
+    
+    Args:
+        category_id_to_name: 分类ID到名称的映射
+        tenant_id: 租户ID
+        searchable_option_dimensions: 可搜索的option维度列表（如果为None则从配置加载）
+        translator: 翻译器实例（如果为None则根据配置初始化）
+        translation_prompts: 翻译提示词配置（如果为None则从配置加载）
+        
+    Returns:
+        SPUDocumentTransformer实例
+    """
+    # 加载租户配置
+    tenant_config_loader = get_tenant_config_loader()
+    tenant_config = tenant_config_loader.get_tenant_config(tenant_id)
+    
+    # 加载搜索配置（如果需要）
+    if searchable_option_dimensions is None or translator is None or translation_prompts is None:
+        try:
+            config_loader = ConfigLoader()
+            config = config_loader.load_config()
+            
+            if searchable_option_dimensions is None:
+                searchable_option_dimensions = config.spu_config.searchable_option_dimensions
+            
+            if translator is None and config.query_config.enable_translation:
+                from query.translator import Translator
+                translator = Translator(
+                    api_key=config.query_config.translation_api_key,
+                    use_cache=True,
+                    glossary_id=config.query_config.translation_glossary_id,
+                    translation_context=config.query_config.translation_context
+                )
+            
+            if translation_prompts is None:
+                translation_prompts = config.query_config.translation_prompts
+        except Exception as e:
+            logger.warning(f"Failed to load config, using defaults: {e}")
+            if searchable_option_dimensions is None:
+                searchable_option_dimensions = ['option1', 'option2', 'option3']
+            if translation_prompts is None:
+                translation_prompts = {}
+    
+    return SPUDocumentTransformer(
+        category_id_to_name=category_id_to_name,
+        searchable_option_dimensions=searchable_option_dimensions,
+        tenant_config=tenant_config,
+        translator=translator,
+        translation_prompts=translation_prompts
+    )
+
@@ -5,14 +5,10 @@ Transforms SPU and SKU data from MySQL into SPU-level ES documents with nested s
 """
 import pandas as pd
-import numpy as np
 import logging
 from typing import Dict, Any, List, Optional
-from sqlalchemy import create_engine, text
-from utils.db_connector import create_db_connection
-from config import ConfigLoader
-from config.tenant_config_loader import get_tenant_config_loader
-from indexer.document_transformer import SPUDocumentTransformer
+from sqlalchemy import text
+from indexer.indexing_utils import load_category_mapping, create_document_transformer
 # Configure logger
 logger = logging.getLogger(__name__)
@@ -21,96 +17,19 @@ logger = logging.getLogger(__name__)
 class SPUTransformer:
     """Transform SPU and SKU data into SPU-level ES documents."""
-    def __init__(
-        self,
-        db_engine: Any,
-        tenant_id: str
-    ):
-        """
-        Initialize SPU transformer.
-
-        Args:
-            db_engine: SQLAlchemy database engine
-            tenant_id: Tenant ID for filtering data
-        """
+    def __init__(self, db_engine: Any, tenant_id: str):
         self.db_engine = db_engine
         self.tenant_id = tenant_id
-        # Load configuration to get searchable_option_dimensions
-        translator = None
-        translation_prompts = {}
-        try:
-            config_loader = ConfigLoader()
-            config = config_loader.load_config()
-            self.searchable_option_dimensions = config.spu_config.searchable_option_dimensions
-            
-            # Initialize translator if translation is enabled
-            if config.query_config.enable_translation:
-                from query.translator import Translator
-                translator = Translator(
-                    api_key=config.query_config.translation_api_key,
-                    use_cache=True,  # 索引时使用缓存避免重复翻译
-                    glossary_id=config.query_config.translation_glossary_id,
-                    translation_context=config.query_config.translation_context
-                )
-                translation_prompts = config.query_config.translation_prompts
-        except Exception as e:
-            logger.warning(f"Failed to load config, using default: {e}")
-            self.searchable_option_dimensions = ['option1', 'option2', 'option3']
-        
         # Load category ID to name mapping
-        self.category_id_to_name = self._load_category_mapping()
-        
-        # Load tenant config
-        tenant_config_loader = get_tenant_config_loader()
-        tenant_config = tenant_config_loader.get_tenant_config(tenant_id)
+        self.category_id_to_name = load_category_mapping(db_engine)
+        logger.info(f"Loaded {len(self.category_id_to_name)} category ID to name mappings")
         # Initialize document transformer
-        self.document_transformer = SPUDocumentTransformer(
+        self.document_transformer = create_document_transformer(
             category_id_to_name=self.category_id_to_name,
-            searchable_option_dimensions=self.searchable_option_dimensions,
-            tenant_config=tenant_config,
-            translator=translator,
-            translation_prompts=translation_prompts
+            tenant_id=tenant_id
         )
-
-    def _load_category_mapping(self) -> Dict[str, str]:
-        """
-        Load category ID to name mapping from database.
-        
-        Returns:
-            Dictionary mapping category_id to category_name
-        """
-        query = text("""
-            SELECT DISTINCT
-                category_id,
-                category
-            FROM shoplazza_product_spu
-            WHERE deleted = 0 AND category_id IS NOT NULL
-        """)
-        
-        mapping = {}
-        with self.db_engine.connect() as conn:
-            result = conn.execute(query)
-            for row in result:
-                category_id = str(int(row.category_id))
-                category_name = row.category
-                
-                if not category_name or not category_name.strip():
-                    logger.warning(f"Category ID {category_id} has empty name, skipping")
-                    continue
-                
-                mapping[category_id] = category_name
-        
-        logger.info(f"Loaded {len(mapping)} category ID to name mappings")
-        
-        # Log all category mappings for debugging
-        if mapping:
-            logger.debug("Category ID mappings:")
-            for cid, name in sorted(mapping.items()):
-                logger.debug(f"  {cid} -> {name}")
-        
-        return mapping
     def load_spu_data(self) -> pd.DataFrame:
         """