搜索服务（6002）不再初始化/挂载 /indexer/* 路由，避免索引阻塞线上搜索

新增 api/indexer_app.py，在独立进程（默认 6004）中初始化 ES + DB + 索引服务，并复用 api/routes/indexer.py 一套路由新增 api/service_registry.py，通过注册表向索引路由注入 ES 客户端和索引服务，消除重复代码与循环依赖 main.py 增加 serve-indexer 子命令；scripts/start.sh / stop.sh / start_backend.sh / start_indexer.sh 支持独立管理索引进程文档中所有索引相关示例由 6002/indexer/* 统一调整为 6004/indexer/*

搜索服务（6002）不再初始化/挂载 /indexer/* 路由，避免索引阻塞线上搜索
新增 api/indexer_app.py，在独立进程（默认 6004）中初始化 ES + DB + 索引服务，并复用 api/routes/indexer.py 一套路由新增 api/service_registry.py，通过注册表向索引路由注入 ES 客户端和索引服务，消除重复代码与循环依赖 main.py 增加 serve-indexer 子命令；scripts/start.sh / stop.sh / start_backend.sh / start_indexer.sh 支持独立管理索引进程文档中所有索引相关示例由 6002/indexer/* 统一调整为 6004/indexer/*
tangwang
1 parent 3ec5bfe6
Showing 14 changed files with 464 additions and 92 deletions Show diff stats
a.sh
api/app.py
api/indexer_app.py
api/routes/indexer.py
api/service_registry.py
docs/搜索API对接指南.md
docs/索引数据接口文档.md
main.py
scripts/amazon_xlsx_to_shoplazza_xlsx.py
scripts/start.sh
scripts/start_backend.sh
scripts/start_indexer.sh
scripts/stop.sh
scripts/tenant3__csv_to_shoplazza_xlsx.sh
@@ -0,0 +1 @@
+curl -X POST "http://localhost:6004/indexer/reindex"   -H "Content-Type: application/json"   -d '{"tenant_id":"162","batch_size":500}'
@@ -46,18 +46,15 @@ sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 from config.env_config import ES_CONFIG, DB_CONFIG
 from config import ConfigLoader
 from utils import ESClient
-from utils.db_connector import create_db_connection
 from search import Searcher
 from query import QueryParser
-from indexer.incremental_service import IncrementalIndexerService
+from .service_registry import set_es_client
 # Global instances
 _es_client: Optional[ESClient] = None
 _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"):
@@ -67,7 +64,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, _bulk_indexing_service
+    global _es_client, _searcher, _query_parser, _config
     start_time = time.time()
     logger.info("Initializing search service (multi-tenant)")
@@ -92,6 +89,8 @@ def init_service(es_host: str = &quot;http://localhost:9200&quot;):
     if not _es_client.ping():
         raise ConnectionError(f"Failed to connect to Elasticsearch at {es_host}")
     logger.info("Elasticsearch connected")
+    # expose ES client for any shared components (e.g. searcher)
+    set_es_client(_es_client)
     # Initialize components
     logger.info("Initializing query parser...")
@@ -100,44 +99,6 @@ def init_service(es_host: str = &quot;http://localhost:9200&quot;):
     logger.info("Initializing searcher...")
     _searcher = Searcher(_es_client, _config, _query_parser)
-    # Initialize indexing services (if DB config is available)
-    try:
-        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 database connection for indexing services...")
-            db_engine = create_db_connection(
-                host=db_host,
-                port=db_port,
-                database=db_database,
-                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 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 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}")
@@ -172,16 +133,6 @@ def get_config():
     return _config
-def get_incremental_service() -> Optional[IncrementalIndexerService]:
-    """Get incremental indexer service instance."""
-    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",
@@ -320,12 +271,11 @@ async def health_check(request: Request):
         )
-# Include routers
-from .routes import search, admin, indexer
+# Include routers (search app should NOT mount indexer routes)
+from .routes import search, admin
 app.include_router(search.router)
 app.include_router(admin.router)
-app.include_router(indexer.router)
 # Mount static files and serve frontend
 frontend_path = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "frontend")
@@ -0,0 +1,236 @@
+"""
+FastAPI application dedicated to indexing (separate from search API).
+
+This service mounts ONLY ONE copy of indexer routes: api/routes/indexer.py
+and injects required services via api/service_registry.py.
+
+Usage:
+    uvicorn api.indexer_app:app --host 0.0.0.0 --port 6004 --reload
+
+This service only exposes /indexer/* routes and can be run in a separate
+process so that heavy indexing work does not block online search traffic.
+"""
+
+import os
+import sys
+import logging
+import time
+from typing import Optional
+
+from fastapi import FastAPI, Request
+from fastapi.responses import JSONResponse
+
+# Configure logging
+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(log_dir / "indexer_api.log", mode="a", encoding="utf-8"),
+    ],
+)
+logger = logging.getLogger(__name__)
+
+# Add parent directory to path
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from config.env_config import ES_CONFIG  # noqa: E402
+from config import ConfigLoader  # noqa: E402
+from utils import ESClient  # noqa: E402
+from utils.db_connector import create_db_connection  # noqa: E402
+from indexer.incremental_service import IncrementalIndexerService  # noqa: E402
+from indexer.bulk_indexing_service import BulkIndexingService  # noqa: E402
+from .routes import indexer as indexer_routes  # noqa: E402
+from .service_registry import set_es_client, set_indexer_services  # noqa: E402
+
+
+_es_client: Optional[ESClient] = None
+_config = None
+_incremental_service: Optional[IncrementalIndexerService] = None
+_bulk_indexing_service: Optional[BulkIndexingService] = None
+
+
+def init_indexer_service(es_host: str = "http://localhost:9200"):
+    """
+    Initialize indexing services (ES client + DB + indexers).
+
+    This mirrors the indexing-related initialization logic in api.app.init_service
+    but without search-related components.
+    """
+    global _es_client, _config, _incremental_service, _bulk_indexing_service
+
+    start_time = time.time()
+    logger.info("Initializing Indexer service")
+
+    # Load configuration (kept for parity/logging; indexer routes don't depend on it)
+    logger.info("Loading configuration...")
+    config_loader = ConfigLoader("config/config.yaml")
+    _config = config_loader.load_config()
+    logger.info("Configuration loaded")
+
+    # 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")
+
+    # Connect to Elasticsearch
+    logger.info(f"Connecting to Elasticsearch at {es_host} for indexer...")
+    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():
+        raise ConnectionError(f"Failed to connect to Elasticsearch at {es_host}")
+    logger.info("Elasticsearch connected for indexer")
+    # publish ES client for routes
+    set_es_client(_es_client)
+
+    # Initialize indexing services (DB is required here)
+    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 database connection for indexing services...")
+        db_engine = create_db_connection(
+            host=db_host,
+            port=db_port,
+            database=db_database,
+            username=db_username,
+            password=db_password,
+        )
+
+        _incremental_service = IncrementalIndexerService(db_engine)
+        _bulk_indexing_service = BulkIndexingService(db_engine, _es_client)
+        set_indexer_services(
+            incremental_service=_incremental_service,
+            bulk_indexing_service=_bulk_indexing_service,
+        )
+        logger.info("Indexer services initialized (incremental + bulk)")
+    else:
+        missing = [
+            name
+            for name, value in [
+                ("DB_HOST", db_host),
+                ("DB_DATABASE", db_database),
+                ("DB_USERNAME", db_username),
+                ("DB_PASSWORD", db_password),
+            ]
+            if not value
+        ]
+        logger.warning(
+            "Database config incomplete for indexer, services will not be available. "
+            f"Missing: {', '.join(missing)}"
+        )
+        _incremental_service = None
+        _bulk_indexing_service = None
+
+    elapsed = time.time() - start_time
+    logger.info(f"Indexer service ready! (took {elapsed:.2f}s)")
+
+    # NOTE: we intentionally do NOT synchronize anything into api.app
+    # to avoid code/route duplication and accidental availability on port 6002.
+
+
+app = FastAPI(
+    title="E-Commerce Indexer API",
+    description="Dedicated indexing service for SearchEngine",
+    version="1.0.0",
+    docs_url="/docs",
+    redoc_url="/redoc",
+    openapi_url="/openapi.json",
+)
+
+
+@app.on_event("startup")
+async def startup_event():
+    es_host = os.getenv("ES_HOST", "http://localhost:9200")
+    logger.info("Starting Indexer API service")
+    logger.info(f"Elasticsearch Host: {es_host}")
+    try:
+        init_indexer_service(es_host=es_host)
+        logger.info("Indexer service initialized successfully")
+    except Exception as e:
+        logger.error(f"Failed to initialize indexer service: {e}", exc_info=True)
+        logger.warning("Indexer service will start but may not function correctly")
+
+
+@app.on_event("shutdown")
+async def shutdown_event():
+    logger.info("Shutting down Indexer API service")
+
+
+@app.exception_handler(Exception)
+async def global_exception_handler(request: Request, exc: Exception):
+    """Global exception handler with basic logging."""
+    client_ip = request.client.host if request.client else "unknown"
+    logger.error(f"Unhandled exception from {client_ip}: {exc}", exc_info=True)
+
+    return JSONResponse(
+        status_code=500,
+        content={
+            "error": "Internal server error",
+            "detail": "An unexpected error occurred in indexer service.",
+            "timestamp": int(time.time()),
+        },
+    )
+
+
+@app.get("/health")
+async def health_check():
+    """Simple health check for indexer service."""
+    try:
+        # ensure ES is reachable (best-effort)
+        if _es_client is None:
+            raise RuntimeError("ES client is not initialized")
+        return {
+            "status": "healthy",
+            "services": {
+                "elasticsearch": "connected",
+                "incremental_indexer": "initialized" if _incremental_service else "unavailable",
+                "bulk_indexer": "initialized" if _bulk_indexing_service else "unavailable",
+            },
+            "timestamp": int(time.time()),
+        }
+    except Exception as e:
+        logger.error(f"Indexer health check failed: {e}")
+        return JSONResponse(
+            status_code=503,
+            content={
+                "status": "unhealthy",
+                "error": str(e),
+                "timestamp": int(time.time()),
+            },
+        )
+
+
+# Mount the single source of truth indexer routes
+app.include_router(indexer_routes.router)
+
+
+if __name__ == "__main__":
+    import argparse
+    import uvicorn
+
+    parser = argparse.ArgumentParser(description="Start Indexer API service")
+    parser.add_argument("--host", default="0.0.0.0", help="Host to bind to")
+    parser.add_argument("--port", type=int, default=6004, help="Port to bind to")
+    parser.add_argument("--es-host", default="http://localhost:9200", help="Elasticsearch host")
+    parser.add_argument("--reload", action="store_true", help="Enable auto-reload")
+    args = parser.parse_args()
+
+    os.environ["ES_HOST"] = args.es_host
+
+    uvicorn.run(
+        "api.indexer_app:app",
+        host=args.host,
+        port=args.port,
+        reload=args.reload,
+    )
+
@@ -8,6 +8,10 @@ from fastapi import APIRouter, HTTPException
 from typing import List
 from pydantic import BaseModel
 import logging
+from sqlalchemy import text
+
+# Indexer routes depend on services provided by api/indexer_app.py via this registry.
+from ..service_registry import get_incremental_service, get_bulk_indexing_service, get_es_client
 logger = logging.getLogger(__name__)
@@ -42,7 +46,6 @@ async def reindex_all(request: ReindexRequest):
     将指定租户的所有SPU数据重新索引到ES。支持删除旧索引并重建。
     """
     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")
@@ -81,8 +84,6 @@ async def index_spus(request: IndexSpusRequest):
     - 最后给出总体统计：total, success_count, failed_count等
     """
     try:
-        from ..app import get_incremental_service, get_es_client
-        
         # 验证请求参数
         if not request.spu_ids and not request.delete_spu_ids:
             raise HTTPException(status_code=400, detail="spu_ids and delete_spu_ids cannot both be empty")
@@ -126,7 +127,6 @@ async def get_documents(request: GetDocumentsRequest):
     根据SPU ID列表获取ES文档数据（不写入ES）。用于查看、调试或验证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:
@@ -165,8 +165,6 @@ async def get_documents(request: GetDocumentsRequest):
 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", "database": "unknown", "preloaded_data": {"category_mappings": 0}}
@@ -0,0 +1,43 @@
+"""
+Minimal service registry shared by multiple FastAPI apps.
+
+We keep only ONE copy of indexer routes (api/routes/indexer.py) and inject
+services via this registry, so the same routes can run in:
+- Search API (6002): does NOT register indexer services, and does NOT mount indexer routes.
+- Indexer API (6004): registers indexer services and mounts indexer routes.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Optional
+
+
+_es_client: Optional[Any] = None
+_incremental_service: Optional[Any] = None
+_bulk_indexing_service: Optional[Any] = None
+
+
+def set_es_client(es_client: Any) -> None:
+    global _es_client
+    _es_client = es_client
+
+
+def get_es_client() -> Any:
+    if _es_client is None:
+        raise RuntimeError("ES client is not initialized")
+    return _es_client
+
+
+def set_indexer_services(*, incremental_service: Any, bulk_indexing_service: Any) -> None:
+    global _incremental_service, _bulk_indexing_service
+    _incremental_service = incremental_service
+    _bulk_indexing_service = bulk_indexing_service
+
+
+def get_incremental_service() -> Optional[Any]:
+    return _incremental_service
+
+
+def get_bulk_indexing_service() -> Optional[Any]:
+    return _bulk_indexing_service
+
@@ -839,7 +839,7 @@ curl &quot;http://localhost:6002/search/12345&quot;
 **首次索引（重建索引）**:
 ```bash
-curl -X POST "http://localhost:6002/indexer/reindex" \
+curl -X POST "http://localhost:6004/indexer/reindex" \
   -H "Content-Type: application/json" \
   -d '{
     "tenant_id": "162",
@@ -859,7 +859,7 @@ tail -f logs/*.log
 **增量更新（不重建索引）**:
 ```bash
-curl -X POST "http://localhost:6002/indexer/reindex" \
+curl -X POST "http://localhost:6004/indexer/reindex" \
   -H "Content-Type: application/json" \
   -d '{
     "tenant_id": "162",
@@ -1015,7 +1015,7 @@ cat logs/indexer.log | jq &#39;select(.operation == &quot;request_complete&quot;) | {timestamp
 **示例1：普通增量索引（自动检测删除）**:
 ```bash
-curl -X POST "http://localhost:6002/indexer/index" \
+curl -X POST "http://localhost:6004/indexer/index" \
   -H "Content-Type: application/json" \
   -d '{
     "tenant_id": "162",
@@ -1026,7 +1026,7 @@ curl -X POST &quot;http://localhost:6002/indexer/index&quot; \
 **示例2：显式删除（批量删除）**:
 ```bash
-curl -X POST "http://localhost:6002/indexer/index" \
+curl -X POST "http://localhost:6004/indexer/index" \
   -H "Content-Type: application/json" \
   -d '{
     "tenant_id": "162",
@@ -1038,7 +1038,7 @@ curl -X POST &quot;http://localhost:6002/indexer/index&quot; \
 **示例3：仅删除（不索引）**:
 ```bash
-curl -X POST "http://localhost:6002/indexer/index" \
+curl -X POST "http://localhost:6004/indexer/index" \
   -H "Content-Type: application/json" \
   -d '{
     "tenant_id": "162",
@@ -1050,7 +1050,7 @@ curl -X POST &quot;http://localhost:6002/indexer/index&quot; \
 **示例4：混合操作（索引+删除）**:
 ```bash
-curl -X POST "http://localhost:6002/indexer/index" \
+curl -X POST "http://localhost:6004/indexer/index" \
   -H "Content-Type: application/json" \
   -d '{
     "tenant_id": "162",
@@ -1133,7 +1133,7 @@ curl -X POST &quot;http://localhost:6002/indexer/index&quot; \
 **单个SPU查询**:
 ```bash
-curl -X POST "http://localhost:6002/indexer/documents" \
+curl -X POST "http://localhost:6004/indexer/documents" \
   -H "Content-Type: application/json" \
   -d '{
     "tenant_id": "162",
@@ -1143,7 +1143,7 @@ curl -X POST &quot;http://localhost:6002/indexer/documents&quot; \
 **批量SPU查询**:
 ```bash
-curl -X POST "http://localhost:6002/indexer/documents" \
+curl -X POST "http://localhost:6004/indexer/documents" \
   -H "Content-Type: application/json" \
   -d '{
     "tenant_id": "162",
@@ -1182,7 +1182,7 @@ curl -X POST &quot;http://localhost:6002/indexer/documents&quot; \
 #### 请求示例
 ```bash
-curl -X GET "http://localhost:6002/indexer/health"
+curl -X GET "http://localhost:6004/indexer/health"
 ```
 ---
@@ -250,12 +250,12 @@ GET /indexer/spu/{spu_id}?tenant_id={tenant_id}
 ```bash
 # cURL
-curl -X GET "http://localhost:6002/indexer/spu/123?tenant_id=1"
+curl -X GET "http://localhost:6004/indexer/spu/123?tenant_id=1"
 # Java (OkHttp)
 OkHttpClient client = new OkHttpClient();
 Request request = new Request.Builder()
-    .url("http://localhost:6002/indexer/spu/123?tenant_id=1")
+    .url("http://localhost:6004/indexer/spu/123?tenant_id=1")
     .get()
     .build();
 Response response = client.newCall(request).execute();
@@ -412,7 +412,7 @@ public void onProductChange(ProductChangeEvent event) {
     String spuId = event.getSpuId();
     // 调用增量接口获取ES文档数据
-    String url = String.format("http://localhost:6002/indexer/spu/%s?tenant_id=%s", spuId, tenantId);
+    String url = String.format("http://localhost:6004/indexer/spu/%s?tenant_id=%s", spuId, tenantId);
     Map<String, Object> esDoc = httpClient.get(url);
     // 推送到ES
@@ -428,7 +428,7 @@ public void onProductChange(ProductChangeEvent event) {
 // 伪代码示例
 List<String> changedSpuIds = getChangedSpuIds();
 for (String spuId : changedSpuIds) {
-    String url = String.format("http://localhost:6002/indexer/spu/%s?tenant_id=%s", spuId, tenantId);
+    String url = String.format("http://localhost:6004/indexer/spu/%s?tenant_id=%s", spuId, tenantId);
     Map<String, Object> esDoc = httpClient.get(url);
     elasticsearchClient.index("search_products", esDoc);
 }
@@ -20,13 +20,20 @@ sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
 from config import ConfigLoader
 from utils import ESClient
-from indexer import DataTransformer, IndexingPipeline
-from embeddings import BgeEncoder, CLIPImageEncoder
 from search import Searcher
 def cmd_ingest(args):
     """Run data ingestion."""
+    # Local imports to avoid hard dependency at module import time
+    import pandas as pd
+    from embeddings import BgeEncoder, CLIPImageEncoder
+    from indexer.bulk_indexer import IndexingPipeline
+    # NOTE: DataTransformer was referenced historically, but the concrete
+    # implementation is now provided via customer-specific scripts
+    # (e.g. data/customer1/ingest_customer1.py). If you still need a generic
+    # ingestion pipeline here, you can wire your own transformer.
+    from indexer.spu_transformer import SPUTransformer as DataTransformer
     print("Starting data ingestion")
     # Load config
@@ -68,7 +75,7 @@ def cmd_serve(args):
     os.environ['ES_HOST'] = args.es_host
     print("Starting API service (multi-tenant)...")
-    print(f"  Host: {args.host}:{args.port}")
+    print(f"  Host: {args.host}:{args.port} (search + indexer routes)")
     print(f"  Elasticsearch: {args.es_host}")
     uvicorn.run(
@@ -79,6 +86,21 @@ def cmd_serve(args):
     )
+def cmd_serve_indexer(args):
+    """Start dedicated Indexer API service (no search endpoints)."""
+    os.environ['ES_HOST'] = args.es_host
+
+    print("Starting Indexer API service...")
+    print(f"  Host: {args.host}:{args.port} (indexer only)")
+    print(f"  Elasticsearch: {args.es_host}")
+
+    uvicorn.run(
+        "api.indexer_app:app",
+        host=args.host,
+        port=args.port,
+        reload=args.reload
+    )
+
 def cmd_search(args):
     """Test search from command line."""
     # Load config
@@ -148,6 +170,16 @@ def main():
     serve_parser.add_argument('--es-host', default='http://localhost:9200', help='Elasticsearch host')
     serve_parser.add_argument('--reload', action='store_true', help='Enable auto-reload')
+    # Serve-indexer command
+    serve_indexer_parser = subparsers.add_parser(
+        'serve-indexer',
+        help='Start dedicated Indexer API service (indexer routes only)'
+    )
+    serve_indexer_parser.add_argument('--host', default='0.0.0.0', help='Host to bind to')
+    serve_indexer_parser.add_argument('--port', type=int, default=6004, help='Port to bind to')
+    serve_indexer_parser.add_argument('--es-host', default='http://localhost:9200', help='Elasticsearch host')
+    serve_indexer_parser.add_argument('--reload', action='store_true', help='Enable auto-reload')
+
     # Search command
     search_parser = subparsers.add_parser('search', help='Test search from command line')
     search_parser.add_argument('query', help='Search query')
@@ -169,6 +201,8 @@ def main():
         return cmd_ingest(args)
     elif args.command == 'serve':
         return cmd_serve(args)
+    elif args.command == 'serve-indexer':
+        return cmd_serve_indexer(args)
     elif args.command == 'search':
         return cmd_search(args)
     else:
@@ -421,8 +421,8 @@ def main():
     parser.add_argument("--max-rows-per-output", type=int, default=40000, help="Max total Excel rows per output file (including模板头部行，默认40000)")
     parser.add_argument("--max-products", type=int, default=None, help="Limit number of SPU groups to output (for testing)")
     # 默认行为：丢弃不符合要求的数据
-    parser.add_argument("--keep-spu-if-parent-missing", action="store_false", dest="skip_spu_if_parent_missing", default=True, help="Keep SPU even if parent ASIN not found in variants (default: skip entire SPU)")
-    parser.add_argument("--fix-sku-if-title-mismatch", action="store_false", dest="skip_sku_if_title_mismatch", default=True, help="Fix SKU title to match parent instead of skipping (default: skip SKU with mismatched title)")
+    parser.add_argument("--keep-spu-if-parent-missing", action="store_false", dest="skip_spu_if_parent_missing", default=False, help="Keep SPU even if parent ASIN not found in variants (default: skip entire SPU)")
+    parser.add_argument("--fix-sku-if-title-mismatch", action="store_false", dest="skip_sku_if_title_mismatch", default=False, help="Fix SKU title to match parent instead of skipping (default: skip SKU with mismatched title)")
     args = parser.parse_args()
     if not os.path.isdir(args.input_dir):
@@ -17,18 +17,18 @@ echo -e &quot;${GREEN}========================================${NC}&quot;
 # Create logs directory if it doesn't exist
 mkdir -p logs
-# Step 1: Start backend in background
-echo -e "\n${YELLOW}Step 1/2: 启动后端服务${NC}"
-echo -e "${YELLOW}后端服务将在后台运行...${NC}"
+# Step 1: Start backend in background (search API)
+echo -e "\n${YELLOW}Step 1/3: 启动后端搜索服务${NC}"
+echo -e "${YELLOW}后端搜索服务将在后台运行...${NC}"
 nohup ./scripts/start_backend.sh > logs/backend.log 2>&1 &
 BACKEND_PID=$!
 echo $BACKEND_PID > logs/backend.pid
-echo -e "${GREEN}后端服务已启动 (PID: $BACKEND_PID)${NC}"
+echo -e "${GREEN}后端搜索服务已启动 (PID: $BACKEND_PID)${NC}"
 echo -e "${GREEN}日志文件: logs/backend.log${NC}"
 # Wait for backend to start
-echo -e "${YELLOW}等待后端服务启动...${NC}"
+echo -e "${YELLOW}等待后端搜索服务启动...${NC}"
 MAX_RETRIES=30
 RETRY_COUNT=0
 BACKEND_READY=false
@@ -45,15 +45,47 @@ done
 # Check if backend is running
 if [ "$BACKEND_READY" = true ]; then
-    echo -e "${GREEN}✓ 后端服务运行正常${NC}"
+    echo -e "${GREEN}✓ 后端搜索服务运行正常${NC}"
 else
-    echo -e "${RED}✗ 后端服务启动失败，请检查日志: logs/backend.log${NC}"
+    echo -e "${RED}✗ 后端搜索服务启动失败，请检查日志: logs/backend.log${NC}"
     echo -e "${YELLOW}提示: 后端服务可能需要更多时间启动，或者检查端口是否被占用${NC}"
     exit 1
 fi
-# Step 2: Start frontend in background
-echo -e "\n${YELLOW}Step 2/2: 启动前端服务${NC}"
+# Step 2: Start indexer in background
+echo -e "\n${YELLOW}Step 2/3: 启动索引服务${NC}"
+echo -e "${YELLOW}索引服务将在后台运行...${NC}"
+
+nohup ./scripts/start_indexer.sh > logs/indexer.log 2>&1 &
+INDEXER_PID=$!
+echo $INDEXER_PID > logs/indexer.pid
+echo -e "${GREEN}索引服务已启动 (PID: $INDEXER_PID)${NC}"
+echo -e "${GREEN}日志文件: logs/indexer.log${NC}"
+
+# Wait for indexer to start
+echo -e "${YELLOW}等待索引服务启动...${NC}"
+MAX_RETRIES=30
+RETRY_COUNT=0
+INDEXER_READY=false
+
+while [ $RETRY_COUNT -lt $MAX_RETRIES ]; do
+    sleep 2
+    if curl -s http://localhost:6004/health > /dev/null 2>&1; then
+        INDEXER_READY=true
+        break
+    fi
+    RETRY_COUNT=$((RETRY_COUNT + 1))
+    echo -e "${YELLOW}  等待中... ($RETRY_COUNT/$MAX_RETRIES)${NC}"
+done
+
+if [ "$INDEXER_READY" = true ]; then
+    echo -e "${GREEN}✓ 索引服务运行正常${NC}"
+else
+    echo -e "${YELLOW}⚠ 索引服务可能还在启动中，请稍后访问 (日志: logs/indexer.log)${NC}"
+fi
+
+# Step 3: Start frontend in background
+echo -e "\n${YELLOW}Step 3/3: 启动前端服务${NC}"
 echo -e "${YELLOW}前端服务将在后台运行...${NC}"
 nohup ./scripts/start_frontend.sh > logs/frontend.log 2>&1 &
@@ -29,7 +29,7 @@ echo &quot;  API Port: ${API_PORT:-6002}&quot;
 echo "  ES Host: ${ES_HOST:-http://localhost:9200}"
 echo "  ES Username: ${ES_USERNAME:-not set}"
-echo -e "\n${YELLOW}Starting service (multi-tenant)...${NC}"
+echo -e "\n${YELLOW}Starting backend API service (search + admin)...${NC}"
 # Export environment variables for the Python process
 export API_HOST=${API_HOST:-0.0.0.0}
@@ -38,8 +38,9 @@ export ES_HOST=${ES_HOST:-http://localhost:9200}
 export ES_USERNAME=${ES_USERNAME:-}
 export ES_PASSWORD=${ES_PASSWORD:-}
-python -m api.app \
+python main.py serve \
   --host $API_HOST \
   --port $API_PORT \
   --es-host $ES_HOST
+
@@ -0,0 +1,46 @@
+#!/bin/bash
+
+# Start dedicated Indexer API Service
+
+set -e
+
+cd "$(dirname "$0")/.."
+source /home/tw/miniconda3/etc/profile.d/conda.sh
+conda activate searchengine
+
+GREEN='\033[0;32m'
+YELLOW='\033[1;33m'
+NC='\033[0m'
+
+echo -e "${GREEN}========================================${NC}"
+echo -e "${GREEN}Starting Indexer API Service${NC}"
+echo -e "${GREEN}========================================${NC}"
+
+# Load config from .env file if it exists
+if [ -f .env ]; then
+    set -a
+    source .env
+    set +a
+fi
+
+echo -e "\n${YELLOW}Configuration:${NC}"
+echo "  INDEXER Host: ${INDEXER_HOST:-0.0.0.0}"
+echo "  INDEXER Port: ${INDEXER_PORT:-6004}"
+echo "  ES Host: ${ES_HOST:-http://localhost:9200}"
+echo "  ES Username: ${ES_USERNAME:-not set}"
+
+echo -e "\n${YELLOW}Starting indexer service...${NC}"
+
+# Export environment variables for the Python process
+export INDEXER_HOST=${INDEXER_HOST:-0.0.0.0}
+export INDEXER_PORT=${INDEXER_PORT:-6004}
+export ES_HOST=${ES_HOST:-http://localhost:9200}
+export ES_USERNAME=${ES_USERNAME:-}
+export ES_PASSWORD=${ES_PASSWORD:-}
+
+python main.py serve-indexer \
+  --host $INDEXER_HOST \
+  --port $INDEXER_PORT \
+  --es-host $ES_HOST
+
+
@@ -7,7 +7,7 @@ echo &quot;========================================&quot;
 echo "Stopping Search Engine Services"
 echo "========================================"
-# Kill processes on port 6002 (backend)
+# Kill processes on port 6002 (backend - search API)
 BACKEND_PIDS=$(lsof -ti:6002 2>/dev/null)
 if [ ! -z "$BACKEND_PIDS" ]; then
     echo "Stopping backend server(s) on port 6002..."
@@ -29,6 +29,28 @@ else
     echo "No backend server found running on port 6002."
 fi
+# Kill processes on port 6004 (indexer API)
+INDEXER_PIDS=$(lsof -ti:6004 2>/dev/null)
+if [ ! -z "$INDEXER_PIDS" ]; then
+    echo "Stopping indexer server(s) on port 6004..."
+    for PID in $INDEXER_PIDS; do
+        echo "  Killing PID: $PID"
+        kill -TERM $PID 2>/dev/null || true
+    done
+    sleep 2
+    # Force kill if still running
+    REMAINING_PIDS=$(lsof -ti:6004 2>/dev/null)
+    if [ ! -z "$REMAINING_PIDS" ]; then
+        echo "  Force killing remaining processes..."
+        for PID in $REMAINING_PIDS; do
+            kill -KILL $PID 2>/dev/null || true
+        done
+    fi
+    echo "Indexer server stopped."
+else
+    echo "No indexer server found running on port 6004."
+fi
+
 # Kill processes on port 6003 (frontend)
 FRONTEND_PIDS=$(lsof -ti:6003 2>/dev/null)
 if [ ! -z "$FRONTEND_PIDS" ]; then
@@ -63,6 +85,15 @@ if [ -f &quot;logs/backend.pid&quot; ]; then
     rm -f logs/backend.pid
 fi
+[ -f "logs/indexer.pid" ] && INDEXER_PID=$(cat logs/indexer.pid 2>/dev/null)
+if [ ! -z "$INDEXER_PID" ] && kill -0 $INDEXER_PID 2>/dev/null; then
+    echo "Stopping indexer server via PID file (PID: $INDEXER_PID)..."
+    kill -TERM $INDEXER_PID 2>/dev/null || true
+    sleep 2
+    kill -KILL $INDEXER_PID 2>/dev/null || true
+fi
+rm -f logs/indexer.pid
+
 if [ -f "logs/frontend.pid" ]; then
     FRONTEND_PID=$(cat logs/frontend.pid 2>/dev/null)
     if [ ! -z "$FRONTEND_PID" ] && kill -0 $FRONTEND_PID 2>/dev/null; then