refactor: rename product annotator to enrich and expand multilingual prompts

- Rename indexer/product_annotator.py to indexer/product_enrich.py and remove CSV-based CLI entrypoint, keeping only in-memory analyze_products API - Introduce dedicated product_enrich logging with separate verbose log file for full LLM requests/responses - Change indexer and /indexer/enrich-content API wiring to use indexer.product_enrich instead of indexer.product_annotator, updating tests and docs accordingly - Switch translate_prompts to share SUPPORTED_INDEX_LANGUAGES from tenant_config_loader and reuse that mapping for language code → display name - Remove hard SUPPORTED_LANGS constraint from LLM content-enrichment flow, driving languages directly from tenant/indexer configuration - Redesign LLM prompt generation to support multi-round, multi-language tables: first round in English, subsequent rounds translate the entire table (headers + cells) into target languages using English instructions

refactor: rename product annotator to enrich and expand multilingual prompts
- Rename indexer/product_annotator.py to indexer/product_enrich.py and remove CSV-based CLI entrypoint, keeping only in-memory analyze_products API - Introduce dedicated product_enrich logging with separate verbose log file for full LLM requests/responses - Change indexer and /indexer/enrich-content API wiring to use indexer.product_enrich instead of indexer.product_annotator, updating tests and docs accordingly - Switch translate_prompts to share SUPPORTED_INDEX_LANGUAGES from tenant_config_loader and reuse that mapping for language code → display name - Remove hard SUPPORTED_LANGS constraint from LLM content-enrichment flow, driving languages directly from tenant/indexer configuration - Redesign LLM prompt generation to support multi-round, multi-language tables: first round in English, subsequent rounds translate the entire table (headers + cells) into target languages using English instructions
tangwang
1 parent 137455af
Showing 18 changed files with 889 additions and 798 deletions Show diff stats
api/routes/indexer.py
api/translator_app.py
config/config.yaml
config/tenant_config_loader.py
config/translate_prompts.py
docs/DEVELOPER_GUIDE.md
docs/工作总结-微服务性能优化与架构.md
docs/搜索API对接指南.md
docs/翻译模块说明.md
indexer/document_transformer.py
indexer/product_annotator.py
indexer/product_enrich.py
providers/translation.py
query/deepl_provider.py
query/llm_translate.py
query/qwen_mt_translate.py
tests/ci/test_service_api_contracts.py
tests/test_process_products_batching.py
@@ -443,23 +443,12 @@ async def build_docs_from_db(request: BuildDocsFromDbRequest):
 def _run_enrich_content(tenant_id: str, items: List[Dict[str, str]], languages: List[str]) -> List[Dict[str, Any]]:
     """
-    同步执行内容理解：调用 product_annotator.analyze_products，按语言批量跑 LLM，
+    同步执行内容理解：调用 product_enrich.analyze_products，按语言批量跑 LLM，
     再聚合成每 SPU 的 qanchors、semantic_attributes、tags。供 run_in_executor 调用。
     """
-    from indexer.product_annotator import analyze_products, SUPPORTED_LANGS
-
-    llm_langs = [lang for lang in languages if lang in SUPPORTED_LANGS]
-    if not llm_langs:
-        return [
-            {
-                "spu_id": it["spu_id"],
-                "qanchors": {},
-                "semantic_attributes": [],
-                "tags": [],
-                "error": "no supported languages (supported: %s)" % sorted(SUPPORTED_LANGS),
-            }
-            for it in items
-        ]
+    from indexer.product_enrich import analyze_products
+
+    llm_langs = list(dict.fromkeys(languages)) or ["en"]
     products = [{"id": it["spu_id"], "title": (it.get("title") or "").strip()} for it in items]
     dim_keys = [
@@ -544,7 +533,7 @@ async def enrich_content(request: EnrichContentRequest):
     - 与 /indexer/build-docs 解耦，避免 build-docs 因 LLM 耗时过长而阻塞；调用方可
       先拿不含 qanchors/tags 的 doc，再异步或离线补齐本接口结果后更新 ES。
-    实现逻辑与 indexer.product_annotator.analyze_products 一致，支持多语言与 Redis 缓存。
+    实现逻辑与 indexer.product_enrich.analyze_products 一致，支持多语言与 Redis 缓存。
     """
     try:
         if not request.items:
@@ -88,7 +88,7 @@ import sys
 import logging
 import argparse
 import uvicorn
-from typing import Optional, Dict
+from typing import Dict, List, Optional, Sequence, Union
 from fastapi import FastAPI, HTTPException
 from fastapi.responses import JSONResponse
 from fastapi.middleware.cors import CORSMiddleware
@@ -162,7 +162,7 @@ def get_translator(model: str = &quot;qwen&quot;) -&gt; object:
 # Request/Response models
 class TranslationRequest(BaseModel):
     """Translation request model."""
-    text: str = Field(..., description="Text to translate")
+    text: Union[str, List[str]] = Field(..., description="Text to translate (string or list of strings)")
     target_lang: str = Field(..., description="Target language code (zh, en, ru, etc.)")
     source_lang: Optional[str] = Field(None, description="Source language code (optional, auto-detect if not provided)")
     model: Optional[str] = Field(None, description="Translation model: qwen-mt | deepl | llm")
@@ -183,10 +183,13 @@ class TranslationRequest(BaseModel):
 class TranslationResponse(BaseModel):
     """Translation response model."""
-    text: str = Field(..., description="Original text")
+    text: Union[str, List[str]] = Field(..., description="Original text (string or list)")
     target_lang: str = Field(..., description="Target language code")
     source_lang: Optional[str] = Field(None, description="Source language code (detected or provided)")
-    translated_text: str = Field(..., description="Translated text")
+    translated_text: Union[str, List[Optional[str]]] = Field(
+        ...,
+        description="Translated text (string or list; list elements may be null on failure)",
+    )
     status: str = Field(..., description="Translation status")
     model: str = Field(..., description="Translation model used")
@@ -260,11 +263,19 @@ async def translate(request: TranslationRequest):
     Supports both Qwen (default) and DeepL models via the 'model' parameter.
     """
-    if not request.text or not request.text.strip():
-        raise HTTPException(
-            status_code=400,
-            detail="Text cannot be empty"
-        )
+    # 允许 text 为字符串或字符串列表
+    if isinstance(request.text, list):
+        if not request.text:
+            raise HTTPException(
+                status_code=400,
+                detail="Text list cannot be empty"
+            )
+    else:
+        if not request.text or not request.text.strip():
+            raise HTTPException(
+                status_code=400,
+                detail="Text cannot be empty"
+            )
     if not request.target_lang:
         raise HTTPException(
@@ -283,24 +294,96 @@ async def translate(request: TranslationRequest):
     try:
         # Get translator instance for the specified model
         translator = get_translator(model=model)
-        
-        # Translate using the fixed prompt
+        raw_text = request.text
+
+        # 如果是列表，并且底层 provider 声明支持 batch，则直接传 list
+        if isinstance(raw_text, list) and getattr(translator, "supports_batch", False):
+            try:
+                translated_list = translator.translate(
+                    text=raw_text,
+                    target_lang=request.target_lang,
+                    source_lang=request.source_lang,
+                    context=request.context,
+                    prompt=request.prompt,
+                )
+            except Exception as exc:
+                logger.error("Batch translation failed: %s", exc, exc_info=True)
+                # 回退到逐条拆分逻辑
+                translated_list = None
+
+            if translated_list is not None:
+                # 规范化为 List[Optional[str]]，并保证长度对应
+                if not isinstance(translated_list, list):
+                    raise HTTPException(
+                        status_code=500,
+                        detail="Batch translation provider returned non-list result",
+                    )
+                normalized: List[Optional[str]] = []
+                for idx, item in enumerate(raw_text):
+                    if idx < len(translated_list):
+                        val = translated_list[idx]
+                    else:
+                        val = None
+                    # 失败语义：失败位置为 None
+                    normalized.append(val)
+
+                return TranslationResponse(
+                    text=raw_text,
+                    target_lang=request.target_lang,
+                    source_lang=request.source_lang,
+                    translated_text=normalized,
+                    status="success",
+                    model=str(getattr(translator, "model", model)),
+                )
+
+        # 否则：统一走逐条拆分逻辑（包括不支持 batch 的 provider）
+        if isinstance(raw_text, list):
+            results: List[Optional[str]] = []
+            for item in raw_text:
+                if item is None or not str(item).strip():
+                    # 空元素不视为失败，直接返回原值
+                    results.append(item)  # type: ignore[arg-type]
+                    continue
+                try:
+                    out = translator.translate(
+                        text=str(item),
+                        target_lang=request.target_lang,
+                        source_lang=request.source_lang,
+                        context=request.context,
+                        prompt=request.prompt,
+                    )
+                except Exception as exc:
+                    logger.warning("Per-item translation failed: %s", exc, exc_info=True)
+                    out = None
+                # 失败语义：该元素为 None
+                results.append(out)
+
+            return TranslationResponse(
+                text=raw_text,
+                target_lang=request.target_lang,
+                source_lang=request.source_lang,
+                translated_text=results,
+                status="success",
+                model=str(getattr(translator, "model", model)),
+            )
+
+        # 单文本模式：保持原有严格失败语义
         translated_text = translator.translate(
-            text=request.text,
+            text=raw_text,
             target_lang=request.target_lang,
             source_lang=request.source_lang,
             context=request.context,
             prompt=request.prompt,
         )
-        
+
         if translated_text is None:
             raise HTTPException(
                 status_code=500,
                 detail="Translation failed"
             )
-        
+
         return TranslationResponse(
-            text=request.text,
+            text=raw_text,
             target_lang=request.target_lang,
             source_lang=request.source_lang,
             translated_text=translated_text,
@@ -224,7 +224,7 @@ spu_config:
 # 租户配置（Tenant Configuration）
 # 每个租户可配置主语言 primary_language 与索引语言 index_languages（主市场语言，商家可勾选）
-# 默认 index_languages: [en, zh]，可配置为任意 SUPPORTED_INDEX_LANGUAGES 的子集
+# 默认 index_languages: [en, zh]，可配置为任意 SOURCE_LANG_CODE_MAP.keys() 的子集
 tenant_config:
   default:
     primary_language: "en"
@@ -11,7 +11,8 @@ from typing import Dict, Any, Optional, List
 logger = logging.getLogger(__name__)
 # 支持的索引语言：code -> display name（供商家勾选主市场语言等场景使用）
-SUPPORTED_INDEX_LANGUAGES: Dict[str, str] = {
+# 语言代码与展示名的双向映射（供翻译/LLM 提示等统一使用）
+SOURCE_LANG_CODE_MAP: Dict[str, str] = {
     "en": "English",
     "zh": "Chinese",
     "zh_tw": "Traditional Chinese",
@@ -51,6 +52,9 @@ SUPPORTED_INDEX_LANGUAGES: Dict[str, str] = {
     "bg": "Bulgarian",
 }
+TARGET_LANG_CODE_MAP: Dict[str, str] = {v: k for k, v in SOURCE_LANG_CODE_MAP.items()}
+
+
 def normalize_index_languages(value: Any, primary_language: str = "en") -> List[str]:
     """
     将 index_languages 配置规范化为合法语言代码列表。
@@ -67,7 +71,7 @@ def normalize_index_languages(value: Any, primary_language: str = &quot;en&quot;) -&gt; List[
         code = (item or "").strip().lower()
         if not code or code in seen:
             continue
-        if code in SUPPORTED_INDEX_LANGUAGES:
+        if code in SOURCE_LANG_CODE_MAP:
             valid.append(code)
             seen.add(code)
     return valid
@@ -91,11 +95,11 @@ def resolve_index_languages(
     to_en = bool(tenant_config.get("translate_to_en"))
     to_zh = bool(tenant_config.get("translate_to_zh"))
     langs: List[str] = []
-    if primary and primary in SUPPORTED_INDEX_LANGUAGES:
+    if primary and primary in SOURCE_LANG_CODE_MAP:
         langs.append(primary)
     for code in ("en", "zh"):
         if code not in langs and ((code == "en" and to_en) or (code == "zh" and to_zh)):
-            if code in SUPPORTED_INDEX_LANGUAGES:
+            if code in SOURCE_LANG_CODE_MAP:
                 langs.append(code)
     return langs if langs else list(default_index_languages)
-SOURCE_LANG_CODE_MAP = {
-    "en": "English",
-    "zh": "Chinese",
-    "zh_tw": "Traditional Chinese",
-    "ru": "Russian",
-    "ja": "Japanese",
-    "ko": "Korean",
-    "es": "Spanish",
-    "fr": "French",
-    "pt": "Portuguese",
-    "de": "German",
-    "it": "Italian",
-    "th": "Thai",
-    "vi": "Vietnamese",
-    "id": "Indonesian",
-    "ms": "Malay",
-    "ar": "Arabic",
-    "hi": "Hindi",
-    "he": "Hebrew",
-    "my": "Burmese",
-    "ta": "Tamil",
-    "ur": "Urdu",
-    "bn": "Bengali",
-    "pl": "Polish",
-    "nl": "Dutch",
-    "ro": "Romanian",
-    "tr": "Turkish",
-    "km": "Khmer",
-    "lo": "Lao",
-    "yue": "Cantonese",
-    "cs": "Czech",
-    "el": "Greek",
-    "sv": "Swedish",
-    "hu": "Hungarian",
-    "da": "Danish",
-    "fi": "Finnish",
-    "uk": "Ukrainian",
-    "bg": "Bulgarian",
-}
-
-TARGET_LANG_CODE_MAP = {v: k for k, v in SOURCE_LANG_CODE_MAP.items()}
+from config.tenant_config_loader import SOURCE_LANG_CODE_MAP, TARGET_LANG_CODE_MAP
 TRANSLATION_PROMPTS = {
     "general": {
@@ -92,7 +92,7 @@ MySQL (店匠 SPU/SKU)
 | indexer | 6004 | 索引 API（reindex/build-docs 等） | ✓ |
 | frontend | 6003 | 调试 UI | ✓ |
 | embedding | 6005 | 向量服务（文本/图片） | 可选 |
-| translator | 6006 | 翻译服务 | 可选 |
+| translator | 6006 | 翻译服务（`POST /translate` 支持单条或批量 list；批量失败用 `null` 占位） | 可选 |
 | reranker | 6007 | 重排服务 | 可选 |
 - 启动：`./run.sh` 仅启动 backend / indexer / frontend；需全功能时通过环境变量或脚本另行启动 embedding / translator / reranker。
@@ -170,6 +170,11 @@ docs/                # 文档（含本指南）
 - **原则**：业务代码只依赖 Provider 接口，不依赖具体 URL 或后端类型；新增调用方式（如新 Provider 类型）在对应 `providers/<capability>.py` 中实现并在工厂中注册。
 - **详见**：本指南 §7.2；[QUICKSTART.md](./QUICKSTART.md) §3。
+补充约定（翻译 provider）：
+
+- `translate(text=...)` 支持 `str` 与 `List[str]` 两种输入；当输入为列表时，输出必须与输入 **等长且顺序对应**，失败位置为 `None`（HTTP JSON 表现为 `null`）。
+- provider 可暴露 `supports_batch: bool`（property）用于标识其是否支持直接批量调用；上层在处理 `text` 为列表时可优先走 batch，否则逐条拆分调用。
+
 ### 4.9 suggestion
 - **职责**：建议索引的构建与检索：从 ES 商品索引与 MySQL 日志等构建 suggestion 索引；搜索 API 的 `/search/suggestions` 使用本模块。
@@ -101,9 +101,9 @@ instruction: &quot;Given a shopping query, rank product titles by relevance&quot;
 **具体内容**：
 - **接口**：`POST /indexer/enrich-content`（Indexer 服务端口 **6004**）。请求体为 `items` 数组，每项含 `spu_id`、`title`（必填）及可选多语言标题等；单次请求最多 **50 条**，建议批量调用。响应 `results` 与 `items` 一一对应，每项含 `spu_id`、`qanchors`（按语言键，如 `qanchors.zh`、`qanchors.en`，逗号分隔短语）、`semantic_attributes`、`tags`。
-- **索引侧**：微服务组合方式下，调用方先拿不含 qanchors/tags 的 doc，再调用本接口补齐后写入 ES 的 `qanchors.{lang}` 等字段；索引 transformer（`indexer/document_transformer.py`、`indexer/product_annotator.py`）内也可在构建 doc 时调用内容理解逻辑，写入 `qanchors.{lang}`。
+-- **索引侧**：微服务组合方式下，调用方先拿不含 qanchors/tags 的 doc，再调用本接口补齐后写入 ES 的 `qanchors.{lang}` 等字段；索引 transformer（`indexer/document_transformer.py`、`indexer/product_enrich.py`）内也可在构建 doc 时调用内容理解逻辑，写入 `qanchors.{lang}`。
 - **Suggest 侧**：`suggestion/builder.py` 从 ES 商品索引读取 `_source: ["id", "spu_id", "title", "qanchors"]`，对 `qanchors.{lang}` 用 `_split_qanchors` 拆成词条，以 `source="qanchor"` 加入候选，排序时 `qanchor` 权重大于纯 title（`add_product("qanchor", ...)`）；suggest 配置中 `sources: ["query_log", "qanchor"]` 表示候选来源包含 qanchor。
-- **实现与依赖**：内容理解内部使用大模型（需 `DASHSCOPE_API_KEY`），支持多语言与 Redis 缓存（如 `product_anchors`）；逻辑与 `indexer/product_annotator` 一致。
+- **实现与依赖**：内容理解内部使用大模型（需 `DASHSCOPE_API_KEY`），支持多语言与 Redis 缓存（如 `product_anchors`）；逻辑与 `indexer/product_enrich` 一致。
 **状态**：内容理解字段已接入索引与 suggest 链路；依赖内容理解（qanchors/tags）的**全量数据尚未全部完成一轮**，后续需持续跑满并校验效果。
@@ -1484,7 +1484,7 @@ curl -X POST &quot;http://127.0.0.1:6004/indexer/build-docs-from-db&quot; \
 ### 5.8 内容理解字段生成接口
 - **端点**: `POST /indexer/enrich-content`
-- **描述**: 根据商品内容信息批量生成 **qanchors**（锚文本）、**semantic_attributes**（语义属性）、**tags**（细分标签），供外部 indexer 在「微服务组合」方式下自行拼装 doc 时使用。请求以 `items[]` 传入商品内容字段（必填/可选见下表）。内部逻辑与 `indexer.product_annotator` 一致，支持多语言与 Redis 缓存；单次请求在线程池中执行，避免阻塞其他接口。
+- **描述**: 根据商品内容信息批量生成 **qanchors**（锚文本）、**semantic_attributes**（语义属性）、**tags**（细分标签），供外部 indexer 在「微服务组合」方式下自行拼装 doc 时使用。请求以 `items[]` 传入商品内容字段（必填/可选见下表）。内部逻辑与 `indexer.product_enrich` 一致，支持多语言与 Redis 缓存；单次请求在线程池中执行，避免阻塞其他接口。
 #### 请求参数
@@ -1862,6 +1862,10 @@ curl &quot;http://localhost:6007/health&quot;
 }
 ```
+> **失败语义（批量）**：当 `text` 为列表时，如果其中某条翻译失败，对应位置返回 `null`（即 `translated_text[i] = null`），并保持数组长度与顺序不变；接口整体仍返回 `status="success"`，用于避免“部分失败”导致整批请求失败。
+
+> **实现提示（可忽略）**：服务端会尽可能使用底层翻译 provider 的批量能力（若支持），否则自动拆分逐条翻译；无论采用哪种方式，上述批量契约保持一致。
+
 **完整 curl 示例**:
 中文 → 英文:
@@ -24,3 +24,83 @@ TRANSLATION_MODEL=qwen  # 或 deepl
 ## Provider 配置
 Provider 与 URL 在 `config/config.yaml` 的 `services.translation`。详见 [QUICKSTART.md](./QUICKSTART.md) §3 与 [DEVELOPER_GUIDE.md](./DEVELOPER_GUIDE.md) §7.2。
+
+## HTTP 接口契约（translator service，端口 6006）
+
+服务默认监听 `http://localhost:6006`，提供：
+
+- `POST /translate`: 文本翻译（支持 `qwen/qwen-mt`、`deepl`、`llm`）
+- `GET /health`: 健康检查
+
+### `POST /translate`
+
+**请求体**：
+
+```json
+{
+  "text": "商品名称",
+  "target_lang": "en",
+  "source_lang": "zh",
+  "model": "qwen",
+  "context": "sku_name",
+  "prompt": null
+}
+```
+
+- `text` 支持两种形式：
+  - 单条：`string`
+  - 批量：`string[]`（等长返回，顺序对应）
+
+**响应体**（单条）：
+
+```json
+{
+  "text": "商品名称",
+  "target_lang": "en",
+  "source_lang": "zh",
+  "translated_text": "Product name",
+  "status": "success",
+  "model": "qwen"
+}
+```
+
+**响应体**（批量）：
+
+```json
+{
+  "text": ["商品名称1", "商品名称2"],
+  "target_lang": "en",
+  "source_lang": "zh",
+  "translated_text": ["Product name 1", null],
+  "status": "success",
+  "model": "qwen"
+}
+```
+
+批量模式下，**单条失败用 `null` 占位**（即 `translated_text[i] = null`），保证长度与顺序一一对应，避免部分失败导致整批报错。
+
+---
+
+## 开发者接口约定（Provider / 代码调用）
+
+除 HTTP 微服务外，代码侧（如 query/indexer）通常通过 `providers.translation.create_translation_provider()` 获取翻译 provider 实例并调用 `translate()`。
+
+### 输入输出形状（Shape）
+
+- `translate(text=...)` 支持：
+  - **单条**：`text: str` → 返回 `Optional[str]`
+  - **批量**：`text: List[str]` → 返回 `List[Optional[str]]`
+- **批量语义**：返回列表必须与输入 **等长且顺序对应**；某条翻译失败时，对应位置为 `None`（HTTP JSON 中表现为 `null`）。
+
+### 批量能力标识（supports_batch）
+
+不同 provider 对批量的实现方式可能不同（例如：真正一次请求传多条，或内部循环逐条翻译并保持 shape）。
+
+为便于上层（如 `api/translator_app.py`）做最优调用，provider 可暴露：
+
+- `supports_batch: bool`（property）
+
+上层在收到 `text` 为列表时：
+
+- **若 `supports_batch=True`**：可以直接将列表传给 `translate(text=[...])`
+- **若 `supports_batch=False`**：上层会逐条拆分调用（仍保证输出列表一一对应、失败为 `null`）
@@ -14,7 +14,7 @@ import logging
 import re
 from typing import Dict, Any, Optional, List
 from config import ConfigLoader
-from indexer.product_annotator import analyze_products, SUPPORTED_LANGS
+from indexer.product_enrich import analyze_products
 logger = logging.getLogger(__name__)
@@ -225,9 +225,8 @@ class SPUDocumentTransformer:
             index_langs = self.tenant_config.get("index_languages") or ["en", "zh"]
         except Exception:
             index_langs = ["en", "zh"]
-        llm_langs = [lang for lang in index_langs if lang in SUPPORTED_LANGS]
-        if not llm_langs:
-            return
+        # 不再限制为固定 SUPPORTED_LANGS，直接按照租户配置的 index_languages 调用
+        llm_langs = list(dict.fromkeys(index_langs))  # 去重并保持顺序
         # 只对有 title 的 SPU 参与 LLM；其余跳过
         id_to_idx: Dict[str, int] = {}
@@ -651,7 +650,7 @@ class SPUDocumentTransformer:
     def _fill_llm_attributes(self, doc: Dict[str, Any], spu_row: pd.Series) -> None:
         """
-        调用 indexer.product_annotator.analyze_products，为当前 SPU 填充：
+        调用 indexer.product_enrich.analyze_products，为当前 SPU 填充：
         - qanchors.{lang}
         - semantic_attributes (lang/name/value)
         """
@@ -660,10 +659,8 @@ class SPUDocumentTransformer:
         except Exception:
             index_langs = ["en", "zh"]
-        # 只在支持的语言集合内调用
-        llm_langs = [lang for lang in index_langs if lang in SUPPORTED_LANGS]
-        if not llm_langs:
-            return
+        # 不再限制为固定 SUPPORTED_LANGS，直接按照租户配置的 index_languages 调用
+        llm_langs = list(dict.fromkeys(index_langs))  # 去重并保持顺序
         spu_id = str(spu_row.get("id") or "").strip()
         title = str(spu_row.get("title") or "").strip()
@@ -1,685 +0,0 @@
-#!/usr/bin/env python3
-"""
-商品品类分析脚本
-批量读取商品标题，调用大模型进行品类分析，并保存结果
-"""
-
-import csv
-import os
-import json
-import logging
-import time
-import hashlib
-from datetime import datetime
-from typing import List, Dict, Tuple, Any, Optional
-
-import redis
-import requests
-from pathlib import Path
-from requests.adapters import HTTPAdapter
-from urllib3.util.retry import Retry
-
-from config.env_config import REDIS_CONFIG
-
-# 配置
-BATCH_SIZE = 20
-# 华北2（北京）：https://dashscope.aliyuncs.com/compatible-mode/v1
-# 新加坡：https://dashscope-intl.aliyuncs.com/compatible-mode/v1
-# 美国（弗吉尼亚）：https://dashscope-us.aliyuncs.com/compatible-mode/v1
-API_BASE_URL = "https://dashscope-us.aliyuncs.com/compatible-mode/v1"
-MODEL_NAME = "qwen-flash"
-API_KEY = os.environ.get("DASHSCOPE_API_KEY")
-MAX_RETRIES = 3
-RETRY_DELAY = 5  # 秒
-REQUEST_TIMEOUT = 180  # 秒
-
-# 文件路径
-INPUT_FILE = "saas_170_products.csv"
-OUTPUT_DIR = Path("output_logs")
-OUTPUT_FILE = OUTPUT_DIR / "products_analyzed.csv"
-LOG_DIR = OUTPUT_DIR / "logs"
-
-# 设置独立日志（不影响全局 indexer.log）
-LOG_DIR.mkdir(parents=True, exist_ok=True)
-timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
-log_file = LOG_DIR / f"process_{timestamp}.log"
-
-logger = logging.getLogger("product_annotator")
-logger.setLevel(logging.INFO)
-
-if not logger.handlers:
-    formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")
-
-    file_handler = logging.FileHandler(log_file, encoding="utf-8")
-    file_handler.setFormatter(formatter)
-
-    stream_handler = logging.StreamHandler()
-    stream_handler.setFormatter(formatter)
-
-    logger.addHandler(file_handler)
-    logger.addHandler(stream_handler)
-
-    # 避免日志向根 logger 传播，防止写入 logs/indexer.log 等其他文件
-    logger.propagate = False
-
-
-# Redis 缓存（用于 anchors / 语义属性）
-ANCHOR_CACHE_PREFIX = REDIS_CONFIG.get("anchor_cache_prefix", "product_anchors")
-ANCHOR_CACHE_EXPIRE_DAYS = int(REDIS_CONFIG.get("anchor_cache_expire_days", 30))
-_anchor_redis: Optional[redis.Redis] = None
-
-try:
-    _anchor_redis = redis.Redis(
-        host=REDIS_CONFIG.get("host", "localhost"),
-        port=REDIS_CONFIG.get("port", 6479),
-        password=REDIS_CONFIG.get("password"),
-        decode_responses=True,
-        socket_timeout=REDIS_CONFIG.get("socket_timeout", 1),
-        socket_connect_timeout=REDIS_CONFIG.get("socket_connect_timeout", 1),
-        retry_on_timeout=REDIS_CONFIG.get("retry_on_timeout", False),
-        health_check_interval=10,
-    )
-    _anchor_redis.ping()
-    logger.info("Redis cache initialized for product anchors and semantic attributes")
-except Exception as e:
-    logger.warning(f"Failed to initialize Redis for anchors cache: {e}")
-    _anchor_redis = None
-
-
-LANG_LABELS: Dict[str, str] = {
-    "zh": "中文",
-    "en": "英文",
-    "de": "德文",
-    "ru": "俄文",
-    "fr": "法文",
-}
-
-SUPPORTED_LANGS = set(LANG_LABELS.keys())
-
-SYSTEM_MESSAGES: Dict[str, str] = {
-    "zh": (
-        "你是一名电商平台的商品标注员，你的工作是对输入的每个商品进行理解、分析和标注，"
-        "并按要求格式返回 Markdown 表格。所有输出内容必须为中文。"
-    ),
-    "en": (
-        "You are a product annotator for an e-commerce platform. "
-        "For each input product, you must understand, analyze and label it, "
-        "and return a Markdown table strictly following the requested format. "
-        "All output must be in English."
-    ),
-    "de": (
-        "Du bist ein Produktannotator für eine E‑Commerce‑Plattform. "
-        "Du sollst jedes Eingabeprodukt verstehen, analysieren und beschriften "
-        "und eine Markdown-Tabelle im geforderten Format zurückgeben. "
-        "Alle Ausgaben müssen auf Deutsch sein."
-    ),
-    "ru": (
-        "Вы — разметчик товаров для платформы электронной коммерции. "
-        "Ваша задача — понимать, анализировать и размечать каждый товар "
-        "и возвращать таблицу Markdown в требуемом формате. "
-        "Весь вывод должен быть на русском языке."
-    ),
-    "fr": (
-        "Vous êtes annotateur de produits pour une plateforme e‑commerce. "
-        "Pour chaque produit en entrée, vous devez le comprendre, l’analyser et l’annoter, "
-        "puis renvoyer un tableau Markdown au format demandé. "
-        "Toute la sortie doit être en français."
-    ),
-}
-
-
-def _make_anchor_cache_key(
-    title: str,
-    target_lang: str,
-    tenant_id: Optional[str] = None,
-) -> str:
-    """构造 anchors/语义属性的缓存 key。"""
-    base = (tenant_id or "global").strip()
-    h = hashlib.md5(title.encode("utf-8")).hexdigest()
-    return f"{ANCHOR_CACHE_PREFIX}:{base}:{target_lang}:{h}"
-
-
-def _get_cached_anchor_result(
-    title: str,
-    target_lang: str,
-    tenant_id: Optional[str] = None,
-) -> Optional[Dict[str, Any]]:
-    if not _anchor_redis:
-        return None
-    try:
-        key = _make_anchor_cache_key(title, target_lang, tenant_id)
-        raw = _anchor_redis.get(key)
-        if not raw:
-            return None
-        return json.loads(raw)
-    except Exception as e:
-        logger.warning(f"Failed to get anchor cache: {e}")
-        return None
-
-
-def _set_cached_anchor_result(
-    title: str,
-    target_lang: str,
-    result: Dict[str, Any],
-    tenant_id: Optional[str] = None,
-) -> None:
-    if not _anchor_redis:
-        return
-    try:
-        key = _make_anchor_cache_key(title, target_lang, tenant_id)
-        ttl = ANCHOR_CACHE_EXPIRE_DAYS * 24 * 3600
-        _anchor_redis.setex(key, ttl, json.dumps(result, ensure_ascii=False))
-    except Exception as e:
-        logger.warning(f"Failed to set anchor cache: {e}")
-
-
-def create_prompt(products: List[Dict[str, str]], target_lang: str = "zh") -> str:
-    """根据目标语言创建 LLM 提示词和表头说明。"""
-    if target_lang == "en":
-        prompt = """Please analyze each input product title and extract the following information:
-
-1. Product title: a natural English product name derived from the input title
-2. Category path: from broad to fine-grained category, separated by ">" (e.g. Clothing>Women>Dresses>Work Dress)
-3. Fine-grained tags: style / features / attributes (e.g. floral, waist-cinching, French style)
-4. Target audience: gender / age group, etc. (e.g. young women)
-5. Usage scene
-6. Applicable season
-7. Key attributes
-8. Material description
-9. Functional features
-10. Selling point: one concise key selling sentence for recommendation
-11. Anchor text: a set of words or phrases that could be used by users as search queries for this product, covering category, fine-grained tags, functional attributes, usage scenes, etc.
-
-Input product list:
-
-"""
-        prompt_tail = """
-Please strictly return a Markdown table in the following format. For any column that can contain multiple values, separate values with commas. Do not add any other explanations:
-
-| No. | Product title | Category path | Fine-grained tags | Target audience | Usage scene | Season | Key attributes | Material | Features | Selling point | Anchor text |
-|----|----|----|----|----|----|----|----|----|----|----|----|
-"""
-    elif target_lang == "de":
-        prompt = """Bitte analysiere jeden eingegebenen Produkttitel und extrahiere die folgenden Informationen:
-
-1. Produkttitel: ein natürlicher deutscher Produkttitel basierend auf dem Eingangstitel
-2. Kategoriepfad: von Oberkategorie bis Feinkategorie, getrennt durch ">" (z. B. Kleidung>Damen>Kleider>Businesskleid)
-3. Feinkörnige Tags: Stil / Merkmale / Eigenschaften (z. B. Blumenmuster, tailliert, französischer Stil)
-4. Zielgruppe: Geschlecht / Altersgruppe usw. (z. B. junge Frauen)
-5. Einsatzszenario
-6. Geeignete Saison
-7. Wichtige Attribute
-8. Materialbeschreibung
-9. Funktionale Merkmale
-10. Verkaufsargument: ein prägnanter, einzeiliger Haupt-Selling-Point für Empfehlungen
-11. Ankertexte: eine Menge von Wörtern oder Phrasen, die Nutzer als Suchanfragen für dieses Produkt verwenden könnten und die Kategorie, feine Tags, Funktion und Nutzungsszenarien abdecken.
-
-Eingabeliste der Produkte:
-
-"""
-        prompt_tail = """
-Gib bitte strikt eine Markdown-Tabelle im folgenden Format zurück. Mehrere Werte in einer Spalte werden durch Kommas getrennt. Füge keine weiteren Erklärungen hinzu:
-
-| Nr. | Produkttitel | Kategoriepfad | Feintags | Zielgruppe | Einsatzszenario | Saison | Wichtige Attribute | Material | Merkmale | Verkaufsargument | Ankertexte |
-|----|----|----|----|----|----|----|----|----|----|----|----|
-"""
-    elif target_lang == "ru":
-        prompt = """Пожалуйста, проанализируйте каждый входной заголовок товара и извлеките следующую информацию:
-
-1. Заголовок товара: естественное русскоязычное название товара на основе исходного заголовка
-2. Путь категории: от широкой до узкой категории, разделённый символом ">" (например: Одежда>Женская одежда>Платья>Деловое платье)
-3. Детализированные теги: стиль / особенности / характеристики (например: цветочный принт, приталенный, французский стиль)
-4. Целевая аудитория: пол / возрастная группа и т. п. (например: молодые женщины)
-5. Сценарий использования
-6. Подходящий сезон
-7. Ключевые характеристики
-8. Описание материала
-9. Функциональные особенности
-10. Торговое преимущество: одно краткое ключевое предложение для рекомендаций
-11. Якорные запросы: набор слов или фраз, которые пользователи могут использовать в качестве поисковых запросов для этого товара, покрывающих категорию, детализированные теги, функциональные характеристики, сценарии использования и т. д.
-
-Список входных товаров:
-
-"""
-        prompt_tail = """
-Пожалуйста, строго верните Markdown‑таблицу в следующем формате. Для колонок с несколькими значениями разделяйте значения запятыми. Не добавляйте никаких дополнительных пояснений:
-
-| № | Заголовок товара | Путь категории | Детализированные теги | Целевая аудитория | Сценарий использования | Сезон | Ключевые характеристики | Материал | Особенности | Торговое преимущество | Якорные запросы |
-|----|----|----|----|----|----|----|----|----|----|----|----|
-"""
-    elif target_lang == "fr":
-        prompt = """Veuillez analyser chaque titre de produit en entrée et extraire les informations suivantes :
-
-1. Titre du produit : un titre de produit naturel en français basé sur le titre d’origine
-2. Chemin de catégorie : de la catégorie la plus large à la plus fine, séparées par ">" (par ex. Vêtements>Femme>Robes>Robe de travail)
-3. Tags détaillés : style / caractéristiques / attributs (par ex. fleuri, cintré, style français)
-4. Public cible : sexe / tranche d’âge, etc. (par ex. jeunes femmes)
-5. Scénario d’utilisation
-6. Saison adaptée
-7. Attributs clés
-8. Description du matériau
-9. Caractéristiques fonctionnelles
-10. Argument de vente : une phrase concise résumant le principal atout pour la recommandation
-11. Texte d’ancrage : un ensemble de mots ou d’expressions que les utilisateurs pourraient saisir comme requêtes de recherche pour ce produit, couvrant la catégorie, les tags détaillés, les fonctions, les scénarios d’usage, etc.
-
-Liste des produits en entrée :
-
-"""
-        prompt_tail = """
-Veuillez strictement renvoyer un tableau Markdown au format suivant. Pour toute colonne pouvant contenir plusieurs valeurs, séparez‑les par des virgules. N’ajoutez aucune autre explication :
-
-| N° | Titre du produit | Chemin de catégorie | Tags détaillés | Public cible | Scénario d’utilisation | Saison | Attributs clés | Matériau | Caractéristiques | Argument de vente | Texte d’ancrage |
-|----|----|----|----|----|----|----|----|----|----|----|----|
-"""
-    else:
-        # 默认中文版本
-        prompt = """请对输入的每条商品标题，分析并提取以下信息：
-
-1. 商品标题：将输入商品名称翻译为自然、完整的中文商品标题
-2. 品类路径：从大类到细分品类，用">"分隔（例如：服装>女装>裤子>工装裤）
-3. 细分标签：商品的风格、特点、功能等（例如：碎花，收腰，法式）
-4. 适用人群：性别/年龄段等（例如：年轻女性）
-5. 使用场景
-6. 适用季节
-7. 关键属性
-8. 材质说明 
-9. 功能特点
-10. 商品卖点：分析和提取一句话核心卖点，用于推荐理由
-11. 锚文本：生成一组能够代表该商品、并可能被用户用于搜索的词语或短语。这些词语应覆盖用户需求的各个维度，如品类、细分标签、功能特性、需求场景等等。
-
-输入商品列表：
-
-"""
-        prompt_tail = """
-请严格按照以下markdown表格格式返回，每列内部的多值内容都用逗号分隔，不要添加任何其他说明：
-
-| 序号 | 商品标题 | 品类路径 | 细分标签 | 适用人群 | 使用场景 | 适用季节 | 关键属性 | 材质说明 | 功能特点 | 商品卖点 | 锚文本 |
-|----|----|----|----|----|----|----|----|----|----|----|----|
-"""
-
-    for idx, product in enumerate(products, 1):
-        prompt += f'{idx}. {product["title"]}\n'
-    prompt += prompt_tail
-
-    return prompt
-
-
-def call_llm(prompt: str, target_lang: str = "zh") -> Tuple[str, str]:
-    """调用大模型API（带重试机制），按目标语言选择系统提示词。"""
-    headers = {
-        "Authorization": f"Bearer {API_KEY}",
-        "Content-Type": "application/json"
-    }
-
-    payload = {
-        "model": MODEL_NAME,
-        "messages": [
-            {
-                "role": "system",
-                "content": SYSTEM_MESSAGES.get(target_lang, SYSTEM_MESSAGES["zh"])
-            },
-            {
-                "role": "user",
-                "content": prompt
-            }
-        ],
-        "temperature": 0.3,
-        "top_p": 0.8
-    }
-
-    request_data = {
-        "headers": {k: v for k, v in headers.items() if k != "Authorization"},
-        "payload": payload
-    }
-
-    logger.info(f"\n{'='*80}")
-    logger.info(f"LLM Request (Model: {MODEL_NAME}):")
-    logger.info(json.dumps(request_data, ensure_ascii=False, indent=2))
-    logger.info(f"\nPrompt:\n{prompt}")
-
-    # 创建session，禁用代理
-    session = requests.Session()
-    session.trust_env = False  # 忽略系统代理设置
-
-    try:
-        # 重试机制
-        for attempt in range(MAX_RETRIES):
-            try:
-                response = session.post(
-                    f"{API_BASE_URL}/chat/completions",
-                    headers=headers,
-                    json=payload,
-                    timeout=REQUEST_TIMEOUT,
-                    proxies={"http": None, "https": None}  # 明确禁用代理
-                )
-
-                response.raise_for_status()
-                result = response.json()
-
-                logger.info(f"\nLLM Response:")
-                logger.info(json.dumps(result, ensure_ascii=False, indent=2))
-
-                content = result["choices"][0]["message"]["content"]
-                logger.info(f"\nExtracted Content:\n{content}")
-
-                return content, json.dumps(result, ensure_ascii=False)
-
-            except requests.exceptions.ProxyError as e:
-                logger.warning(f"Attempt {attempt + 1}/{MAX_RETRIES}: Proxy error - {str(e)}")
-                if attempt < MAX_RETRIES - 1:
-                    logger.info(f"Retrying in {RETRY_DELAY} seconds...")
-                    time.sleep(RETRY_DELAY)
-                else:
-                    raise
-
-            except requests.exceptions.RequestException as e:
-                logger.warning(f"Attempt {attempt + 1}/{MAX_RETRIES}: Request error - {str(e)}")
-                if attempt < MAX_RETRIES - 1:
-                    logger.info(f"Retrying in {RETRY_DELAY} seconds...")
-                    time.sleep(RETRY_DELAY)
-                else:
-                    raise
-
-            except Exception as e:
-                logger.error(f"Unexpected error on attempt {attempt + 1}/{MAX_RETRIES}: {str(e)}")
-                if attempt < MAX_RETRIES - 1:
-                    logger.info(f"Retrying in {RETRY_DELAY} seconds...")
-                    time.sleep(RETRY_DELAY)
-                else:
-                    raise
-
-    finally:
-        session.close()
-
-
-def parse_markdown_table(markdown_content: str) -> List[Dict[str, str]]:
-    """解析markdown表格内容"""
-    lines = markdown_content.strip().split('\n')
-    data = []
-    data_started = False
-
-    for line in lines:
-        line = line.strip()
-        if not line:
-            continue
-
-        # 表格行处理
-        if line.startswith('|'):
-            # 分隔行（---- 或 :---: 等；允许空格，如 "| ---- | ---- |"）
-            sep_chars = line.replace('|', '').strip().replace(' ', '')
-            if sep_chars and set(sep_chars) <= {'-', ':'}:
-                data_started = True
-                continue
-
-            # 首个表头行：无论语言如何，统一跳过
-            if not data_started:
-                # 等待下一行数据行
-                continue
-
-            # 解析数据行
-            parts = [p.strip() for p in line.split('|')]
-            parts = [p for p in parts if p]  # 移除空字符串
-
-            if len(parts) >= 2:
-                row = {
-                    "seq_no": parts[0],
-                    "title": parts[1],  # 商品标题（按目标语言）
-                    "category_path": parts[2] if len(parts) > 2 else "",  # 品类路径
-                    "tags": parts[3] if len(parts) > 3 else "",  # 细分标签
-                    "target_audience": parts[4] if len(parts) > 4 else "",  # 适用人群
-                    "usage_scene": parts[5] if len(parts) > 5 else "",  # 使用场景
-                    "season": parts[6] if len(parts) > 6 else "",  # 适用季节
-                    "key_attributes": parts[7] if len(parts) > 7 else "",  # 关键属性
-                    "material": parts[8] if len(parts) > 8 else "",  # 材质说明
-                    "features": parts[9] if len(parts) > 9 else "",  # 功能特点
-                    "selling_points": parts[10] if len(parts) > 10 else "",  # 商品卖点
-                    "anchor_text": parts[11] if len(parts) > 11 else ""  # 锚文本
-                }
-                data.append(row)
-
-    return data
-
-
-def process_batch(
-    batch_data: List[Dict[str, str]],
-    batch_num: int,
-    target_lang: str = "zh"
-) -> List[Dict[str, str]]:
-    """处理一个批次的数据"""
-    logger.info(f"\n{'#'*80}")
-    logger.info(f"Processing Batch {batch_num} ({len(batch_data)} items)")
-
-    # 创建提示词
-    prompt = create_prompt(batch_data, target_lang=target_lang)
-
-    # 调用LLM
-    try:
-        raw_response, full_response_json = call_llm(prompt, target_lang=target_lang)
-
-        # 解析结果
-        parsed_results = parse_markdown_table(raw_response)
-
-        logger.info(f"\nParsed Results ({len(parsed_results)} items):")
-        logger.info(json.dumps(parsed_results, ensure_ascii=False, indent=2))
-
-        # 映射回原始ID
-        results_with_ids = []
-        for i, parsed_item in enumerate(parsed_results):
-            if i < len(batch_data):
-                original_id = batch_data[i]["id"]
-                result = {
-                    "id": original_id,
-                    "lang": target_lang,
-                    "title_input": batch_data[i]["title"],  # 原始输入标题
-                    "title": parsed_item.get("title", ""),  # 模型生成的标题
-                    "category_path": parsed_item.get("category_path", ""),  # 品类路径
-                    "tags": parsed_item.get("tags", ""),  # 细分标签
-                    "target_audience": parsed_item.get("target_audience", ""),  # 适用人群
-                    "usage_scene": parsed_item.get("usage_scene", ""),  # 使用场景
-                    "season": parsed_item.get("season", ""),  # 适用季节
-                    "key_attributes": parsed_item.get("key_attributes", ""),  # 关键属性
-                    "material": parsed_item.get("material", ""),  # 材质说明
-                    "features": parsed_item.get("features", ""),  # 功能特点
-                    "selling_points": parsed_item.get("selling_points", ""),  # 商品卖点
-                    "anchor_text": parsed_item.get("anchor_text", "")  # 锚文本
-                }
-                results_with_ids.append(result)
-                logger.info(f"Mapped: seq={parsed_item['seq_no']} -> original_id={original_id}")
-
-        # 保存日志
-        batch_log = {
-            "batch_num": batch_num,
-            "timestamp": datetime.now().isoformat(),
-            "input_products": batch_data,
-            "raw_response": raw_response,
-            "full_response_json": full_response_json,
-            "parsed_results": parsed_results,
-            "final_results": results_with_ids
-        }
-
-        batch_log_file = LOG_DIR / f"batch_{batch_num:04d}_{timestamp}.json"
-        with open(batch_log_file, 'w', encoding='utf-8') as f:
-            json.dump(batch_log, f, ensure_ascii=False, indent=2)
-
-        logger.info(f"Batch log saved to: {batch_log_file}")
-
-        return results_with_ids
-
-    except Exception as e:
-        logger.error(f"Error processing batch {batch_num}: {str(e)}", exc_info=True)
-        # 返回空结果，保持ID映射
-        return [{
-            "id": item["id"],
-            "lang": target_lang,
-            "title_input": item["title"],
-            "title": "",
-            "category_path": "",
-            "tags": "",
-            "target_audience": "",
-            "usage_scene": "",
-            "season": "",
-            "key_attributes": "",
-            "material": "",
-            "features": "",
-            "selling_points": "",
-            "anchor_text": "",
-            "error": str(e),
-        } for item in batch_data]
-
-
-def read_products(input_file: str) -> List[Dict[str, str]]:
-    """读取CSV文件"""
-    products = []
-    with open(input_file, 'r', encoding='utf-8') as f:
-        reader = csv.DictReader(f)
-        for row in reader:
-            products.append({
-                "id": row["id"],
-                "title": row["title"]
-            })
-    return products
-
-
-def write_results(results: List[Dict[str, str]], output_file: Path):
-    """写入结果到CSV文件"""
-    output_file.parent.mkdir(parents=True, exist_ok=True)
-
-    fieldnames = [
-        "id",
-        "lang",
-        "title_input",
-        "title",
-        "category_path",
-        "tags",
-        "target_audience",
-        "usage_scene",
-        "season",
-        "key_attributes",
-        "material",
-        "features",
-        "selling_points",
-        "anchor_text",
-    ]
-
-    with open(output_file, 'w', encoding='utf-8', newline='') as f:
-        writer = csv.DictWriter(f, fieldnames=fieldnames)
-        writer.writeheader()
-        writer.writerows(results)
-
-    logger.info(f"\nResults written to: {output_file}")
-
-
-def main():
-    """主函数"""
-    if not API_KEY:
-        logger.error("Error: DASHSCOPE_API_KEY environment variable not set!")
-        return
-
-    logger.info(f"Starting product analysis process")
-    logger.info(f"Input file: {INPUT_FILE}")
-    logger.info(f"Output file: {OUTPUT_FILE}")
-    logger.info(f"Batch size: {BATCH_SIZE}")
-    logger.info(f"Model: {MODEL_NAME}")
-
-    # 读取产品数据
-    logger.info(f"\nReading products from {INPUT_FILE}...")
-    products = read_products(INPUT_FILE)
-    logger.info(f"Total products to process: {len(products)}")
-
-    # 分批处理
-    all_results = []
-    total_batches = (len(products) + BATCH_SIZE - 1) // BATCH_SIZE
-
-    for i in range(0, len(products), BATCH_SIZE):
-        batch_num = i // BATCH_SIZE + 1
-        batch = products[i:i + BATCH_SIZE]
-
-        logger.info(f"\nProgress: Batch {batch_num}/{total_batches}")
-
-        results = process_batch(batch, batch_num, target_lang="zh")
-        all_results.extend(results)
-
-        # 每处理完一个批次就写入一次（断点续传）
-        write_results(all_results, OUTPUT_FILE)
-        logger.info(f"Progress saved: {len(all_results)}/{len(products)} items completed")
-
-    logger.info(f"\n{'='*80}")
-    logger.info(f"Processing completed!")
-    logger.info(f"Total processed: {len(all_results)} items")
-    logger.info(f"Output file: {OUTPUT_FILE}")
-    logger.info(f"Log file: {log_file}")
-
-
-if __name__ == "__main__":
-    main()
-
-
-def analyze_products(
-    products: List[Dict[str, str]],
-    target_lang: str = "zh",
-    batch_size: Optional[int] = None,
-    tenant_id: Optional[str] = None,
-) -> List[Dict[str, Any]]:
-    """
-    库调用入口：根据输入+语言，返回锚文本及各维度信息。
-
-    Args:
-        products: [{"id": "...", "title": "..."}]
-        target_lang: 输出语言，需在 SUPPORTED_LANGS 内
-        batch_size: 批大小，默认使用全局 BATCH_SIZE
-    """
-    if not API_KEY:
-        raise RuntimeError("DASHSCOPE_API_KEY is not set, cannot call LLM")
-
-    if target_lang not in SUPPORTED_LANGS:
-        raise ValueError(f"Unsupported target_lang={target_lang}, supported={sorted(SUPPORTED_LANGS)}")
-
-    if not products:
-        return []
-
-    # 简单路径：索引阶段通常 batch_size=1，这里优先做单条缓存命中
-    if len(products) == 1:
-        p = products[0]
-        title = str(p.get("title") or "").strip()
-        if title:
-            cached = _get_cached_anchor_result(title, target_lang, tenant_id=tenant_id)
-            if cached:
-                logger.info(
-                    f"[analyze_products] Cache hit for title='{title[:50]}...', "
-                    f"lang={target_lang}, tenant_id={tenant_id or 'global'}"
-                )
-                return [cached]
-
-    # call_llm 一次处理上限固定为 BATCH_SIZE（默认 20）：
-    # - 尽可能攒批处理；
-    # - 即便调用方传入更大的 batch_size，也会自动按上限拆批。
-    req_bs = BATCH_SIZE if batch_size is None else int(batch_size)
-    bs = max(1, min(req_bs, BATCH_SIZE))
-    all_results: List[Dict[str, Any]] = []
-    total_batches = (len(products) + bs - 1) // bs
-
-    for i in range(0, len(products), bs):
-        batch_num = i // bs + 1
-        batch = products[i:i + bs]
-        logger.info(
-            f"[analyze_products] Processing batch {batch_num}/{total_batches}, "
-            f"size={len(batch)}, target_lang={target_lang}"
-        )
-        batch_results = process_batch(batch, batch_num=batch_num, target_lang=target_lang)
-        all_results.extend(batch_results)
-
-        # 写入缓存
-        for item in batch_results:
-            title_input = str(item.get("title_input") or "").strip()
-            if not title_input:
-                continue
-            if item.get("error"):
-                # 不缓存错误结果，避免放大临时故障
-                continue
-            try:
-                _set_cached_anchor_result(title_input, target_lang, item, tenant_id=tenant_id)
-            except Exception:
-                # 已在内部记录 warning
-                pass
-
-    return all_results
@@ -0,0 +1,516 @@
+#!/usr/bin/env python3
+"""
+商品内容理解与属性补充模块（product_enrich）
+
+提供基于 LLM 的商品锚文本 / 语义属性 / 标签等分析能力，
+供 indexer 与 API 在内存中调用（不再负责 CSV 读写）。
+"""
+
+import os
+import json
+import logging
+import time
+import hashlib
+from datetime import datetime
+from typing import List, Dict, Tuple, Any, Optional
+
+import redis
+import requests
+from pathlib import Path
+
+from config.env_config import REDIS_CONFIG
+from config.tenant_config_loader import SOURCE_LANG_CODE_MAP
+
+# 配置
+BATCH_SIZE = 20
+# 华北2（北京）：https://dashscope.aliyuncs.com/compatible-mode/v1
+# 新加坡：https://dashscope-intl.aliyuncs.com/compatible-mode/v1
+# 美国（弗吉尼亚）：https://dashscope-us.aliyuncs.com/compatible-mode/v1
+API_BASE_URL = "https://dashscope-us.aliyuncs.com/compatible-mode/v1"
+MODEL_NAME = "qwen-flash"
+API_KEY = os.environ.get("DASHSCOPE_API_KEY")
+MAX_RETRIES = 3
+RETRY_DELAY = 5  # 秒
+REQUEST_TIMEOUT = 180  # 秒
+
+# 日志路径
+OUTPUT_DIR = Path("output_logs")
+LOG_DIR = OUTPUT_DIR / "logs"
+
+# 设置独立日志（不影响全局 indexer.log）
+LOG_DIR.mkdir(parents=True, exist_ok=True)
+timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+log_file = LOG_DIR / f"product_enrich_{timestamp}.log"
+verbose_log_file = LOG_DIR / "product_enrich_verbose.log"
+
+# 主日志 logger：执行流程、批次信息等
+logger = logging.getLogger("product_enrich")
+logger.setLevel(logging.INFO)
+
+if not logger.handlers:
+    formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")
+
+    file_handler = logging.FileHandler(log_file, encoding="utf-8")
+    file_handler.setFormatter(formatter)
+
+    stream_handler = logging.StreamHandler()
+    stream_handler.setFormatter(formatter)
+
+    logger.addHandler(file_handler)
+    logger.addHandler(stream_handler)
+
+    # 避免日志向根 logger 传播，防止写入 logs/indexer.log 等其他文件
+    logger.propagate = False
+
+# 详尽日志 logger：专门记录 LLM 请求与响应
+verbose_logger = logging.getLogger("product_enrich_verbose")
+verbose_logger.setLevel(logging.INFO)
+
+if not verbose_logger.handlers:
+    verbose_formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")
+    verbose_file_handler = logging.FileHandler(verbose_log_file, encoding="utf-8")
+    verbose_file_handler.setFormatter(verbose_formatter)
+    verbose_logger.addHandler(verbose_file_handler)
+    verbose_logger.propagate = False
+
+
+# Redis 缓存（用于 anchors / 语义属性）
+ANCHOR_CACHE_PREFIX = REDIS_CONFIG.get("anchor_cache_prefix", "product_anchors")
+ANCHOR_CACHE_EXPIRE_DAYS = int(REDIS_CONFIG.get("anchor_cache_expire_days", 30))
+_anchor_redis: Optional[redis.Redis] = None
+
+try:
+    _anchor_redis = redis.Redis(
+        host=REDIS_CONFIG.get("host", "localhost"),
+        port=REDIS_CONFIG.get("port", 6479),
+        password=REDIS_CONFIG.get("password"),
+        decode_responses=True,
+        socket_timeout=REDIS_CONFIG.get("socket_timeout", 1),
+        socket_connect_timeout=REDIS_CONFIG.get("socket_connect_timeout", 1),
+        retry_on_timeout=REDIS_CONFIG.get("retry_on_timeout", False),
+        health_check_interval=10,
+    )
+    _anchor_redis.ping()
+    logger.info("Redis cache initialized for product anchors and semantic attributes")
+except Exception as e:
+    logger.warning(f"Failed to initialize Redis for anchors cache: {e}")
+    _anchor_redis = None
+
+
+SYSTEM_MESSAGES = (
+    "You are a product annotator for an e-commerce platform. "
+    "For each input product, you must understand, analyze and label it, "
+    "and return a Markdown table strictly following the requested format. "
+    "All output must be in English."
+)
+
+
+def _make_anchor_cache_key(
+    title: str,
+    target_lang: str,
+    tenant_id: Optional[str] = None,
+) -> str:
+    """构造 anchors/语义属性的缓存 key。"""
+    base = (tenant_id or "global").strip()
+    h = hashlib.md5(title.encode("utf-8")).hexdigest()
+    return f"{ANCHOR_CACHE_PREFIX}:{base}:{target_lang}:{h}"
+
+
+def _get_cached_anchor_result(
+    title: str,
+    target_lang: str,
+    tenant_id: Optional[str] = None,
+) -> Optional[Dict[str, Any]]:
+    if not _anchor_redis:
+        return None
+    try:
+        key = _make_anchor_cache_key(title, target_lang, tenant_id)
+        raw = _anchor_redis.get(key)
+        if not raw:
+            return None
+        return json.loads(raw)
+    except Exception as e:
+        logger.warning(f"Failed to get anchor cache: {e}")
+        return None
+
+
+def _set_cached_anchor_result(
+    title: str,
+    target_lang: str,
+    result: Dict[str, Any],
+    tenant_id: Optional[str] = None,
+) -> None:
+    if not _anchor_redis:
+        return
+    try:
+        key = _make_anchor_cache_key(title, target_lang, tenant_id)
+        ttl = ANCHOR_CACHE_EXPIRE_DAYS * 24 * 3600
+        _anchor_redis.setex(key, ttl, json.dumps(result, ensure_ascii=False))
+    except Exception as e:
+        logger.warning(f"Failed to set anchor cache: {e}")
+
+
+def create_prompt(products: List[Dict[str, str]], target_lang: str = "zh") -> str:
+    """根据目标语言创建 LLM 提示词和表头说明。
+
+    约定：
+    - 提示词始终使用英文；
+    - 当 target_lang == "en" 时，直接要求用英文分析并输出英文表头；
+    - 当 target_lang 为其他语言时，视作“多轮对话”的后续轮次：
+      * 默认上一轮已经用英文完成了分析；
+      * 当前轮只需要在保持结构和含义不变的前提下，将整张表格翻译为目标语言，
+        包含表头与所有单元格内容。
+    """
+    lang_name = SOURCE_LANG_CODE_MAP.get(target_lang, target_lang)
+
+    prompt = """Please analyze each input product title and extract the following information:
+
+1. Product title: a natural English product name derived from the input title
+2. Category path: from broad to fine-grained category, separated by ">" (e.g. Clothing>Women>Dresses>Work Dress)
+3. Fine-grained tags: style / features / attributes (e.g. floral, waist-cinching, French style)
+4. Target audience: gender / age group, etc. (e.g. young women)
+5. Usage scene
+6. Applicable season
+7. Key attributes
+8. Material description
+9. Functional features
+10. Selling point: one concise key selling sentence for recommendation
+11. Anchor text: a set of words or phrases that could be used by users as search queries for this product, covering category, fine-grained tags, functional attributes, usage scenes, etc.
+
+Input product list:
+
+"""
+
+    for idx, product in enumerate(products, 1):
+        prompt += f'{idx}. {product["title"]}\n'
+
+    if target_lang == "en":
+        # 英文首轮：直接要求英文表头 + 英文内容
+        prompt += """
+Please strictly return a Markdown table in the following format. For any column that can contain multiple values, separate values with commas. Do not add any other explanations:
+
+| No. | Product title | Category path | Fine-grained tags | Target audience | Usage scene | Season | Key attributes | Material | Features | Selling point | Anchor text |
+|----|----|----|----|----|----|----|----|----|----|----|----|
+"""
+    else:
+        # 非英文语言：视作“下一轮对话”，只做翻译，要求表头与内容全部用目标语言
+        prompt += f"""
+Now we will output the same table in {lang_name}.
+
+IMPORTANT:
+- Assume you have already generated the full table in English in a previous round.
+- In this round, you must output exactly the same table structure and content,
+  but fully translated into {lang_name}, including ALL column headers and ALL cell values.
+- Do NOT change the meaning, fields, or the number/order of rows and columns.
+- Keep valid Markdown table syntax.
+
+Please return ONLY the Markdown table in {lang_name}, without any extra explanations.
+"""
+
+    return prompt
+
+
+def call_llm(prompt: str, target_lang: str = "zh") -> Tuple[str, str]:
+    """调用大模型API（带重试机制），按目标语言选择系统提示词。"""
+    headers = {
+        "Authorization": f"Bearer {API_KEY}",
+        "Content-Type": "application/json",
+    }
+
+    payload = {
+        "model": MODEL_NAME,
+        "messages": [
+            {
+                "role": "system",
+                "content": SYSTEM_MESSAGES,
+            },
+            {
+                "role": "user",
+                "content": prompt,
+            },
+        ],
+        "temperature": 0.3,
+        "top_p": 0.8,
+    }
+
+    request_data = {
+        "headers": {k: v for k, v in headers.items() if k != "Authorization"},
+        "payload": payload,
+    }
+
+    # 主日志 + 详尽日志：LLM Request
+    logger.info(f"\n{'=' * 80}")
+    logger.info(f"LLM Request (Model: {MODEL_NAME}):")
+    logger.info(json.dumps(request_data, ensure_ascii=False, indent=2))
+    logger.info(f"\nPrompt:\n{prompt}")
+
+    verbose_logger.info(f"\n{'=' * 80}")
+    verbose_logger.info(f"LLM Request (Model: {MODEL_NAME}):")
+    verbose_logger.info(json.dumps(request_data, ensure_ascii=False, indent=2))
+    verbose_logger.info(f"\nPrompt:\n{prompt}")
+
+    # 创建session，禁用代理
+    session = requests.Session()
+    session.trust_env = False  # 忽略系统代理设置
+
+    try:
+        # 重试机制
+        for attempt in range(MAX_RETRIES):
+            try:
+                response = session.post(
+                    f"{API_BASE_URL}/chat/completions",
+                    headers=headers,
+                    json=payload,
+                    timeout=REQUEST_TIMEOUT,
+                    proxies={"http": None, "https": None},  # 明确禁用代理
+                )
+
+                response.raise_for_status()
+                result = response.json()
+
+                # 主日志 + 详尽日志：LLM Response
+                logger.info(f"\nLLM Response:")
+                logger.info(json.dumps(result, ensure_ascii=False, indent=2))
+
+                verbose_logger.info(f"\nLLM Response:")
+                verbose_logger.info(json.dumps(result, ensure_ascii=False, indent=2))
+
+                content = result["choices"][0]["message"]["content"]
+                logger.info(f"\nExtracted Content:\n{content}")
+                verbose_logger.info(f"\nExtracted Content:\n{content}")
+
+                return content, json.dumps(result, ensure_ascii=False)
+
+            except requests.exceptions.ProxyError as e:
+                logger.warning(f"Attempt {attempt + 1}/{MAX_RETRIES}: Proxy error - {str(e)}")
+                if attempt < MAX_RETRIES - 1:
+                    logger.info(f"Retrying in {RETRY_DELAY} seconds...")
+                    time.sleep(RETRY_DELAY)
+                else:
+                    raise
+
+            except requests.exceptions.RequestException as e:
+                logger.warning(f"Attempt {attempt + 1}/{MAX_RETRIES}: Request error - {str(e)}")
+                if attempt < MAX_RETRIES - 1:
+                    logger.info(f"Retrying in {RETRY_DELAY} seconds...")
+                    time.sleep(RETRY_DELAY)
+                else:
+                    raise
+
+            except Exception as e:
+                logger.error(f"Unexpected error on attempt {attempt + 1}/{MAX_RETRIES}: {str(e)}")
+                if attempt < MAX_RETRIES - 1:
+                    logger.info(f"Retrying in {RETRY_DELAY} seconds...")
+                    time.sleep(RETRY_DELAY)
+                else:
+                    raise
+
+    finally:
+        session.close()
+
+
+def parse_markdown_table(markdown_content: str) -> List[Dict[str, str]]:
+    """解析markdown表格内容"""
+    lines = markdown_content.strip().split("\n")
+    data = []
+    data_started = False
+
+    for line in lines:
+        line = line.strip()
+        if not line:
+            continue
+
+        # 表格行处理
+        if line.startswith("|"):
+            # 分隔行（---- 或 :---: 等；允许空格，如 "| ---- | ---- |"）
+            sep_chars = line.replace("|", "").strip().replace(" ", "")
+            if sep_chars and set(sep_chars) <= {"-", ":"}:
+                data_started = True
+                continue
+
+            # 首个表头行：无论语言如何，统一跳过
+            if not data_started:
+                # 等待下一行数据行
+                continue
+
+            # 解析数据行
+            parts = [p.strip() for p in line.split("|")]
+            parts = [p for p in parts if p]  # 移除空字符串
+
+            if len(parts) >= 2:
+                row = {
+                    "seq_no": parts[0],
+                    "title": parts[1],  # 商品标题（按目标语言）
+                    "category_path": parts[2] if len(parts) > 2 else "",  # 品类路径
+                    "tags": parts[3] if len(parts) > 3 else "",  # 细分标签
+                    "target_audience": parts[4] if len(parts) > 4 else "",  # 适用人群
+                    "usage_scene": parts[5] if len(parts) > 5 else "",  # 使用场景
+                    "season": parts[6] if len(parts) > 6 else "",  # 适用季节
+                    "key_attributes": parts[7] if len(parts) > 7 else "",  # 关键属性
+                    "material": parts[8] if len(parts) > 8 else "",  # 材质说明
+                    "features": parts[9] if len(parts) > 9 else "",  # 功能特点
+                    "selling_points": parts[10] if len(parts) > 10 else "",  # 商品卖点
+                    "anchor_text": parts[11] if len(parts) > 11 else "",  # 锚文本
+                }
+                data.append(row)
+
+    return data
+
+
+def process_batch(
+    batch_data: List[Dict[str, str]],
+    batch_num: int,
+    target_lang: str = "zh",
+) -> List[Dict[str, str]]:
+    """处理一个批次的数据"""
+    logger.info(f"\n{'#' * 80}")
+    logger.info(f"Processing Batch {batch_num} ({len(batch_data)} items)")
+
+    # 创建提示词
+    prompt = create_prompt(batch_data, target_lang=target_lang)
+
+    # 调用LLM
+    try:
+        raw_response, full_response_json = call_llm(prompt, target_lang=target_lang)
+
+        # 解析结果
+        parsed_results = parse_markdown_table(raw_response)
+
+        logger.info(f"\nParsed Results ({len(parsed_results)} items):")
+        logger.info(json.dumps(parsed_results, ensure_ascii=False, indent=2))
+
+        # 映射回原始ID
+        results_with_ids = []
+        for i, parsed_item in enumerate(parsed_results):
+            if i < len(batch_data):
+                original_id = batch_data[i]["id"]
+                result = {
+                    "id": original_id,
+                    "lang": target_lang,
+                    "title_input": batch_data[i]["title"],  # 原始输入标题
+                    "title": parsed_item.get("title", ""),  # 模型生成的标题
+                    "category_path": parsed_item.get("category_path", ""),  # 品类路径
+                    "tags": parsed_item.get("tags", ""),  # 细分标签
+                    "target_audience": parsed_item.get("target_audience", ""),  # 适用人群
+                    "usage_scene": parsed_item.get("usage_scene", ""),  # 使用场景
+                    "season": parsed_item.get("season", ""),  # 适用季节
+                    "key_attributes": parsed_item.get("key_attributes", ""),  # 关键属性
+                    "material": parsed_item.get("material", ""),  # 材质说明
+                    "features": parsed_item.get("features", ""),  # 功能特点
+                    "selling_points": parsed_item.get("selling_points", ""),  # 商品卖点
+                    "anchor_text": parsed_item.get("anchor_text", ""),  # 锚文本
+                }
+                results_with_ids.append(result)
+                logger.info(f"Mapped: seq={parsed_item['seq_no']} -> original_id={original_id}")
+
+        # 保存批次 JSON 日志到独立文件
+        batch_log = {
+            "batch_num": batch_num,
+            "timestamp": datetime.now().isoformat(),
+            "input_products": batch_data,
+            "raw_response": raw_response,
+            "full_response_json": full_response_json,
+            "parsed_results": parsed_results,
+            "final_results": results_with_ids,
+        }
+
+        batch_log_file = LOG_DIR / f"batch_{batch_num:04d}_{timestamp}.json"
+        with open(batch_log_file, "w", encoding="utf-8") as f:
+            json.dump(batch_log, f, ensure_ascii=False, indent=2)
+
+        logger.info(f"Batch log saved to: {batch_log_file}")
+
+        return results_with_ids
+
+    except Exception as e:
+        logger.error(f"Error processing batch {batch_num}: {str(e)}", exc_info=True)
+        # 返回空结果，保持ID映射
+        return [
+            {
+                "id": item["id"],
+                "lang": target_lang,
+                "title_input": item["title"],
+                "title": "",
+                "category_path": "",
+                "tags": "",
+                "target_audience": "",
+                "usage_scene": "",
+                "season": "",
+                "key_attributes": "",
+                "material": "",
+                "features": "",
+                "selling_points": "",
+                "anchor_text": "",
+                "error": str(e),
+            }
+            for item in batch_data
+        ]
+
+
+def analyze_products(
+    products: List[Dict[str, str]],
+    target_lang: str = "zh",
+    batch_size: Optional[int] = None,
+    tenant_id: Optional[str] = None,
+) -> List[Dict[str, Any]]:
+    """
+    库调用入口：根据输入+语言，返回锚文本及各维度信息。
+
+    Args:
+        products: [{"id": "...", "title": "..."}]
+        target_lang: 输出语言
+        batch_size: 批大小，默认使用全局 BATCH_SIZE
+    """
+    if not API_KEY:
+        raise RuntimeError("DASHSCOPE_API_KEY is not set, cannot call LLM")
+
+    if not products:
+        return []
+
+    # 简单路径：索引阶段通常 batch_size=1，这里优先做单条缓存命中
+    if len(products) == 1:
+        p = products[0]
+        title = str(p.get("title") or "").strip()
+        if title:
+            cached = _get_cached_anchor_result(title, target_lang, tenant_id=tenant_id)
+            if cached:
+                logger.info(
+                    f"[analyze_products] Cache hit for title='{title[:50]}...', "
+                    f"lang={target_lang}, tenant_id={tenant_id or 'global'}"
+                )
+                return [cached]
+
+    # call_llm 一次处理上限固定为 BATCH_SIZE（默认 20）：
+    # - 尽可能攒批处理；
+    # - 即便调用方传入更大的 batch_size，也会自动按上限拆批。
+    req_bs = BATCH_SIZE if batch_size is None else int(batch_size)
+    bs = max(1, min(req_bs, BATCH_SIZE))
+    all_results: List[Dict[str, Any]] = []
+    total_batches = (len(products) + bs - 1) // bs
+
+    for i in range(0, len(products), bs):
+        batch_num = i // bs + 1
+        batch = products[i : i + bs]
+        logger.info(
+            f"[analyze_products] Processing batch {batch_num}/{total_batches}, "
+            f"size={len(batch)}, target_lang={target_lang}"
+        )
+        batch_results = process_batch(batch, batch_num=batch_num, target_lang=target_lang)
+        all_results.extend(batch_results)
+
+        # 写入缓存
+        for item in batch_results:
+            title_input = str(item.get("title_input") or "").strip()
+            if not title_input:
+                continue
+            if item.get("error"):
+                # 不缓存错误结果，避免放大临时故障
+                continue
+            try:
+                _set_cached_anchor_result(title_input, target_lang, item, tenant_id=tenant_id)
+            except Exception:
+                # 已在内部记录 warning
+                pass
+
+    return all_results
+
@@ -2,7 +2,7 @@
 from __future__ import annotations
 import logging
-from typing import Any, Dict, Optional
+from typing import Any, Dict, List, Optional, Sequence, Union
 import requests
 from config.services_config import get_translation_config, get_translation_base_url
@@ -23,16 +23,49 @@ class HttpTranslationProvider:
         self.model = model or "qwen"
         self.timeout_sec = float(timeout_sec or 10.0)
+    @property
+    def supports_batch(self) -> bool:
+        """
+        Whether this provider supports list input natively.
+
+        当前实现中，我们已经在 `_translate_once` 内处理了 list，
+        所以可以直接视为支持 batch。
+        """
+        return True
+
     def _translate_once(
         self,
-        text: str,
+        text: Union[str, Sequence[str]],
         target_lang: str,
         source_lang: Optional[str] = None,
         context: Optional[str] = None,
         prompt: Optional[str] = None,
-    ) -> Optional[str]:
+    ) -> Union[Optional[str], List[Optional[str]]]:
+        # 允许 text 为单个字符串或字符串列表
+        if isinstance(text, (list, tuple)):
+            # 上游约定：列表输入时，输出列表一一对应；失败位置为 None
+            results: List[Optional[str]] = []
+            for item in text:
+                if item is None or not str(item).strip():
+                    # 空字符串/None 不视为失败，原样返回以保持语义
+                    results.append(item)  # type: ignore[arg-type]
+                    continue
+                try:
+                    single = self._translate_once(
+                        text=str(item),
+                        target_lang=target_lang,
+                        source_lang=source_lang,
+                        context=context,
+                        prompt=prompt,
+                    )
+                    results.append(single)  # type: ignore[arg-type]
+                except Exception:
+                    # 理论上不会进入，因为内部已捕获；兜底保持长度一致
+                    results.append(None)
+            return results
+
         if not text or not str(text).strip():
-            return text
+            return text  # type: ignore[return-value]
         try:
             url = f"{self.base_url}/translate"
             payload = {
@@ -62,12 +95,12 @@ class HttpTranslationProvider:
     def translate(
         self,
-        text: str,
+        text: Union[str, Sequence[str]],
         target_lang: str,
         source_lang: Optional[str] = None,
         context: Optional[str] = None,
         prompt: Optional[str] = None,
-    ) -> Optional[str]:
+    ) -> Union[Optional[str], List[Optional[str]]]:
         return self._translate_once(
             text=text,
             target_lang=target_lang,
@@ -10,7 +10,7 @@ from __future__ import annotations
 import logging
 import os
 import re
-from typing import Dict, Optional, Tuple
+from typing import Dict, List, Optional, Sequence, Tuple, Union
 import requests
 from config.services_config import get_translation_config
@@ -88,6 +88,14 @@ class DeepLProvider:
         if not self.api_key:
             logger.warning("DEEPL_AUTH_KEY not set; DeepL translation is unavailable")
+    @property
+    def supports_batch(self) -> bool:
+        """
+        DeepL HTTP API 本身支持一次传多条 text，这里先返回 False，
+        由上层逐条拆分，后续如果要真正批量，可调整实现。
+        """
+        return False
+
     def _resolve_request_context(
         self,
         target_lang: str,
@@ -108,12 +116,28 @@ class DeepLProvider:
     def translate(
         self,
-        text: str,
+        text: Union[str, Sequence[str]],
         target_lang: str,
         source_lang: Optional[str] = None,
         context: Optional[str] = None,
         prompt: Optional[str] = None,
-    ) -> Optional[str]:
+    ) -> Union[Optional[str], List[Optional[str]]]:
+        if isinstance(text, (list, tuple)):
+            results: List[Optional[str]] = []
+            for item in text:
+                if item is None or not str(item).strip():
+                    results.append(item)  # type: ignore[arg-type]
+                    continue
+                out = self.translate(
+                    text=str(item),
+                    target_lang=target_lang,
+                    source_lang=source_lang,
+                    context=context,
+                    prompt=prompt,
+                )
+                results.append(out)
+            return results
+
         if not self.api_key:
             return None
@@ -11,13 +11,14 @@ from __future__ import annotations
 import logging
 import os
 import time
-from typing import Optional
+from typing import List, Optional, Sequence, Union
 from openai import OpenAI
 from config.env_config import DASHSCOPE_API_KEY
 from config.services_config import get_translation_config
-from config.translate_prompts import TRANSLATION_PROMPTS, SOURCE_LANG_CODE_MAP
+from config.translate_prompts import TRANSLATION_PROMPTS
+from config.tenant_config_loader import SOURCE_LANG_CODE_MAP, TARGET_LANG_CODE_MAP
 logger = logging.getLogger(__name__)
@@ -96,6 +97,12 @@ class LLMTranslatorProvider:
         )
         self.client = self._create_client()
+    @property
+    def supports_batch(self) -> bool:
+        """Whether this provider efficiently supports list input."""
+        # 我们在 translate 中已经原生支持 list，所以这里返回 True
+        return True
+
     def _create_client(self) -> Optional[OpenAI]:
         api_key = DASHSCOPE_API_KEY or os.getenv("DASHSCOPE_API_KEY")
         if not api_key:
@@ -107,7 +114,7 @@ class LLMTranslatorProvider:
             logger.error("Failed to initialize llm translation client: %s", exc, exc_info=True)
             return None
-    def translate(
+    def _translate_single(
         self,
         text: str,
         target_lang: str,
@@ -148,7 +155,14 @@ class LLMTranslatorProvider:
             if not content:
                 logger.warning("[llm] Empty result | src=%s tgt=%s latency=%.1fms", src, tgt, latency_ms)
                 return None
-            logger.info("[llm] Success | src=%s tgt=%s src_text=%s response=%s latency=%.1fms", src, tgt, text, content, latency_ms)
+            logger.info(
+                "[llm] Success | src=%s tgt=%s src_text=%s response=%s latency=%.1fms",
+                src,
+                tgt,
+                text,
+                content,
+                latency_ms,
+            )
             return content
         except Exception as exc:
             latency_ms = (time.time() - start) * 1000
@@ -162,16 +176,56 @@ class LLMTranslatorProvider:
             )
             return None
+    def translate(
+        self,
+        text: Union[str, Sequence[str]],
+        target_lang: str,
+        source_lang: Optional[str] = None,
+        context: Optional[str] = None,
+        prompt: Optional[str] = None,
+    ) -> Union[Optional[str], List[Optional[str]]]:
+        """
+        Translate a single string or a list of strings.
+
+        - If input is a list, returns a list of the same length.
+        - Per-item failures are returned as None.
+        """
+        if isinstance(text, (list, tuple)):
+            results: List[Optional[str]] = []
+            for item in text:
+                # 保证一一对应，即使某个元素为空也占位
+                if item is None:
+                    results.append(None)
+                    continue
+                results.append(
+                    self._translate_single(
+                        text=str(item),
+                        target_lang=target_lang,
+                        source_lang=source_lang,
+                        context=context,
+                        prompt=prompt,
+                    )
+                )
+            return results
+
+        return self._translate_single(
+            text=str(text),
+            target_lang=target_lang,
+            source_lang=source_lang,
+            context=context,
+            prompt=prompt,
+        )
+
 def llm_translate(
-    text: str,
+    text: Union[str, Sequence[str]],
     target_lang: str,
     *,
     source_lang: Optional[str] = None,
     source_lang_label: Optional[str] = None,
     target_lang_label: Optional[str] = None,
     timeout_sec: Optional[float] = None,
-) -> Optional[str]:
+) -> Union[Optional[str], List[Optional[str]]]:
     provider = LLMTranslatorProvider(timeout_sec=timeout_sec or 30.0)
     return provider.translate(
         text=text,
@@ -7,14 +7,14 @@ import logging
 import os
 import re
 import time
-from typing import Dict, List, Optional
+from typing import Dict, List, Optional, Sequence, Union
 import redis
 from openai import OpenAI
 from config.env_config import DASHSCOPE_API_KEY, REDIS_CONFIG
 from config.services_config import get_translation_cache_config
-from config.translate_prompts import SOURCE_LANG_CODE_MAP
+from config.tenant_config_loader import SOURCE_LANG_CODE_MAP, TARGET_LANG_CODE_MAP
 logger = logging.getLogger(__name__)
@@ -62,6 +62,16 @@ class Translator:
         if self.use_cache and bool(cache_cfg.get("enabled", True)):
             self.redis_client = self._init_redis_client()
+    @property
+    def supports_batch(self) -> bool:
+        """
+        标记该 provider 已支持列表输入。
+
+        当前实现为循环单条调用（带缓存），不是真正的并行批量请求，
+        但对上层来说可以直接传 list，返回 list。
+        """
+        return True
+
     @staticmethod
     def _normalize_model(model: str) -> str:
         m = (model or "qwen").strip().lower()
@@ -117,14 +127,31 @@ class Translator:
     def translate(
         self,
-        text: str,
+        text: Union[str, Sequence[str]],
         target_lang: str,
         source_lang: Optional[str] = None,
         context: Optional[str] = None,
         prompt: Optional[str] = None,
-    ) -> Optional[str]:
-        if not text or not text.strip():
-            return text
+    ) -> Union[Optional[str], List[Optional[str]]]:
+        if isinstance(text, (list, tuple)):
+            results: List[Optional[str]] = []
+            for item in text:
+                if item is None or not str(item).strip():
+                    results.append(item)  # type: ignore[arg-type]
+                    continue
+                # 对于 batch，这里沿用单条的缓存与规则，逐条调用
+                out = self.translate(
+                    text=str(item),
+                    target_lang=target_lang,
+                    source_lang=source_lang,
+                    context=context,
+                    prompt=prompt,
+                )
+                results.append(out)
+            return results
+
+        if not text or not str(text).strip():
+            return text  # type: ignore[return-value]
         tgt = (target_lang or "").strip().lower()
         src = (source_lang or "").strip().lower() or None
@@ -342,7 +342,7 @@ def test_indexer_build_docs_from_db_contract(indexer_client: TestClient):
 def test_indexer_enrich_content_contract(indexer_client: TestClient, monkeypatch):
-    import indexer.product_annotator as process_products
+    import indexer.product_enrich as process_products
     def _fake_analyze_products(
         products: List[Dict[str, str]],
@@ -2,7 +2,7 @@ from __future__ import annotations
 from typing import Any, Dict, List
-import indexer.product_annotator as process_products
+import indexer.product_enrich as process_products
 def _mk_products(n: int) -> List[Dict[str, str]]: