"""
  SPU文档转换器 - 公共转换逻辑。
  
  提取全量和增量索引共用的文档转换逻辑，避免代码冗余。

  输出文档结构与 mappings/search_products.json 及 索引字段说明v2 一致，
  供 search/searcher 与 search/es_query_builder 使用。
  - 多语言字段：title, brief, description, vendor, category_path, category_name_text

  - 嵌套：specifications, skus；向量：title_embedding、image_embedding（可选，需提供 image_encoder）

  """
  
  import pandas as pd

  import numpy as np

  import logging

  from typing import Dict, Any, Optional, List

  from indexer.product_enrich import build_index_content_fields

  
  logger = logging.getLogger(__name__)
  

  class SPUDocumentTransformer:
      """SPU文档转换器，将SPU、SKU、Option数据转换为ES文档格式。"""
  
      def __init__(
          self,
          category_id_to_name: Dict[str, str],
          searchable_option_dimensions: List[str],
          tenant_config: Optional[Dict[str, Any]] = None,
          translator: Optional[Any] = None,

          encoder: Optional[Any] = None,

          enable_title_embedding: bool = True,
          image_encoder: Optional[Any] = None,
          enable_image_embedding: bool = False,

      ):
          """
          初始化文档转换器。
  
          Args:
              category_id_to_name: 分类ID到名称的映射
              searchable_option_dimensions: 可搜索的option维度列表
              tenant_config: 租户配置（包含主语言和翻译配置）
              translator: 翻译器实例（可选，如果提供则启用翻译功能）

              encoder: 文本编码器实例（可选，用于生成title_embedding）
              enable_title_embedding: 是否启用标题向量化（默认True）

              image_encoder: 图片编码器实例（可选，需实现 encode_image_urls(urls) -> List[Optional[np.ndarray]]）
              enable_image_embedding: 是否启用图片向量化（默认False）

          """
          self.category_id_to_name = category_id_to_name
          self.searchable_option_dimensions = searchable_option_dimensions
          self.tenant_config = tenant_config or {}
          self.translator = translator

          self.encoder = encoder
          self.enable_title_embedding = enable_title_embedding

          self.image_encoder = image_encoder
          self.enable_image_embedding = bool(enable_image_embedding and image_encoder is not None)

      def _translate_index_languages(
          self,
          text: str,
          source_lang: str,
          index_languages: List[str],
          scene: str,
      ) -> Dict[str, Optional[str]]:
          translations: Dict[str, Optional[str]] = {}
          if not self.translator or not text or not str(text).strip():
              return translations
          for lang in index_languages:
              if lang == source_lang:
                  translations[lang] = text
                  continue
              translations[lang] = self.translator.translate(
                  text=text,
                  target_lang=lang,
                  source_lang=source_lang,

                  scene=scene,

              )
          return translations
  

      def _build_core_language_text_object(
          self,
          text: Optional[str],
          source_lang: str,
          scene: str = "general",
      ) -> Dict[str, str]:
          """
          构建与 mapping 中 core_language_text(_with_keyword) 对齐的对象。
          当前核心语言固定为 zh/en。
          """
          if not text or not str(text).strip():
              return {}
  
          source_text = str(text).strip()
          obj: Dict[str, str] = {}
  
          if source_lang in CORE_INDEX_LANGUAGES:
              obj[source_lang] = source_text
  
          if self.translator:
              translations = self._translate_index_languages(
                  text=source_text,
                  source_lang=source_lang,
                  index_languages=CORE_INDEX_LANGUAGES,
                  scene=scene,
              )
              for lang in CORE_INDEX_LANGUAGES:
                  val = translations.get(lang)
                  if val and str(val).strip():
                      obj[lang] = str(val).strip()
  
          return obj
  

      def transform_spu_to_doc(
          self,
          tenant_id: str,
          spu_row: pd.Series,
          skus: pd.DataFrame,

          options: pd.DataFrame,
          fill_llm_attributes: bool = True,

      ) -> Optional[Dict[str, Any]]:
          """
          将单个SPU行和其SKUs转换为ES文档。
  
          Args:
              tenant_id: 租户ID
              spu_row: SPU行数据
              skus: SKU数据DataFrame
              options: Option数据DataFrame
  
          Returns:
              ES文档字典
          """
          doc = {}
  
          # Tenant ID (required)
          doc['tenant_id'] = str(tenant_id)
  
          # SPU ID
          spu_id = spu_row['id']
          doc['spu_id'] = str(spu_id)
          
          # Validate required fields
          if pd.isna(spu_row.get('title')) or not str(spu_row['title']).strip():
              logger.error(f"SPU {spu_id} has no title, this may cause search issues")
  
          # 获取租户配置

          primary_lang = self.tenant_config.get('primary_language', 'en')

          # 文本字段处理（使用translator的内部逻辑自动处理多语言翻译）
          self._fill_text_fields(doc, spu_row, primary_lang)
          
          # 标题向量化处理（如果启用）
          if self.enable_title_embedding and self.encoder:
              self._fill_title_embedding(doc)

          # Tags：统一转成与 mapping 一致的 core-language object

          if pd.notna(spu_row.get('enriched_tags')):
              tags_str = str(spu_row['enriched_tags'])

              tags_obj = self._build_core_language_text_object(
                  tags_str,
                  source_lang=primary_lang,
                  scene="general",
              )
              if tags_obj:

                  doc['enriched_tags'] = tags_obj

  
          # Category相关字段
          self._fill_category_fields(doc, spu_row)
  
          # Option名称（从option表获取）
          self._fill_option_names(doc, options)
  
          # Image URL
          self._fill_image_url(doc, spu_row)
  

          # Image embedding（与 mappings/search_products.json 中 image_embedding 嵌套结构一致）
          if self.enable_image_embedding:
              self._fill_image_embedding(doc, spu_row, skus)
  

          # Sales (fake_sales)
          if pd.notna(spu_row.get('fake_sales')):
              try:
                  doc['sales'] = int(spu_row['fake_sales'])
              except (ValueError, TypeError):
                  doc['sales'] = 0
          else:
              doc['sales'] = 0
  
          # Process SKUs and build specifications
          skus_list, prices, compare_prices, sku_prices, sku_weights, sku_weight_units, total_inventory, specifications = \
              self._process_skus(skus, options)
  
          doc['skus'] = skus_list
          doc['specifications'] = specifications
  
          # 提取option值（根据配置的searchable_option_dimensions）
          self._fill_option_values(doc, skus)
  
          # Calculate price ranges
          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
  

          # SPU 不再读取 compare_at_price 字段；ES 的 compare_at_price 使用所有 SKU 中的最大对比价
          if compare_prices:

              doc['compare_at_price'] = float(max(compare_prices))
          else:
              doc['compare_at_price'] = None
  
          # SKU扁平化字段
          doc['sku_prices'] = sku_prices
          doc['sku_weights'] = sku_weights
          doc['sku_weight_units'] = list(set(sku_weight_units))  # 去重
          doc['total_inventory'] = total_inventory
  
          # Time fields - convert datetime to ISO format string for ES DATE type
          if pd.notna(spu_row.get('create_time')):
              create_time = spu_row['create_time']
              if hasattr(create_time, 'isoformat'):
                  doc['create_time'] = create_time.isoformat()
              else:
                  doc['create_time'] = str(create_time)
          
          if pd.notna(spu_row.get('update_time')):
              update_time = spu_row['update_time']
              if hasattr(update_time, 'isoformat'):
                  doc['update_time'] = update_time.isoformat()
              else:
                  doc['update_time'] = str(update_time)
  

          # 基于 LLM 的锚文本与语义属性（默认开启，失败时仅记录日志）

          # 注意：批处理场景（build-docs / bulk / incremental）应优先在外层攒批，
          # 再调用 fill_llm_attributes_batch()，避免逐条调用 LLM。
          if fill_llm_attributes:
              self._fill_llm_attributes(doc, spu_row)

          return doc
  

      def fill_llm_attributes_batch(self, docs: List[Dict[str, Any]], spu_rows: List[pd.Series]) -> None:
          """
          批量调用 LLM，为一批 doc 填充：
          - qanchors.{lang}

          - enriched_tags.{lang}

          - enriched_attributes[].value.{lang}

          - enriched_taxonomy_attributes[].value.{lang}

  
          设计目标：
          - 尽可能攒批调用 LLM；
          - 单次 LLM 调用最多 20 条（由 analyze_products 内部强制 cap 并自动拆批）。
          """
          if not docs or not spu_rows or len(docs) != len(spu_rows):
              return
  

          id_to_idx: Dict[str, int] = {}

          items: List[Dict[str, str]] = []

          for i, row in enumerate(spu_rows):
              raw_id = row.get("id")
              spu_id = "" if raw_id is None else str(raw_id).strip()
              title = str(row.get("title") or "").strip()
              if not spu_id or not title:
                  continue

              id_to_idx[spu_id] = i

              items.append(
                  {
                      "id": spu_id,
                      "title": title,
                      "brief": str(row.get("brief") or "").strip(),
                      "description": str(row.get("description") or "").strip(),
                      "image_url": str(row.get("image_src") or "").strip(),

                  }
              )
          if not items:

              return
  
          tenant_id = str(docs[0].get("tenant_id") or "").strip() or None

          try:

              # TODO: 从数据库读取该 tenant 的真实行业，并据此替换当前默认的 apparel profile。
              results = build_index_content_fields(
                  items=items,
                  tenant_id=tenant_id,
                  category_taxonomy_profile="apparel",
              )

          except Exception as e:
              logger.warning("LLM batch attribute fill failed: %s", e)
              return

          for result in results:
              spu_id = str(result.get("id") or "").strip()
              if not spu_id:

                  continue

              idx = id_to_idx.get(spu_id)
              if idx is None:
                  continue
              self._apply_content_enrichment(docs[idx], result)

      def _apply_content_enrichment(self, doc: Dict[str, Any], enrichment: Dict[str, Any]) -> None:
          """将 product_enrich 产出的 ES-ready 内容字段写入 doc。"""

          try:

              if enrichment.get("qanchors"):
                  doc["qanchors"] = enrichment["qanchors"]

              if enrichment.get("enriched_tags"):
                  doc["enriched_tags"] = enrichment["enriched_tags"]

              if enrichment.get("enriched_attributes"):
                  doc["enriched_attributes"] = enrichment["enriched_attributes"]

              if enrichment.get("enriched_taxonomy_attributes"):
                  doc["enriched_taxonomy_attributes"] = enrichment["enriched_taxonomy_attributes"]

          except Exception as e:

              logger.warning("Failed to apply enrichment to doc (spu_id=%s): %s", doc.get("spu_id"), e)

      def _fill_text_fields(
          self,
          doc: Dict[str, Any],
          spu_row: pd.Series,

          primary_lang: str

      ):

          """

          填充文本字段（根据租户 index_languages 处理多语言翻译）。
          仅写入 primary_language 及 index_languages 中配置的语言。

          """

          index_langs = self.tenant_config.get("index_languages") or ["en", "zh"]
  
          def _set_lang_obj(field_name: str, source_text: Optional[str], translations: Optional[Dict[str, Optional[str]]] = None):
              """写入多语言对象 doc[field_name] = {"zh": "...", "en": "...", ...}，仅包含 index_languages。"""

              if not source_text or not str(source_text).strip():
                  return

              obj: Dict[str, str] = {}
              src = str(source_text)
              obj[primary_lang] = src

              tr = translations or {}

              for lang in index_langs:
                  if lang == primary_lang:
                      continue
                  val = tr.get(lang)
                  if val and str(val).strip():
                      obj[lang] = str(val)

              if obj:
                  doc[field_name] = obj
  

          # Title
          if pd.notna(spu_row.get('title')):
              title_text = str(spu_row['title'])

              translations: Dict[str, Optional[str]] = {}

              if self.translator:

                  translations = self._translate_index_languages(
                      text=title_text,

                      source_lang=primary_lang,

                      index_languages=index_langs,

                      scene="sku_name",

                  )

              _set_lang_obj("title", title_text, translations)

  
          # Brief
          if pd.notna(spu_row.get('brief')):
              brief_text = str(spu_row['brief'])

              translations = {}

              if self.translator:

                  translations = self._translate_index_languages(
                      text=brief_text,

                      source_lang=primary_lang,

                      index_languages=index_langs,

                      scene="general",

                  )

              _set_lang_obj("brief", brief_text, translations)

  
          # Description
          if pd.notna(spu_row.get('description')):
              desc_text = str(spu_row['description'])

              translations = {}

              if self.translator:

                  translations = self._translate_index_languages(
                      text=desc_text,

                      source_lang=primary_lang,

                      index_languages=index_langs,

                      scene="general",

                  )

              _set_lang_obj("description", desc_text, translations)

  
          # Vendor
          if pd.notna(spu_row.get('vendor')):
              vendor_text = str(spu_row['vendor'])

              translations = {}

              if self.translator:

                  translations = self._translate_index_languages(
                      text=vendor_text,

                      source_lang=primary_lang,

                      index_languages=index_langs,

                      scene="general",

                  )

              _set_lang_obj("vendor", vendor_text, translations)

  
      def _fill_category_fields(self, doc: Dict[str, Any], spu_row: pd.Series):
          """填充类目相关字段。"""

          # 数据质量兜底：
          # - 当商品的类目ID在映射中不存在时，视为“不合法类目”，整条类目相关字段都不写入（当成没有类目）
          # - 仅记录错误日志，不阻塞索引流程
  

          primary_lang = self.tenant_config.get('primary_language', 'en')

          if pd.notna(spu_row.get('category_path')):
              category_path = str(spu_row['category_path'])
              
              # 解析category_path - 这是逗号分隔的类目ID列表
              category_ids = [cid.strip() for cid in category_path.split(',') if cid.strip()]

              # 将ID映射为名称，如果找不到映射则记录错误并跳过

              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)
              
              # 如果有缺失的类目ID，记录错误日志，不写入类目字段（当成没有类目）
              if missing_ids:
                  logger.error(
                      f"Category ID(s) not found in mapping for SPU {spu_row.get('id')} "
                      f"(title: {spu_row.get('title', 'N/A')}), missing_ids={missing_ids}, "
                      f"category_path={category_path}. Treating as no-category."
                  )
                  return

              
              # 构建类目路径字符串（用于搜索）
              if category_names:
                  category_path_str = '/'.join(category_names)

                  doc['category_path'] = {primary_lang: category_path_str}

                  # 与查询使用的 category_name_text.zh/en 对齐，便于类目搜索
                  doc['category_name_text'] = {primary_lang: category_path_str}

                  
                  # 填充分层类目名称
                  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]
          elif pd.notna(spu_row.get('category')):
              # 如果category_path为空，使用category字段作为category1_name的备选
              category = str(spu_row['category'])

              doc['category_name_text'] = {primary_lang: category}

              doc['category_name'] = 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()
              else:
                  # 如果category不包含"/"，直接作为category1_name
                  doc['category1_name'] = category.strip()
  
          if pd.notna(spu_row.get('category')):
              # 确保category相关字段都被设置（如果前面没有设置）
              category_name = str(spu_row['category'])

              if 'category_name_text' not in doc:
                  doc['category_name_text'] = {primary_lang: category_name}

              if 'category_name' not in doc:
                  doc['category_name'] = category_name
  
          if pd.notna(spu_row.get('category_id')):
              doc['category_id'] = str(int(spu_row['category_id']))
  
          if pd.notna(spu_row.get('category_level')):
              doc['category_level'] = int(spu_row['category_level'])
  
      def _fill_option_names(self, doc: Dict[str, Any], options: pd.DataFrame):
          """填充Option名称字段。"""
          if not options.empty:
              # 按position排序获取option名称
              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'])
  
      def _fill_image_url(self, doc: Dict[str, Any], spu_row: pd.Series):
          """填充图片URL字段。"""
          if pd.notna(spu_row.get('image_src')):
              image_src = str(spu_row['image_src'])
              if not image_src.startswith('http'):

                  # 仅当尚未是协议相对 URL 时才补 "//"，避免 "//host" 变成 "////host"
                  image_src = f"//{image_src}" if not image_src.startswith('//') else image_src

              doc['image_url'] = image_src
  

      def _fill_image_embedding(
          self, doc: Dict[str, Any], spu_row: pd.Series, skus: pd.DataFrame
      ) -> None:
          """
          填充 image_embedding 嵌套字段，与 mappings/search_products.json 一致：
          [{ "vector": [float, ...], "url": "..." }, ...]
          收集 SPU 主图 + SKU 图片 URL，去重后调用 image_encoder 生成向量。
          """
          urls: List[str] = []
          seen: set = set()
  
          def _add(url: str) -> None:
              if not url or not str(url).strip():
                  return
              u = str(url).strip()
              if u.startswith("//"):
                  u = "https:" + u
              if u not in seen:
                  seen.add(u)
                  urls.append(u)
  
          if doc.get("image_url"):
              _add(doc["image_url"])
          if pd.notna(spu_row.get("image_src")):
              _add(str(spu_row["image_src"]))
          if not skus.empty and "image_src" in skus.columns:
              for _, row in skus.iterrows():
                  if pd.notna(row.get("image_src")):
                      _add(str(row["image_src"]))
  
          if not urls:
              return

          vectors = self.image_encoder.encode_image_urls(urls, batch_size=8)
          if not vectors or len(vectors) != len(urls):
              raise RuntimeError(
                  f"image_embedding response length mismatch for SPU {doc.get('spu_id')}: "
                  f"expected {len(urls)}, got {0 if vectors is None else len(vectors)}"
              )
          out = []
          for url, vec in zip(urls, vectors):
              arr = np.asarray(vec, dtype=np.float32)
              if arr.ndim != 1 or arr.size == 0 or not np.isfinite(arr).all():
                  raise RuntimeError(
                      f"Invalid image embedding for SPU {doc.get('spu_id')} and URL {url}"
                  )
              out.append({"vector": arr.tolist(), "url": url})
          doc["image_embedding"] = out

      def _process_skus(
          self,
          skus: pd.DataFrame,
          options: pd.DataFrame
      ) -> tuple:
          """处理SKU数据，返回处理结果。"""
          skus_list = []
          prices = []
          compare_prices = []
          sku_prices = []
          sku_weights = []
          sku_weight_units = []
          total_inventory = 0
          specifications = []
  
          # 构建option名称映射（position -> name）
          option_name_map = {}
          if not options.empty:
              for _, opt_row in options.iterrows():
                  position = opt_row.get('position')
                  name = opt_row.get('name')
                  if pd.notna(position) and pd.notna(name):
                      option_name_map[int(position)] = str(name)
  

          primary_lang = self.tenant_config.get('primary_language', 'en')
  
          def _build_specification(name: str, raw_value: Any, sku_id: str) -> Optional[Dict[str, Any]]:
              value = "" if raw_value is None else str(raw_value).strip()
              if not value:
                  return None
              return {
                  'sku_id': sku_id,
                  'name': name,
                  'value_keyword': value,
                  'value_text': self._build_core_language_text_object(
                      value,
                      source_lang=primary_lang,
                      scene="general",
                  ) or normalize_core_text_field_value(value, primary_lang),
              }
  

          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)
                  
                  # 收集价格信息
                  if 'price' in sku_data and sku_data['price'] is not None:
                      try:
                          price_val = float(sku_data['price'])
                          prices.append(price_val)
                          sku_prices.append(price_val)
                      except (ValueError, TypeError):
                          pass
                  
                  if 'compare_at_price' in sku_data and sku_data['compare_at_price'] is not None:
                      try:
                          compare_prices.append(float(sku_data['compare_at_price']))
                      except (ValueError, TypeError):
                          pass
                  
                  # 收集重量信息
                  if 'weight' in sku_data and sku_data['weight'] is not None:
                      try:
                          sku_weights.append(int(float(sku_data['weight'])))
                      except (ValueError, TypeError):
                          pass
                  
                  if 'weight_unit' in sku_data and sku_data['weight_unit']:
                      sku_weight_units.append(str(sku_data['weight_unit']))
                  
                  # 收集库存信息
                  if 'stock' in sku_data and sku_data['stock'] is not None:
                      try:
                          total_inventory += int(sku_data['stock'])
                      except (ValueError, TypeError):
                          pass
                  
                  # 构建specifications（从SKU的option值和option表的name）
                  sku_id = str(sku_row['id'])
                  if pd.notna(sku_row.get('option1')) and 1 in option_name_map:

                      spec = _build_specification(option_name_map[1], sku_row['option1'], sku_id)
                      if spec:
                          specifications.append(spec)

                  if pd.notna(sku_row.get('option2')) and 2 in option_name_map:

                      spec = _build_specification(option_name_map[2], sku_row['option2'], sku_id)
                      if spec:
                          specifications.append(spec)

                  if pd.notna(sku_row.get('option3')) and 3 in option_name_map:

                      spec = _build_specification(option_name_map[3], sku_row['option3'], sku_id)
                      if spec:
                          specifications.append(spec)

  
          return skus_list, prices, compare_prices, sku_prices, sku_weights, sku_weight_units, total_inventory, specifications
  
      def _fill_option_values(self, doc: Dict[str, Any], skus: pd.DataFrame):
          """填充option值字段。"""
          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']))
          
          # 去重并根据配置决定是否写入索引
          if 'option1' in self.searchable_option_dimensions:
              doc['option1_values'] = list(set(option1_values)) if option1_values else []
          else:
              doc['option1_values'] = []
          
          if 'option2' in self.searchable_option_dimensions:
              doc['option2_values'] = list(set(option2_values)) if option2_values else []
          else:
              doc['option2_values'] = []
          
          if 'option3' in self.searchable_option_dimensions:
              doc['option3_values'] = list(set(option3_values)) if option3_values else []
          else:
              doc['option3_values'] = []
  

      def _fill_llm_attributes(self, doc: Dict[str, Any], spu_row: pd.Series) -> None:
          """

          调用 indexer.product_enrich 的高层内容理解入口，为当前 SPU 填充：

          - qanchors.{lang}

          - enriched_tags.{lang}

          - enriched_attributes[].value.{lang}

          """

          spu_id = str(spu_row.get("id") or "").strip()
          title = str(spu_row.get("title") or "").strip()
          if not spu_id or not title:
              return
  

          tenant_id = doc.get("tenant_id")

          try:

              # TODO: 从数据库读取该 tenant 的真实行业，并据此替换当前默认的 apparel profile。

              results = build_index_content_fields(
                  items=[
                      {
                          "id": spu_id,
                          "title": title,
                          "brief": str(spu_row.get("brief") or "").strip(),
                          "description": str(spu_row.get("description") or "").strip(),
                          "image_url": str(spu_row.get("image_src") or "").strip(),

                      }
                  ],
                  tenant_id=str(tenant_id),

                  category_taxonomy_profile="apparel",

              )
          except Exception as e:
              logger.warning("LLM attribute fill failed for SPU %s: %s", spu_id, e)
              return

          if results:
              self._apply_content_enrichment(doc, results[0])

      def _transform_sku_row(self, sku_row: pd.Series, option_name_map: Dict[int, str] = None) -> Optional[Dict[str, Any]]:
          """
          将SKU行转换为SKU对象。
  
          Args:
              sku_row: SKU行数据
              option_name_map: position到option名称的映射
  
          Returns:
              SKU字典
          """
          sku_data = {}
  
          # SKU ID
          sku_data['sku_id'] = str(sku_row['id'])
  
          # Price
          if pd.notna(sku_row.get('price')):
              try:
                  sku_data['price'] = float(sku_row['price'])
              except (ValueError, TypeError):
                  sku_data['price'] = None
          else:
              sku_data['price'] = None
  
          # Compare at price
          if pd.notna(sku_row.get('compare_at_price')):
              try:
                  sku_data['compare_at_price'] = float(sku_row['compare_at_price'])
              except (ValueError, TypeError):
                  sku_data['compare_at_price'] = None
          else:
              sku_data['compare_at_price'] = None
  
          # SKU Code
          if pd.notna(sku_row.get('sku')):
              sku_data['sku_code'] = str(sku_row['sku'])
  
          # Stock
          if pd.notna(sku_row.get('inventory_quantity')):
              try:
                  sku_data['stock'] = int(sku_row['inventory_quantity'])
              except (ValueError, TypeError):
                  sku_data['stock'] = 0
          else:
              sku_data['stock'] = 0
  
          # Weight
          if pd.notna(sku_row.get('weight')):
              try:
                  sku_data['weight'] = float(sku_row['weight'])
              except (ValueError, TypeError):
                  sku_data['weight'] = None
          else:
              sku_data['weight'] = None
  
          # Weight unit
          if pd.notna(sku_row.get('weight_unit')):
              sku_data['weight_unit'] = str(sku_row['weight_unit'])
  
          # Option values
          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'])
          
          # Image src
          if pd.notna(sku_row.get('image_src')):
              sku_data['image_src'] = str(sku_row['image_src'])
  
          return sku_data

      
      def _fill_title_embedding(self, doc: Dict[str, Any]) -> None:
          """
          填充标题向量化字段。
          

          使用英文标题（title.en）生成embedding。如果title.en不存在，则使用title.zh。

          
          Args:
              doc: ES文档字典
          """

          # 优先使用英文标题，如果没有则使用中文标题；再没有则取任意可用语言
          title_obj = doc.get("title") or {}
          if isinstance(title_obj, dict):
              title_text = title_obj.get("en") or title_obj.get("zh")
              if not title_text:
                  for v in title_obj.values():
                      if v and str(v).strip():
                          title_text = str(v)
                          break
          else:
              title_text = None

          
          if not title_text or not title_text.strip():
              logger.debug(f"No title text available for embedding, SPU: {doc.get('spu_id')}")
              return
          

          # 使用文本向量编码器生成 embedding
          # encode方法返回numpy数组，形状为(n, d)
          embeddings = self.encoder.encode(title_text)
          if embeddings is None or len(embeddings) == 0:
              raise RuntimeError(f"Failed to generate title embedding for SPU {doc.get('spu_id')}")
  
          embedding = np.asarray(embeddings[0], dtype=np.float32)
          if embedding.ndim != 1 or embedding.size == 0 or not np.isfinite(embedding).all():
              raise RuntimeError(f"Invalid title embedding for SPU {doc.get('spu_id')}")
          doc['title_embedding'] = embedding.tolist()
          logger.debug(f"Generated title_embedding for SPU: {doc.get('spu_id')}, title: {title_text[:50]}...")