# MySQL 到 Elasticsearch 文档映射说明

## 1. 概述

本文档详细说明从 MySQL 数据库到 Elasticsearch 的完整数据转换流程，包括 SQL 查询逻辑、字段映射规则、以及复杂字段（类别、多款式）的处理方法。

### 1.1 数据流转架构

```
MySQL 数据库
    ↓
[SQL 查询] → SPU表、SKU表、Option表
    ↓
[数据加载] → Pandas DataFrame
    ↓
[文档转换] → ES 文档字典
    ↓
[批量写入] → Elasticsearch
```

### 1.2 核心组件

- **IncrementalIndexerService**: 增量索引服务，负责数据获取和索引
- **SPUDocumentTransformer**: 文档转换器，负责将数据库记录转换为 ES 文档
- **BulkIndexer**: 批量写入器，负责将文档批量写入 ES

---

## 2. 数据源表结构

### 2.1 SPU 表（shoplazza_product_spu）

SPU（Standard Product Unit）表存储商品的基本信息。

**主要字段：**

| 字段名 | 类型 | 说明 |
|--------|------|------|
| `id` | BIGINT | SPU主键ID |
| `tenant_id` | BIGINT | 租户ID |
| `title` | VARCHAR(500) | 商品标题 |
| `brief` | VARCHAR(1000) | 商品简介 |
| `description` | TEXT | 商品详细描述 |
| `vendor` | VARCHAR(255) | 供应商/品牌 |
| `category` | VARCHAR(255) | 类目名称 |
| `category_id` | BIGINT | 类目ID |
| `category_path` | VARCHAR(500) | 类目路径（逗号分隔的ID列表） |
| `category_level` | INT | 类目层级 |
| `tags` | TEXT | 标签（逗号分隔） |
| `image_src` | VARCHAR(500) | 主图URL |
| `fake_sales` | INT | 假销量 |
| `create_time` | DATETIME | 创建时间 |
| `update_time` | DATETIME | 更新时间 |
| `deleted` | BIT(1) | 删除标记 |

### 2.2 SKU 表（shoplazza_product_sku）

SKU（Stock Keeping Unit）表存储商品的变体信息，一个 SPU 可以有多个 SKU。

**主要字段：**

| 字段名 | 类型 | 说明 |
|--------|------|------|
| `id` | BIGINT | SKU主键ID |
| `spu_id` | BIGINT | 关联的SPU ID |
| `tenant_id` | BIGINT | 租户ID |
| `sku` | VARCHAR(100) | SKU编码 |
| `price` | DECIMAL(10,2) | 价格 |
| `compare_at_price` | DECIMAL(10,2) | 对比价格（原价） |
| `inventory_quantity` | INT | 库存数量 |
| `weight` | DECIMAL(10,2) | 重量 |
| `weight_unit` | VARCHAR(10) | 重量单位 |
| `option1` | VARCHAR(255) | 选项1的值（如：颜色-红色） |
| `option2` | VARCHAR(255) | 选项2的值（如：尺寸-L） |
| `option3` | VARCHAR(255) | 选项3的值（如：材质-棉） |
| `image_src` | VARCHAR(500) | SKU图片URL |
| `deleted` | BIT(1) | 删除标记 |

### 2.3 Option 表（shoplazza_product_option）

Option 表存储选项的名称定义，用于定义 SKU 的选项维度（如：颜色、尺寸、材质）。

**主要字段：**

| 字段名 | 类型 | 说明 |
|--------|------|------|
| `id` | BIGINT | Option主键ID |
| `spu_id` | BIGINT | 关联的SPU ID |
| `tenant_id` | BIGINT | 租户ID |
| `position` | INT | 位置序号（1、2、3对应option1、option2、option3） |
| `name` | VARCHAR(255) | 选项名称（如："颜色"、"尺寸"） |
| `values` | TEXT | 选项值列表（JSON格式） |
| `deleted` | BIT(1) | 删除标记 |

---

## 3. SQL 查询逻辑

### 3.1 批量查询 SPU 数据

**方法：** `_load_spus_for_spu_ids(tenant_id, spu_ids, include_deleted=True)`

**SQL 查询：**

```sql
SELECT 
    id, shop_id, shoplazza_id, title, brief, description,
    spu, vendor, vendor_url,
    image_src, image_width, image_height, image_path, image_alt,
    tags, note, category, category_id, category_google_id,
    category_level, category_path,
    fake_sales, display_fake_sales,
    tenant_id, creator, create_time, updater, update_time, deleted
FROM shoplazza_product_spu
WHERE tenant_id = :tenant_id 
  AND id IN :spu_ids
  [AND deleted = 0]  -- 如果 include_deleted=False
```

**查询特点：**
- 使用 `IN` 子句批量查询多个 SPU
- 支持查询已删除的记录（用于检测删除状态）
- 使用 SQLAlchemy 的 `bindparam` 和 `expanding=True` 处理动态 IN 列表

**代码位置：** `indexer/incremental_service.py:223-262`

### 3.2 批量查询 SKU 数据

**方法：** `_load_skus_for_spu_ids(tenant_id, spu_ids)`

**SQL 查询：**

```sql
SELECT 
    id, spu_id, shop_id, shoplazza_id, shoplazza_product_id,
    shoplazza_image_id, title, sku, barcode, position,
    price, compare_at_price, cost_price,
    option1, option2, option3,
    inventory_quantity, weight, weight_unit, image_src,
    wholesale_price, note, extend,
    shoplazza_created_at, shoplazza_updated_at, tenant_id,
    creator, create_time, updater, update_time, deleted
FROM shoplazza_product_sku
WHERE tenant_id = :tenant_id 
  AND deleted = 0 
  AND spu_id IN :spu_ids
```

**查询特点：**
- 批量查询多个 SPU 的所有 SKU
- 只查询未删除的 SKU（`deleted = 0`）
- 结果按 `spu_id` 分组处理

**代码位置：** `indexer/incremental_service.py:264-288`

### 3.3 批量查询 Option 数据

**方法：** `_load_options_for_spu_ids(tenant_id, spu_ids)`

**SQL 查询：**

```sql
SELECT 
    id, spu_id, shop_id, shoplazza_id, shoplazza_product_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 
  AND spu_id IN :spu_ids
ORDER BY spu_id, position
```

**查询特点：**
- 批量查询多个 SPU 的所有 Option
- 按 `spu_id` 和 `position` 排序，确保选项顺序一致
- `position` 字段对应 SKU 表中的 `option1`、`option2`、`option3`

**代码位置：** `indexer/incremental_service.py:290-310`

### 3.4 分类映射加载

**方法：** `load_category_mapping(db_engine)`

**SQL 查询：**

```sql
SELECT DISTINCT
    category_id,
    category
FROM shoplazza_product_spu
WHERE deleted = 0 AND category_id IS NOT NULL
```

**查询特点：**
- 全局加载所有租户的分类映射（category_id → category_name）
- 在服务初始化时预加载，避免重复查询
- 用于将 `category_path` 中的 ID 转换为名称

**代码位置：** `indexer/indexing_utils.py:17-51`

---

## 4. 字段映射详细说明

### 4.1 基础字段映射

#### 4.1.1 标识字段

| ES 字段 | 数据来源 | 转换逻辑 |
|---------|----------|----------|
| `tenant_id` | `spu_row['tenant_id']` | 直接转换为字符串：`str(tenant_id)` |
| `spu_id` | `spu_row['id']` | 直接转换为字符串：`str(spu_id)` |

#### 4.1.2 文本字段（多语言处理）

文本字段支持中英文双语，根据租户配置进行自动翻译。

**字段列表：**
- `title_zh` / `title_en`
- `brief_zh` / `brief_en`
- `description_zh` / `description_en`
- `vendor_zh` / `vendor_en`

**填充逻辑：**

1. **获取租户主语言**：从 `tenant_config` 读取 `primary_language`（默认 'zh'）

2. **翻译处理**：
   - 如果配置了 `translate_to_en=true` 且主语言不是 'en'，则翻译到英文
   - 如果配置了 `translate_to_zh=true` 且主语言不是 'zh'，则翻译到中文
   - 使用 `Translator.translate_for_indexing()` 方法进行翻译

3. **字段填充**：
   ```python
   # 示例：title 字段
   if pd.notna(spu_row.get('title')):
       title_text = str(spu_row['title'])
       if self.translator:
           translations = self.translator.translate_for_indexing(
               title_text,
               shop_language=primary_lang,
               source_lang=primary_lang,
               translate_to_en=translate_to_en,
               translate_to_zh=translate_to_zh,
           )
           doc['title_zh'] = translations.get('zh') or (title_text if primary_lang == 'zh' else None)
           doc['title_en'] = translations.get('en') or (title_text if primary_lang == 'en' else None)
   ```

**代码位置：** `indexer/document_transformer.py:168-298`

#### 4.1.3 标签字段

| ES 字段 | 数据来源 | 转换逻辑 |
|---------|----------|----------|
| `tags` | `spu_row['tags']` | 逗号分隔字符串 → 列表：`[tag.strip() for tag in tags_str.split(',') if tag.strip()]` |

**示例：**
- 数据库：`"标签1,标签2,标签3"`
- ES：`["标签1", "标签2", "标签3"]`

#### 4.1.4 图片字段

| ES 字段 | 数据来源 | 转换逻辑 |
|---------|----------|----------|
| `image_url` | `spu_row['image_src']` | 如果 URL 不以 `http` 开头，添加 `//` 前缀 |

**代码位置：** `indexer/document_transformer.py:390-396`

#### 4.1.5 销量字段

| ES 字段 | 数据来源 | 转换逻辑 |
|---------|----------|----------|
| `sales` | `spu_row['fake_sales']` | 转换为整数，如果为空或转换失败则设为 0 |

#### 4.1.6 时间字段

| ES 字段 | 数据来源 | 转换逻辑 |
|---------|----------|----------|
| `create_time` | `spu_row['create_time']` | 转换为 ISO 格式字符串：`datetime.isoformat()` |
| `update_time` | `spu_row['update_time']` | 转换为 ISO 格式字符串：`datetime.isoformat()` |

---

### 4.2 类别（Category）字段映射

类别字段是复杂字段，涉及多级分类的解析和映射。

#### 4.2.1 类别数据来源

类别信息主要来自 SPU 表的以下字段：
- `category_path`: 类目路径，逗号分隔的类目ID列表（如：`"1,2,3"`）
- `category`: 类目名称（备用字段）
- `category_id`: 当前类目ID
- `category_level`: 类目层级

#### 4.2.2 类别映射流程

**步骤 1：加载分类映射**

在服务初始化时，从数据库加载全局分类映射：

```python
category_id_to_name = load_category_mapping(db_engine)
# 结果：{"1": "电子产品", "2": "手机", "3": "iPhone"}
```

**步骤 2：解析 category_path**

```python
if pd.notna(spu_row.get('category_path')):
    category_path = str(spu_row['category_path'])  # 例如："1,2,3"
    category_ids = [cid.strip() for cid in category_path.split(',') if cid.strip()]
    # 结果：["1", "2", "3"]
```

**步骤 3：ID 转名称**

```python
category_names = []
missing_ids = []
for cid in category_ids:
    if cid in self.category_id_to_name:
        category_names.append(self.category_id_to_name[cid])
    else:
        missing_ids.append(cid)
```

**步骤 4：数据质量检查**

如果存在缺失的类目ID（在映射中找不到），则：
- 记录错误日志
- **不写入任何类目字段**（视为没有类目）
- 不阻塞索引流程

**步骤 5：填充 ES 字段**

```python
if category_names:
    # 构建类目路径字符串
    category_path_str = '/'.join(category_names)
    doc['category_path_zh'] = category_path_str  # 例如："电子产品/手机/iPhone"
    
    # 填充分层类目名称
    if len(category_names) > 0:
        doc['category1_name'] = category_names[0]  # "电子产品"
    if len(category_names) > 1:
        doc['category2_name'] = category_names[1]  # "手机"
    if len(category_names) > 2:
        doc['category3_name'] = category_names[2]  # "iPhone"
```

**步骤 6：备用处理（category_path 为空时）**

如果 `category_path` 为空，使用 `category` 字段作为备选：

```python
elif pd.notna(spu_row.get('category')):
    category = str(spu_row['category'])
    doc['category_name_zh'] = category
    
    # 尝试从 category 字段解析多级分类（如果包含 "/"）
    if '/' in category:
        path_parts = category.split('/')
        if len(path_parts) > 0:
            doc['category1_name'] = path_parts[0].strip()
        if len(path_parts) > 1:
            doc['category2_name'] = path_parts[1].strip()
        if len(path_parts) > 2:
            doc['category3_name'] = path_parts[2].strip()
```

#### 4.2.3 类别相关 ES 字段

| ES 字段 | 类型 | 说明 |
|---------|------|------|
| `category_path_zh` | text | 类目路径字符串（如："电子产品/手机/iPhone"） |
| `category_path_en` | text | 类目路径英文（暂未实现） |
| `category1_name` | keyword | 一级类目名称 |
| `category2_name` | keyword | 二级类目名称 |
| `category3_name` | keyword | 三级类目名称 |
| `category_id` | keyword | 当前类目ID |
| `category_level` | integer | 类目层级 |

**代码位置：** `indexer/document_transformer.py:300-376`

---

### 4.3 多款式（SKU/Options）字段映射

多款式字段是最复杂的部分，涉及 SKU 嵌套结构、选项值提取、规格构建等。

#### 4.3.1 SKU 数据结构

一个 SPU 可以有多个 SKU，每个 SKU 代表一个商品变体（如：红色-L码、蓝色-M码）。

**ES 中的 SKU 嵌套结构：**

```json
{
  "skus": [
    {
      "sku_id": "123",
      "price": 99.99,
      "compare_at_price": 129.99,
      "sku_code": "SKU001",
      "stock": 100,
      "weight": 0.5,
      "weight_unit": "kg",
      "option1_value": "红色",
      "option2_value": "L",
      "option3_value": "棉",
      "image_src": "https://..."
    },
    {
      "sku_id": "124",
      "price": 99.99,
      "compare_at_price": 129.99,
      "sku_code": "SKU002",
      "stock": 50,
      "weight": 0.5,
      "weight_unit": "kg",
      "option1_value": "蓝色",
      "option2_value": "M",
      "option3_value": "棉",
      "image_src": "https://..."
    }
  ]
}
```

#### 4.3.2 SKU 处理流程

**步骤 1：构建 Option 名称映射**

从 Option 表获取选项名称：

```python
option_name_map = {}
if not options.empty:
    for _, opt_row in options.iterrows():
        position = opt_row.get('position')  # 1, 2, 3
        name = opt_row.get('name')          # "颜色", "尺寸", "材质"
        if pd.notna(position) and pd.notna(name):
            option_name_map[int(position)] = str(name)
# 结果：{1: "颜色", 2: "尺寸", 3: "材质"}
```

**步骤 2：转换每个 SKU**

```python
for _, sku_row in skus.iterrows():
    sku_data = self._transform_sku_row(sku_row, option_name_map)
    if sku_data:
        skus_list.append(sku_data)
```

**SKU 转换逻辑：**

```python
def _transform_sku_row(self, sku_row, option_name_map):
    sku_data = {
        'sku_id': str(sku_row['id']),
        'price': float(sku_row['price']) if pd.notna(sku_row.get('price')) else None,
        'compare_at_price': float(sku_row['compare_at_price']) if pd.notna(sku_row.get('compare_at_price')) else None,
        'sku_code': str(sku_row['sku']) if pd.notna(sku_row.get('sku')) else None,
        'stock': int(sku_row['inventory_quantity']) if pd.notna(sku_row.get('inventory_quantity')) else 0,
        'weight': float(sku_row['weight']) if pd.notna(sku_row.get('weight')) else None,
        'weight_unit': str(sku_row['weight_unit']) if pd.notna(sku_row.get('weight_unit')) else None,
        'image_src': str(sku_row['image_src']) if pd.notna(sku_row.get('image_src')) else None,
    }
    
    # 填充选项值
    if pd.notna(sku_row.get('option1')):
        sku_data['option1_value'] = str(sku_row['option1'])
    if pd.notna(sku_row.get('option2')):
        sku_data['option2_value'] = str(sku_row['option2'])
    if pd.notna(sku_row.get('option3')):
        sku_data['option3_value'] = str(sku_row['option3'])
    
    return sku_data
```

**步骤 3：收集聚合信息**

在处理 SKU 的同时，收集价格、重量、库存等聚合信息：

```python
prices = []
compare_prices = []
sku_prices = []
sku_weights = []
sku_weight_units = []
total_inventory = 0

for sku_data in skus_list:
    if 'price' in sku_data and sku_data['price'] is not None:
        prices.append(sku_data['price'])
        sku_prices.append(sku_data['price'])
    
    if 'compare_at_price' in sku_data and sku_data['compare_at_price'] is not None:
        compare_prices.append(sku_data['compare_at_price'])
    
    if 'weight' in sku_data and sku_data['weight'] is not None:
        sku_weights.append(int(float(sku_data['weight'])))
    
    if 'weight_unit' in sku_data and sku_data['weight_unit']:
        sku_weight_units.append(sku_data['weight_unit'])
    
    if 'stock' in sku_data and sku_data['stock'] is not None:
        total_inventory += sku_data['stock']
```

**步骤 4：填充 ES 字段**

```python
doc['skus'] = skus_list  # 嵌套 SKU 数组

# 价格范围
if prices:
    doc['min_price'] = float(min(prices))
    doc['max_price'] = float(max(prices))
else:
    doc['min_price'] = 0.0
    doc['max_price'] = 0.0

if compare_prices:
    doc['compare_at_price'] = float(max(compare_prices))

# SKU 扁平化字段（用于过滤和聚合）
doc['sku_prices'] = sku_prices           # [99.99, 99.99, 129.99]
doc['sku_weights'] = sku_weights          # [500, 500, 600]
doc['sku_weight_units'] = list(set(sku_weight_units))  # ["kg"]
doc['total_inventory'] = total_inventory  # 150
```

**代码位置：** `indexer/document_transformer.py:398-480`

#### 4.3.3 选项名称字段

从 Option 表获取选项名称，填充到 ES 文档的顶层：

```python
def _fill_option_names(self, doc, options):
    if not options.empty:
        sorted_options = options.sort_values('position')
        if len(sorted_options) > 0 and pd.notna(sorted_options.iloc[0].get('name')):
            doc['option1_name'] = str(sorted_options.iloc[0]['name'])  # "颜色"
        if len(sorted_options) > 1 and pd.notna(sorted_options.iloc[1].get('name')):
            doc['option2_name'] = str(sorted_options.iloc[1]['name'])  # "尺寸"
        if len(sorted_options) > 2 and pd.notna(sorted_options.iloc[2].get('name')):
            doc['option3_name'] = str(sorted_options.iloc[2]['name'])  # "材质"
```

**ES 字段：**
- `option1_name`: 选项1的名称（如："颜色"）
- `option2_name`: 选项2的名称（如："尺寸"）
- `option3_name`: 选项3的名称（如："材质"）

**代码位置：** `indexer/document_transformer.py:378-388`

#### 4.3.4 选项值字段

从所有 SKU 中提取选项值，去重后填充到 ES 文档：

```python
def _fill_option_values(self, doc, skus):
    option1_values = []
    option2_values = []
    option3_values = []
    
    for _, sku_row in skus.iterrows():
        if pd.notna(sku_row.get('option1')):
            option1_values.append(str(sku_row['option1']))
        if pd.notna(sku_row.get('option2')):
            option2_values.append(str(sku_row['option2']))
        if pd.notna(sku_row.get('option3')):
            option3_values.append(str(sku_row['option3']))
    
    # 根据配置决定是否写入索引（searchable_option_dimensions）
    if 'option1' in self.searchable_option_dimensions:
        doc['option1_values'] = list(set(option1_values))  # ["红色", "蓝色", "绿色"]
    else:
        doc['option1_values'] = []
    
    # 同样处理 option2_values 和 option3_values
```

**ES 字段：**
- `option1_values`: 选项1的所有值列表（如：`["红色", "蓝色", "绿色"]`）
- `option2_values`: 选项2的所有值列表（如：`["S", "M", "L"]`）
- `option3_values`: 选项3的所有值列表（如：`["棉", "涤纶"]`）

**注意：** 只有配置在 `searchable_option_dimensions` 中的选项才会被索引。

**代码位置：** `indexer/document_transformer.py:482-510`

#### 4.3.5 规格（Specifications）字段

规格字段用于支持按规格过滤和分面搜索，将 SKU 的选项值结构化存储。

**构建逻辑：**

```python
specifications = []

for _, sku_row in skus.iterrows():
    sku_id = str(sku_row['id'])
    
    # 如果 SKU 有 option1 且 Option 表中有对应的名称
    if pd.notna(sku_row.get('option1')) and 1 in option_name_map:
        specifications.append({
            'sku_id': sku_id,
            'name': option_name_map[1],      # "颜色"
            'value': str(sku_row['option1']) # "红色"
        })
    
    # 同样处理 option2 和 option3
    if pd.notna(sku_row.get('option2')) and 2 in option_name_map:
        specifications.append({
            'sku_id': sku_id,
            'name': option_name_map[2],      # "尺寸"
            'value': str(sku_row['option2']) # "L"
        })
    
    if pd.notna(sku_row.get('option3')) and 3 in option_name_map:
        specifications.append({
            'sku_id': sku_id,
            'name': option_name_map[3],      # "材质"
            'value': str(sku_row['option3']) # "棉"
        })

doc['specifications'] = specifications
```

**ES 字段结构：**

```json
{
  "specifications": [
    {"sku_id": "123", "name": "颜色", "value": "红色"},
    {"sku_id": "123", "name": "尺寸", "value": "L"},
    {"sku_id": "123", "name": "材质", "value": "棉"},
    {"sku_id": "124", "name": "颜色", "value": "蓝色"},
    {"sku_id": "124", "name": "尺寸", "value": "M"},
    {"sku_id": "124", "name": "材质", "value": "棉"}
  ]
}
```

**用途：**
- 支持按规格过滤：`{"specifications": {"name": "颜色", "value": "红色"}}`
- 支持规格分面：统计每个规格值的数量

**代码位置：** `indexer/document_transformer.py:459-478`

---

### 4.4 向量字段映射

#### 4.4.1 标题向量（title_embedding）

**生成时机：**
- 在增量索引中，批量生成 embedding（性能优化）
- 在单条查询中，按需生成

**生成逻辑：**

```python
# 批量生成（增量索引）
if enable_embedding and encoder and documents:
    title_texts = []
    title_doc_indices = []
    for i, (_, doc) in enumerate(documents):
        title_text = doc.get("title_en") or doc.get("title_zh")
        if title_text and str(title_text).strip():
            title_texts.append(str(title_text))
            title_doc_indices.append(i)
    
    if title_texts:
        embeddings = encoder.encode_batch(title_texts, batch_size=32)
        for j, emb in enumerate(embeddings):
            doc_idx = title_doc_indices[j]
            if isinstance(emb, np.ndarray):
                documents[doc_idx][1]["title_embedding"] = emb.tolist()
```

**ES 字段：**
- `title_embedding`: 1024 维浮点数组，用于语义搜索

**代码位置：** 
- 批量生成：`indexer/incremental_service.py:558-576`
- 单条生成：`indexer/document_transformer.py:586-619`

---

## 5. 数据转换完整流程

### 5.1 增量索引流程

**入口方法：** `IncrementalIndexerService.index_spus_to_es()`

**流程步骤：**

1. **参数处理**
   - 去重 SPU ID 列表
   - 生成索引名称（如果未提供）

2. **显式删除处理**（如果提供了 `delete_spu_ids`）
   - 遍历 `delete_spu_ids`，从 ES 中删除对应文档

3. **批量加载数据**
   - 调用 `_load_spus_for_spu_ids()` 批量加载 SPU 数据
   - 调用 `_load_skus_for_spu_ids()` 批量加载 SKU 数据
   - 调用 `_load_options_for_spu_ids()` 批量加载 Option 数据

4. **自动删除检测**
   - 检查 SPU 是否在数据库中存在
   - 检查 SPU 的 `deleted` 字段
   - 如果不存在或已删除，从 ES 中删除对应文档

5. **数据分组**
   - 将 SKU 按 `spu_id` 分组
   - 将 Option 按 `spu_id` 分组

6. **文档转换**
   - 遍历每个 SPU
   - 调用 `transformer.transform_spu_to_doc()` 转换为 ES 文档
   - 收集所有文档

7. **批量生成 Embedding**（如果启用）
   - 收集所有文档的标题文本
   - 批量调用 `encoder.encode_batch()` 生成 embedding
   - 填充到对应文档

8. **批量写入 ES**
   - 使用 `BulkIndexer` 批量写入文档
   - 使用 `spu_id` 作为文档 ID

9. **返回结果**
   - 返回成功/失败统计
   - 返回每个 SPU 的处理状态

**代码位置：** `indexer/incremental_service.py:391-679`

### 5.2 文档转换流程

**入口方法：** `SPUDocumentTransformer.transform_spu_to_doc()`

**流程步骤：**

1. **初始化文档字典**
   ```python
   doc = {}
   doc['tenant_id'] = str(tenant_id)
   doc['spu_id'] = str(spu_id)
   ```

2. **填充文本字段**（多语言）
   - 调用 `_fill_text_fields()` 填充 title、brief、description、vendor

3. **填充标签**
   - 解析 `tags` 字段，转换为列表

4. **填充类别字段**
   - 调用 `_fill_category_fields()` 处理类别映射

5. **填充选项名称**
   - 调用 `_fill_option_names()` 从 Option 表获取选项名称

6. **填充图片 URL**
   - 调用 `_fill_image_url()` 处理图片 URL

7. **填充销量**
   - 转换 `fake_sales` 为整数

8. **处理 SKU**
   - 调用 `_process_skus()` 处理所有 SKU
   - 构建 `skus` 嵌套数组
   - 构建 `specifications` 数组
   - 计算价格范围、总库存等聚合字段

9. **填充选项值**
   - 调用 `_fill_option_values()` 提取所有选项值

10. **填充时间字段**
    - 转换 `create_time` 和 `update_time` 为 ISO 格式

11. **返回文档**
    ```python
    return doc
    ```

**代码位置：** `indexer/document_transformer.py:57-166`

---

## 6. 特殊处理逻辑

### 6.1 删除检测

系统支持两种删除方式：

1. **显式删除**：通过 `delete_spu_ids` 参数显式指定要删除的 SPU
2. **自动检测删除**：
   - SPU 在数据库中不存在
   - SPU 的 `deleted` 字段为 `1` 或 `b'\x01'`

**删除处理：**

```python
# 检查 deleted 字段（可能是 bit 类型）
def _is_deleted_value(v):
    if isinstance(v, bytes):
        return v == b"\x01" or v == 1
    return bool(v)

spu_df["_is_deleted"] = spu_df["deleted"].apply(_is_deleted_value)
```

**代码位置：** `indexer/incremental_service.py:481-517`

### 6.2 数据质量检查

#### 6.2.1 类别映射缺失

如果 `category_path` 中的类目ID在映射中不存在：
- 记录错误日志
- 不写入任何类目字段（视为没有类目）
- 不阻塞索引流程

#### 6.2.2 标题缺失

如果 SPU 没有标题：
- 记录错误日志
- 继续处理（但可能影响搜索效果）

#### 6.2.3 SKU 缺失

如果 SPU 没有 SKU：
- 记录警告日志
- 继续处理（但价格、库存等字段可能为空）

### 6.3 性能优化

1. **批量查询**：使用 `IN` 子句批量查询，减少数据库往返
2. **缓存分类映射**：在服务初始化时预加载，避免重复查询
3. **缓存 Transformer**：按租户缓存 Transformer 实例，避免重复初始化
4. **批量生成 Embedding**：收集所有标题后批量生成，利用批处理性能
5. **批量写入 ES**：使用 `BulkIndexer` 批量写入，提高写入效率

---

## 7. ES 文档完整示例

```json
{
  "tenant_id": "1",
  "spu_id": "12345",
  "title_zh": "iPhone 15 Pro Max",
  "title_en": "iPhone 15 Pro Max",
  "brief_zh": "最新款 iPhone",
  "brief_en": "Latest iPhone",
  "description_zh": "详细描述...",
  "description_en": "Detailed description...",
  "vendor_zh": "Apple",
  "vendor_en": "Apple",
  "tags": ["手机", "智能手机", "Apple"],
  "category_path_zh": "电子产品/手机/iPhone",
  "category1_name": "电子产品",
  "category2_name": "手机",
  "category3_name": "iPhone",
  "category_id": "3",
  "category_level": 3,
  "option1_name": "颜色",
  "option2_name": "存储容量",
  "option3_name": null,
  "option1_values": ["深空黑色", "原色钛金属", "白色钛金属"],
  "option2_values": ["256GB", "512GB", "1TB"],
  "option3_values": [],
  "image_url": "https://example.com/image.jpg",
  "sales": 1000,
  "min_price": 8999.0,
  "max_price": 12999.0,
  "compare_at_price": 12999.0,
  "sku_prices": [8999.0, 10999.0, 12999.0],
  "sku_weights": [221, 221, 221],
  "sku_weight_units": ["g"],
  "total_inventory": 500,
  "create_time": "2024-01-01T00:00:00",
  "update_time": "2024-01-15T10:30:00",
  "title_embedding": [0.123, 0.456, ...],  // 1024维向量
  "skus": [
    {
      "sku_id": "1001",
      "price": 8999.0,
      "compare_at_price": 9999.0,
      "sku_code": "IP15PM-256-BLK",
      "stock": 100,
      "weight": 221.0,
      "weight_unit": "g",
      "option1_value": "深空黑色",
      "option2_value": "256GB",
      "option3_value": null,
      "image_src": "https://example.com/sku1.jpg"
    },
    {
      "sku_id": "1002",
      "price": 10999.0,
      "compare_at_price": 11999.0,
      "sku_code": "IP15PM-512-BLK",
      "stock": 200,
      "weight": 221.0,
      "weight_unit": "g",
      "option1_value": "深空黑色",
      "option2_value": "512GB",
      "option3_value": null,
      "image_src": "https://example.com/sku2.jpg"
    }
  ],
  "specifications": [
    {"sku_id": "1001", "name": "颜色", "value": "深空黑色"},
    {"sku_id": "1001", "name": "存储容量", "value": "256GB"},
    {"sku_id": "1002", "name": "颜色", "value": "深空黑色"},
    {"sku_id": "1002", "name": "存储容量", "value": "512GB"}
  ]
}
```

---

## 8. 总结

### 8.1 关键要点

1. **数据源**：三个主要表（SPU、SKU、Option）
2. **批量查询**：使用 `IN` 子句提高查询效率
3. **类别映射**：通过预加载的映射表将 ID 转换为名称
4. **多款式处理**：通过嵌套结构和扁平化字段支持复杂的 SKU 查询
5. **多语言支持**：自动翻译文本字段，支持中英文双语
6. **向量搜索**：批量生成 embedding，支持语义搜索

### 8.2 复杂字段处理总结

| 字段类型 | 处理方式 | 关键逻辑 |
|---------|---------|---------|
| **类别** | ID映射 + 路径解析 | 从 `category_path` 解析ID，通过映射表转换为名称，构建多级分类 |
| **SKU** | 嵌套数组 + 聚合计算 | 将每个 SKU 转换为嵌套对象，同时计算价格范围、总库存等聚合值 |
| **选项** | 名称 + 值分离 | 从 Option 表获取名称，从 SKU 表提取值，分别存储 |
| **规格** | 结构化数组 | 将 SKU 的选项值转换为 `{name, value, sku_id}` 结构，支持过滤和分面 |

---

## 附录：相关代码文件

- `indexer/incremental_service.py`: 增量索引服务，数据获取逻辑
- `indexer/document_transformer.py`: 文档转换器，字段填充逻辑
- `indexer/indexing_utils.py`: 工具函数，分类映射加载和 Transformer 创建
- `indexer/bulk_indexer.py`: 批量写入器，ES 写入逻辑