last

tangwang
1 parent 985752f5
Showing 12 changed files with 1393 additions and 8 deletions Show diff stats
api/translator_app.py
config/config.yaml
docs/系统设计文档.md
indexer/document_transformer.py
indexer/test_indexing.py
providers/translation.py
query/__init__.py
query/llm_translate.py
query/qwen_mt_translate.py
query/test_translation.py
query/translator.py
tests/test_translator_failure_semantics.py
@@ -97,7 +97,7 @@ from pydantic import BaseModel, Field
 # Add parent directory to path
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
-from query.translator import Translator
+from query.qwen_mt_translate import Translator
 from config.env_config import DEEPL_AUTH_KEY, DASHSCOPE_API_KEY, REDIS_CONFIG
 # Configure logging
@@ -119,7 +119,7 @@ rerank:
 # 可扩展服务/provider 注册表（单一配置源）
 services:
   translation:
-    provider: "direct"  # direct | http | google(reserved)
+    provider: "llm"  # direct | http | google(reserved)
     base_url: "http://127.0.0.1:6006"
     model: "qwen"
     timeout_sec: 10.0
@@ -130,6 +130,12 @@ services:
         base_url: "http://127.0.0.1:6006"
         model: "qwen"
         timeout_sec: 10.0
+      llm:
+        model: "qwen-flash"
+        # 可选：覆盖 DashScope 兼容模式的 Endpoint 与超时
+        # base_url 留空则使用 DASHSCOPE_BASE_URL 或默认地域
+        base_url: ""
+        timeout_sec: 30.0
       google:
         enabled: false
         project_id: ""
@@ -384,6 +384,15 @@ query_config:
 实际代码中，通过通用的 translation provider 抽象来选择具体后端和模型，文档不固定绑定某一个具体翻译服务或模型名称，以保持可配置性。
+此外，为了支持**高质量、提示词可控的 LLM 翻译**（例如商品富化脚本、离线分析工具），在 `query/llm_translate.py` 中提供了一个独立的 LLM 翻译辅助模块：
+
+- **配置入口**：`config/config.yaml -> services.translation.providers.llm`，用于指定：
+  - `model`: 例如 `qwen-flash`（DashScope 兼容模式的对话模型）
+  - `base_url`: 可选；为空时使用环境变量 `DASHSCOPE_BASE_URL` 或默认 Endpoint
+  - `timeout_sec`: LLM 调用超时
+- **环境变量**：仍通过 `DASHSCOPE_API_KEY` 注入 DashScope API Key。
+- **使用方式**：主查询路径继续使用 machine translation（`query.translator.Translator`），只在需要更强表达控制的场景（如批量标注、产品分类脚本）中显式调用 `llm_translate()`。
+
 #### 功能特性
 1. **语言检测**：自动检测查询语言
 2. **智能翻译**：
@@ -20,7 +20,7 @@ logger = logging.getLogger(__name__)
 # Try to import translator (optional dependency)
 try:
-    from query.translator import Translator
+    from query.qwen_mt_translate import Translator
     TRANSLATOR_AVAILABLE = True
 except ImportError:
     TRANSLATOR_AVAILABLE = False
@@ -273,7 +273,7 @@ def test_document_transformer():
         tenant_config = tenant_config_loader.get_tenant_config('162')
         # 初始化翻译器（测试环境总是启用，具体翻译方向由tenant_config控制）
-        from query.translator import Translator
+        from query.qwen_mt_translate import Translator
         translator = Translator(
             api_key=config.query_config.translation_api_key,
             use_cache=True
@@ -0,0 +1,169 @@
+"""
+Translation provider - direct (in-process) or HTTP service.
+"""
+from __future__ import annotations
+
+import logging
+from typing import Any, Dict, List, Optional, Union
+
+from concurrent.futures import Future, ThreadPoolExecutor
+import requests
+
+from config.services_config import get_translation_config, get_translation_base_url
+
+logger = logging.getLogger(__name__)
+
+
+class HttpTranslationProvider:
+    """Translation via HTTP service."""
+
+    def __init__(
+        self,
+        base_url: str,
+        model: str = "qwen",
+        timeout_sec: float = 10.0,
+        translation_context: Optional[str] = None,
+    ):
+        self.base_url = (base_url or "").rstrip("/")
+        self.model = model or "qwen"
+        self.timeout_sec = float(timeout_sec or 10.0)
+        self.translation_context = translation_context or "e-commerce product search"
+        self.executor = ThreadPoolExecutor(max_workers=4, thread_name_prefix="http-translator")
+
+    def _translate_once(
+        self,
+        text: str,
+        target_lang: str,
+        source_lang: Optional[str] = None,
+    ) -> Optional[str]:
+        if not text or not str(text).strip():
+            return text
+        try:
+            url = f"{self.base_url}/translate"
+            payload = {
+                "text": text,
+                "target_lang": target_lang,
+                "source_lang": source_lang or "auto",
+                "model": self.model,
+            }
+            response = requests.post(url, json=payload, timeout=self.timeout_sec)
+            if response.status_code != 200:
+                logger.warning(
+                    "HTTP translator failed: status=%s body=%s",
+                    response.status_code,
+                    (response.text or "")[:200],
+                )
+                return None
+            data = response.json()
+            translated = data.get("translated_text")
+            return translated if translated is not None else None
+        except Exception as exc:
+            logger.warning("HTTP translator request failed: %s", exc, exc_info=True)
+            return None
+
+    def translate(
+        self,
+        text: str,
+        target_lang: str,
+        source_lang: Optional[str] = None,
+        context: Optional[str] = None,
+        prompt: Optional[str] = None,
+    ) -> Optional[str]:
+        del context, prompt
+        result = self._translate_once(text=text, target_lang=target_lang, source_lang=source_lang)
+        return result if result is not None else text
+
+    def translate_multi(
+        self,
+        text: str,
+        target_langs: List[str],
+        source_lang: Optional[str] = None,
+        context: Optional[str] = None,
+        async_mode: bool = True,
+        prompt: Optional[str] = None,
+    ) -> Dict[str, Optional[str]]:
+        del context, async_mode, prompt
+        out: Dict[str, Optional[str]] = {}
+        for lang in target_langs:
+            out[lang] = self.translate(text, lang, source_lang=source_lang)
+        return out
+
+    def translate_multi_async(
+        self,
+        text: str,
+        target_langs: List[str],
+        source_lang: Optional[str] = None,
+        context: Optional[str] = None,
+        prompt: Optional[str] = None,
+    ) -> Dict[str, Union[str, Future]]:
+        del context, prompt
+        out: Dict[str, Union[str, Future]] = {}
+        for lang in target_langs:
+            out[lang] = self.executor.submit(self.translate, text, lang, source_lang)
+        return out
+
+    def translate_for_indexing(
+        self,
+        text: str,
+        shop_language: str,
+        source_lang: Optional[str] = None,
+        context: Optional[str] = None,
+        prompt: Optional[str] = None,
+        index_languages: Optional[List[str]] = None,
+    ) -> Dict[str, Optional[str]]:
+        del context, prompt
+        langs = index_languages if index_languages else ["en", "zh"]
+        source = source_lang or shop_language or "auto"
+        out: Dict[str, Optional[str]] = {}
+        for lang in langs:
+            if lang == shop_language:
+                out[lang] = text
+            else:
+                out[lang] = self.translate(text, target_lang=lang, source_lang=source)
+        return out
+
+
+def create_translation_provider(query_config: Any = None) -> Any:
+    """
+    Create translation provider from services config.
+
+    query_config: optional, for api_key/glossary_id/context (used by direct provider).
+    """
+    cfg = get_translation_config()
+    provider = cfg.provider
+    pc = cfg.get_provider_cfg()
+
+    if provider in ("direct", "local", "inprocess"):
+        from query.qwen_mt_translate import Translator
+        model = pc.get("model") or "qwen"
+        qc = query_config or _empty_query_config()
+        return Translator(
+            model=model,
+            api_key=getattr(qc, "translation_api_key", None),
+            use_cache=True,
+            glossary_id=getattr(qc, "translation_glossary_id", None),
+            translation_context=getattr(qc, "translation_context", "e-commerce product search"),
+        )
+
+    if provider in ("http", "service"):
+        base_url = get_translation_base_url()
+        model = pc.get("model") or "qwen"
+        timeout = pc.get("timeout_sec", 10.0)
+        qc = query_config or _empty_query_config()
+        return HttpTranslationProvider(
+            base_url=base_url,
+            model=model,
+            timeout_sec=float(timeout),
+            translation_context=getattr(qc, "translation_context", "e-commerce product search"),
+        )
+
+    raise ValueError(f"Unsupported translation provider: {provider}")
+
+
+def _empty_query_config() -> Any:
+    """Minimal object with default translation attrs."""
+    class _QC:
+        translation_api_key = None
+        translation_glossary_id = None
+        translation_context = "e-commerce product search"
+    return _QC()
 """Query package initialization."""
 from .language_detector import LanguageDetector
-from .translator import Translator
+from .qwen_mt_translate import Translator
 from .query_rewriter import QueryRewriter, QueryNormalizer
 from .query_parser import QueryParser, ParsedQuery
@@ -0,0 +1,238 @@
+"""
+LLM-based translation helper using Qwen chat model.
+
+This module provides a thin wrapper around DashScope's `qwen-flash` model
+for high-quality, prompt-controlled translation, independent of the main
+`Translator` (machine translation) pipeline.
+
+Usage example:
+
+    from query.llm_translate import llm_translate
+
+    result = llm_translate(
+        text="我看到这个视频后没有笑",
+        target_lang="en",
+        source_lang="zh",
+        source_lang_label="中文",
+        target_lang_label="英文",
+    )
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+import time
+from typing import Dict, Optional
+
+from openai import OpenAI
+
+from config.env_config import DASHSCOPE_API_KEY
+from config.services_config import get_translation_config
+
+logger = logging.getLogger(__name__)
+
+
+# 华北2（北京）：https://dashscope.aliyuncs.com/compatible-mode/v1
+# 新加坡：https://dashscope-intl.aliyuncs.com/compatible-mode/v1
+# 美国（弗吉尼亚）：https://dashscope-us.aliyuncs.com/compatible-mode/v1
+#
+# 默认保持与现有翻译/索引脚本相同的美国地域，可通过环境变量覆盖：
+#   DASHSCOPE_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1
+DEFAULT_QWEN_BASE_URL = "https://dashscope-us.aliyuncs.com/compatible-mode/v1"
+QWEN_MODEL_NAME = "qwen-flash"
+
+
+# 由调用方提供的语言标签/代码填充，占位符说明：
+# - source_lang: 源语言的人类可读名称（按目标语言本地化，例如 "中文", "English"）
+# - target_lang: 目标语言的人类可读名称
+# - src_lang_code: 源语言代码，例如 "zh"
+# - tgt_lang_code: 目标语言代码，例如 "en"
+TRANSLATION_PROMPTS: Dict[str, str] = {
+    "zh": """你是一名专业的 {source_lang}（{src_lang_code}）到 {target_lang}（{tgt_lang_code}）翻译员。你的目标是在遵循 {target_lang} 的语法、词汇和文化习惯的前提下，准确传达原始 {source_lang} 文本的含义和细微差别。请只输出 {target_lang} 的翻译内容，不要包含任何额外的解释或评论。请将以下 {source_lang} 文本翻译成 {target_lang}：
+
+{text}""",
+    "en": """You are a professional {source_lang} ({src_lang_code}) to {target_lang} ({tgt_lang_code}) translator. Your goal is to accurately convey the meaning and nuances of the original {source_lang} text while adhering to {target_lang} grammar, vocabulary, and cultural sensitivities. Produce only the {target_lang} translation, without any additional explanations or commentary. Please translate the following {source_lang} text into {target_lang}:
+
+{text}""",
+    "ru": """Вы профессиональный переводчик с {source_lang} ({src_lang_code}) на {target_lang} ({tgt_lang_code}). Ваша задача — точно передать смысл и нюансы исходного текста на {source_lang}, соблюдая грамматику, лексику и культурные особенности {target_lang}. Выводите только перевод на {target_lang}, без каких-либо дополнительных объяснений или комментариев. Пожалуйста, переведите следующий текст с {source_lang} на {target_lang}:
+
+{text}""",
+    "ar": """أنت مترجم محترف من {source_lang} ({src_lang_code}) إلى {target_lang} ({tgt_lang_code}). هدفك هو نقل المعنى والدلالات الدقيقة للنص الأصلي بلغة {source_lang} بدقة، مع الالتزام بقواعد اللغة والمفردات والحساسيات الثقافية الخاصة بلغة {target_lang}. قم بإنتاج الترجمة إلى {target_lang} فقط دون أي شروحات أو تعليقات إضافية. يرجى ترجمة النص التالي من {source_lang} إلى {target_lang}:
+
+{text}""",
+    "ja": """あなたは {source_lang}（{src_lang_code}）から {target_lang}（{tgt_lang_code}）へのプロの翻訳者です。{target_lang} の文法、語彙、文化的配慮に従いながら、元の {source_lang} テキストの意味やニュアンスを正確に伝えることが目的です。追加の説明やコメントは一切含めず、{target_lang} の翻訳のみを出力してください。次の {source_lang} テキストを {target_lang} に翻訳してください：
+
+{text}""",
+    "es": """Eres un traductor profesional de {source_lang} ({src_lang_code}) a {target_lang} ({tgt_lang_code}). Tu objetivo es transmitir con precisión el significado y los matices del texto original en {source_lang}, respetando la gramática, el vocabulario y las sensibilidades culturales de {target_lang}. Produce únicamente la traducción en {target_lang}, sin explicaciones ni comentarios adicionales. Por favor, traduce el siguiente texto de {source_lang} a {target_lang}:
+
+{text}""",
+    "de": """Du bist ein professioneller Übersetzer von {source_lang} ({src_lang_code}) nach {target_lang} ({tgt_lang_code}). Dein Ziel ist es, die Bedeutung und Nuancen des ursprünglichen {source_lang}-Textes genau zu vermitteln und dabei die Grammatik, den Wortschatz und die kulturellen Besonderheiten von {target_lang} zu berücksichtigen. Gib ausschließlich die Übersetzung in {target_lang} aus, ohne zusätzliche Erklärungen oder Kommentare. Bitte übersetze den folgenden {source_lang}-Text in {target_lang}:
+
+{text}""",
+    "fr": """Vous êtes un traducteur professionnel de {source_lang} ({src_lang_code}) vers {target_lang} ({tgt_lang_code}). Votre objectif est de transmettre fidèlement le sens et les nuances du texte original en {source_lang}, tout en respectant la grammaire, le vocabulaire et les sensibilités culturelles de {target_lang}. Produisez uniquement la traduction en {target_lang}, sans explications ni commentaires supplémentaires. Veuillez traduire le texte suivant de {source_lang} vers {target_lang} :
+
+{text}""",
+    "it": """Sei un traduttore professionista da {source_lang} ({src_lang_code}) a {target_lang} ({tgt_lang_code}). Il tuo obiettivo è trasmettere con precisione il significato e le sfumature del testo originale in {source_lang}, rispettando la grammatica, il vocabolario e le sensibilità culturali di {target_lang}. Produci solo la traduzione in {target_lang}, senza spiegazioni o commenti aggiuntivi. Per favore traduci il seguente testo da {source_lang} a {target_lang}:
+
+{text}""",
+    "pt": """Você é um tradutor profissional de {source_lang} ({src_lang_code}) para {target_lang} ({tgt_lang_code}). Seu objetivo é transmitir com precisão o significado e as nuances do texto original em {source_lang}, respeitando a gramática, o vocabulário e as sensibilidades culturais de {target_lang}. Produza apenas a tradução em {target_lang}, sem quaisquer explicações ou comentários adicionais. Por favor, traduza o seguinte texto de {source_lang} para {target_lang}:
+
+{text}""",
+}
+
+
+def _get_qwen_client(base_url: Optional[str] = None) -> Optional[OpenAI]:
+    """
+    Lazily construct an OpenAI-compatible client for DashScope.
+
+    Uses DASHSCOPE_API_KEY and base_url (provider config / env) to configure endpoint.
+    """
+    api_key = DASHSCOPE_API_KEY or os.getenv("DASHSCOPE_API_KEY")
+    if not api_key:
+        logger.warning("DASHSCOPE_API_KEY not set; llm-based translation will be disabled")
+        return None
+
+    # 优先使用显式传入的 base_url，其次环境变量，最后默认地域。
+    base_url = (
+        (base_url or "").strip()
+        or os.getenv("DASHSCOPE_BASE_URL")
+        or DEFAULT_QWEN_BASE_URL
+    )
+
+    try:
+        client = OpenAI(api_key=api_key, base_url=base_url)
+        return client
+    except Exception as exc:
+        logger.error("Failed to initialize DashScope OpenAI client: %s", exc, exc_info=True)
+        return None
+
+
+def _build_prompt(
+    text: str,
+    target_lang: str,
+    source_lang_label: str,
+    target_lang_label: str,
+    src_lang_code: str,
+    tgt_lang_code: str,
+) -> str:
+    """
+    Build translation prompt for given target language, defaulting to English template.
+    """
+    key = (target_lang or "").lower()
+    template = TRANSLATION_PROMPTS.get(key) or TRANSLATION_PROMPTS["en"]
+    return template.format(
+        source_lang=source_lang_label,
+        target_lang=target_lang_label,
+        src_lang_code=src_lang_code,
+        tgt_lang_code=tgt_lang_code,
+        text=text,
+    )
+
+
+def llm_translate(
+    text: 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]:
+    """
+    Translate text with Qwen chat model using rich prompts.
+
+    - 根据目标语言选择提示词，如果没匹配到则退回英文模板。
+    - 不对 text 做语言检测或缓存，调用方自行控制。
+
+    Args:
+        text: 原始文本
+        target_lang: 目标语言代码（如 "zh", "en"）
+        source_lang: 源语言代码（可选，不影响提示词选择，仅用于日志）
+        source_lang_label: 源语言展示名称，用于 prompt（默认使用 source_lang）
+        target_lang_label: 目标语言展示名称，用于 prompt（默认使用 target_lang）
+        timeout_sec: 请求超时时间（秒，可选；若未配置则从 config 读取或采用默认）
+
+    Returns:
+        翻译后的文本；如失败则返回 None。
+    """
+    if not text or not str(text).strip():
+        return text
+
+    cfg = get_translation_config()
+    provider_cfg = cfg.providers.get("llm", {}) if isinstance(cfg.providers, dict) else {}
+
+    model_name = provider_cfg.get("model") or QWEN_MODEL_NAME
+    req_timeout = float(provider_cfg.get("timeout_sec") or timeout_sec or 30.0)
+    base_url = (provider_cfg.get("base_url") or "").strip() or None
+
+    client = _get_qwen_client(base_url=base_url)
+    if not client:
+        # 无法调用云端，直接回退
+        logger.warning(
+            "[llm_translate] Client init failed; returning original text. "
+            "text=%r target_lang=%s source_lang=%s",
+            text[:80],
+            target_lang,
+            source_lang or "auto",
+        )
+        return text
+
+    tgt = (target_lang or "").lower() or "en"
+    src = (source_lang or "auto").lower()
+    src_label = source_lang_label or src
+    tgt_label = target_lang_label or tgt
+
+    prompt = _build_prompt(
+        text=text,
+        target_lang=tgt,
+        source_lang_label=src_label,
+        target_lang_label=tgt_label,
+        src_lang_code=src,
+        tgt_lang_code=tgt,
+    )
+
+    start = time.time()
+    try:
+        completion = client.chat.completions.create(
+            model=model_name,
+            messages=[
+                {
+                    "role": "user",
+                    "content": prompt,
+                }
+            ],
+            timeout=req_timeout,
+        )
+        content = (completion.choices[0].message.content or "").strip()
+        duration_ms = (time.time() - start) * 1000
+        logger.info(
+            "[llm_translate] Success | model=%s src=%s tgt=%s latency=%.1fms text=%r -> %r",
+            model_name,
+            src,
+            tgt,
+            duration_ms,
+            text[:80],
+            content[:80],
+        )
+        return content or text
+    except Exception as exc:
+        duration_ms = (time.time() - start) * 1000
+        logger.warning(
+            "[llm_translate] Failed | model=%s src=%s tgt=%s latency=%.1fms error=%s",
+            model_name,
+            src,
+            tgt,
+            duration_ms,
+            exc,
+            exc_info=True,
+        )
+        # 安全回退：出错时返回原文，避免中断上游流程
+        return text
+
+
+__all__ = [
+    "TRANSLATION_PROMPTS",
+    "llm_translate",
+]
+
@@ -0,0 +1,963 @@
+"""
+Translation service for multi-language query support.
+
+Supports multiple translation models:
+- Qwen (default): Alibaba Cloud DashScope API using qwen-mt-flash model
+- DeepL: DeepL API for high-quality translations
+
+重要说明（Qwen 机翻限速）:
+- 当前默认使用的 `qwen-mt-flash` 为云端机翻模型，**官方限速较低，约 RPM=60（每分钟约 60 请求）**
+- 在高并发场景必须依赖 Redis 翻译缓存与批量预热，避免在用户实时请求路径上直接打满 DashScope 限流
+- 若业务侧存在大规模离线翻译或更高吞吐需求，建议评估 DeepL 或自建翻译后端
+
+使用方法 (Usage):
+
+```python
+from query.translator import Translator
+
+# 使用默认的 qwen 模型（推荐）
+translator = Translator()  # 默认使用 qwen 模型
+
+# 或显式指定模型
+translator = Translator(model='qwen')  # 使用 qwen 模型
+translator = Translator(model='deepl')  # 使用 DeepL 模型
+
+# 翻译文本
+result = translator.translate(
+    text="我看到这个视频后没有笑",
+    target_lang="en",
+    source_lang="auto"  # 自动检测源语言
+)
+```
+
+配置说明 (Configuration):
+- Qwen 模型需要设置 DASHSCOPE_API_KEY 环境变量（在 .env 文件中）
+- DeepL 模型需要设置 DEEPL_AUTH_KEY 环境变量（在 .env 文件中）
+
+Qwen 模型参考文档：
+- 官方文档：https://help.aliyun.com/zh/model-studio/get-api-key
+- 模型：qwen-mt-flash（快速翻译模型）
+
+DeepL 官方文档：
+https://developers.deepl.com/api-reference/translate/request-translation
+"""
+
+import os
+import requests
+import re
+import redis
+from concurrent.futures import ThreadPoolExecutor, Future
+from datetime import timedelta
+from typing import Dict, List, Optional, Union
+import logging
+import time
+
+logger = logging.getLogger(__name__)
+
+from config.env_config import DEEPL_AUTH_KEY, DASHSCOPE_API_KEY, REDIS_CONFIG
+from openai import OpenAI
+
+
+class Translator:
+    """
+    Multi-language translator supporting Qwen and DeepL APIs.
+    
+    Default model is 'qwen' which uses Alibaba Cloud DashScope API.
+    """
+# 华北2（北京）：https://dashscope.aliyuncs.com/compatible-mode/v1
+# 新加坡：https://dashscope-intl.aliyuncs.com/compatible-mode/v1
+# 美国（弗吉尼亚）：https://dashscope-us.aliyuncs.com/compatible-mode/v1
+
+    DEEPL_API_URL = "https://api.deepl.com/v2/translate"  # Pro tier
+    QWEN_BASE_URL = "https://dashscope-us.aliyuncs.com/compatible-mode/v1"  # 北京地域
+    # QWEN_BASE_URL = "https://dashscope-intl.aliyuncs.com/compatible-mode/v1"  # 新加坡
+    # 如果使用新加坡地域的模型，需要将base_url替换为：https://dashscope-intl.aliyuncs.com/compatible-mode/v1
+    QWEN_MODEL = "qwen-mt-flash"  # 快速翻译模型
+
+    # Language code mapping
+    LANG_CODE_MAP = {
+        'zh': 'ZH',
+        'en': 'EN',
+        'ru': 'RU',
+        'ar': 'AR',
+        'ja': 'JA',
+        'es': 'ES',
+        'de': 'DE',
+        'fr': 'FR',
+        'it': 'IT',
+        'pt': 'PT',
+    }
+
+    def __init__(
+        self,
+        model: str = "qwen",
+        api_key: Optional[str] = None,
+        use_cache: bool = True,
+        timeout: int = 10,
+        glossary_id: Optional[str] = None,
+        translation_context: Optional[str] = None
+    ):
+        """
+        Initialize translator.
+
+        Args:
+            model: Translation model to use. Options: 'qwen' (default) or 'deepl'
+            api_key: API key for the selected model (or None to use from config/env)
+            use_cache: Whether to cache translations
+            timeout: Request timeout in seconds
+            glossary_id: DeepL glossary ID for custom terminology (optional, only for DeepL)
+            translation_context: Context hint for translation (e.g., "e-commerce", "product search")
+        """
+        self.model = model.lower()
+        if self.model not in ['qwen', 'deepl']:
+            raise ValueError(f"Unsupported model: {model}. Supported models: 'qwen', 'deepl'")
+        
+        # Get API key from config if not provided
+        if api_key is None:
+            if self.model == 'qwen':
+                api_key = DASHSCOPE_API_KEY or os.getenv("DASHSCOPE_API_KEY")
+            else:  # deepl
+                api_key = DEEPL_AUTH_KEY or os.getenv("DEEPL_AUTH_KEY")
+
+        self.api_key = api_key
+        self.timeout = timeout
+        self.use_cache = use_cache
+        self.glossary_id = glossary_id
+        self.translation_context = translation_context or "e-commerce product search"
+        
+        # Initialize OpenAI client for Qwen if needed
+        self.qwen_client = None
+        if self.model == 'qwen':
+            if not self.api_key:
+                logger.warning("DASHSCOPE_API_KEY not set. Qwen translation will not work.")
+            else:
+                self.qwen_client = OpenAI(
+                    api_key=self.api_key,
+                    base_url=self.QWEN_BASE_URL,
+                )
+
+        # Initialize Redis cache if enabled
+        if use_cache:
+            try:
+                self.redis_client = redis.Redis(
+                    host=REDIS_CONFIG.get('host', 'localhost'),
+                    port=REDIS_CONFIG.get('port', 6479),
+                    password=REDIS_CONFIG.get('password'),
+                    decode_responses=True,  # Return str instead of bytes
+                    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,  # 避免复用坏连接
+                )
+                # Test connection
+                self.redis_client.ping()
+                expire_days = REDIS_CONFIG.get('translation_cache_expire_days', 360)
+                self.expire_time = timedelta(days=expire_days)
+                self.expire_seconds = int(self.expire_time.total_seconds())  # Redis 需要秒数
+                self.cache_prefix = REDIS_CONFIG.get('translation_cache_prefix', 'trans')
+                logger.info("Redis cache initialized for translations")
+            except Exception as e:
+                logger.warning(f"Failed to initialize Redis cache: {e}, falling back to no cache")
+                self.redis_client = None
+                self.cache = None
+        else:
+            self.redis_client = None
+            self.cache = None
+        
+        # Thread pool for async translation
+        self.executor = ThreadPoolExecutor(max_workers=2, thread_name_prefix="translator")
+
+    def translate(
+        self,
+        text: str,
+        target_lang: str,
+        source_lang: Optional[str] = None,
+        context: Optional[str] = None,
+        prompt: Optional[str] = None
+    ) -> Optional[str]:
+        """
+        Translate text to target language (synchronous mode).
+
+        Args:
+            text: Text to translate
+            target_lang: Target language code ('zh', 'en', 'ru', etc.)
+            source_lang: Source language code (option al, auto-detect if None)
+            context: Additional context for translation (overrides default context)
+            prompt: Translation prompt/instruction (optional, for better translation quality)
+
+        Returns:
+            Translated text or None if translation fails
+        """
+        if not text or not text.strip():
+            return text
+
+        # Normalize language codes
+        target_lang = target_lang.lower()
+        if source_lang:
+            source_lang = source_lang.lower()
+
+        # Optimization: Skip translation if not needed
+        if target_lang == 'en' and self._is_english_text(text):
+            logger.info(f"[Translator] Text is already English, skipping translation: '{text[:50]}...'")
+            return text
+        
+        if target_lang == 'zh' and (self._contains_chinese(text) or self._is_pure_number(text)):
+            logger.info(
+                f"[Translator] Translation request | Original text: '{text}' | Target language: {target_lang} | "
+                f"Source language: {source_lang or 'auto'} | Result: Skip translation (contains Chinese or pure number)"
+            )
+            return text
+
+        # Use provided context or default context
+        translation_context = context or self.translation_context
+        
+        # Build cache key (include prompt in cache key if provided)
+        cache_key_parts = [source_lang or 'auto', target_lang, translation_context]
+        if prompt:
+            cache_key_parts.append(prompt)
+        cache_key_parts.append(text)
+        cache_key = ':'.join(cache_key_parts)
+
+        # Check cache (include context and prompt in cache key for accuracy)
+        if self.use_cache and self.redis_client:
+            cached = self._get_cached_translation_redis(text, target_lang, source_lang, translation_context, prompt)
+            if cached:
+                logger.info(
+                    f"[Translator] Translation request | Original text: '{text}' | Target language: {target_lang} | "
+                    f"Source language: {source_lang or 'auto'} | Result: '{cached}' | Source: Cache hit"
+                )
+                return cached
+
+        # If no API key, return mock translation (for testing)
+        if not self.api_key:
+            logger.info(
+                f"[Translator] Translation request | Original text: '{text}' | Target language: {target_lang} | "
+                f"Source language: {source_lang or 'auto'} | Result: '{text}' | Source: Mock mode (no API key)"
+            )
+            return text
+
+        # Translate using selected model
+        logger.info(
+            f"[Translator] Translation request | Model: {self.model} | Original text: '{text}' | Target language: {target_lang} | "
+            f"Source language: {source_lang or 'auto'} | Context: {translation_context} | "
+            f"Prompt: {'yes' if prompt else 'no'} | Status: Starting translation"
+        )
+        
+        if self.model == 'qwen':
+            result = self._translate_qwen(text, target_lang, source_lang, translation_context, prompt)
+        else:  # deepl
+            result = self._translate_deepl(text, target_lang, source_lang, translation_context, prompt)
+
+        # Surface translation failure to the caller instead of silently
+        # masquerading the source text as a successful translation.
+        if result is None:
+            logger.warning(
+                f"[Translator] Translation request | Original text: '{text}' | Target language: {target_lang} | "
+                f"Source language: {source_lang or 'auto'} | Status: Translation failed"
+            )
+        else:
+            logger.info(
+                f"[Translator] Translation request | Original text: '{text}' | Target language: {target_lang} | "
+                f"Source language: {source_lang or 'auto'} | Result: '{result}' | Status: Translation successful"
+            )
+
+        # Cache only successful translations. Failed attempts must not poison
+        # Redis with the original text.
+        if result is not None and self.use_cache and self.redis_client:
+            self._set_cached_translation_redis(text, target_lang, result, source_lang, translation_context, prompt)
+
+        return result
+
+    def _translate_qwen(
+        self,
+        text: str,
+        target_lang: str,
+        source_lang: Optional[str],
+        context: Optional[str] = None,
+        prompt: Optional[str] = None
+    ) -> Optional[str]:
+        """
+        Translate using Qwen MT Flash model via Alibaba Cloud DashScope API.
+        
+        Args:
+            text: Text to translate
+            target_lang: Target language code ('zh', 'en', 'ru', etc.)
+            source_lang: Source language code (optional, 'auto' if None)
+            context: Context hint for translation (optional)
+            prompt: Translation prompt/instruction (optional)
+            
+        Returns:
+            Translated text or None if translation fails
+        """
+        if not self.qwen_client:
+            logger.error("[Translator] Qwen client not initialized. Check DASHSCOPE_API_KEY.")
+            return None
+        
+        # Qwen (qwen-mt-plus/flash/turbo) supported languages mapping
+        # 标准来自：你提供的“语言 / 英文名 / 代码”表
+        qwen_lang_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",
+        }
+        
+        # Convert target language
+        target_lang_normalized = target_lang.lower()
+        target_lang_qwen = qwen_lang_map.get(target_lang_normalized, target_lang.capitalize())
+
+        # Convert source language
+        source_lang_normalized = (source_lang or "").strip().lower()
+        if not source_lang_normalized or source_lang_normalized == "auto":
+            source_lang_qwen = "auto"
+        else:
+            source_lang_qwen = qwen_lang_map.get(source_lang_normalized, source_lang.capitalize())
+        
+        # Prepare translation options
+        translation_options = {
+            "source_lang": source_lang_qwen,
+            "target_lang": target_lang_qwen,
+        }
+        
+        # Prepare messages
+        messages = [
+            {
+                "role": "user",
+                "content": text
+            }
+        ]
+        
+        start_time = time.time()
+        try:
+            completion = self.qwen_client.chat.completions.create(
+                model=self.QWEN_MODEL,
+                messages=messages,
+                extra_body={
+                    "translation_options": translation_options
+                }
+            )
+            
+            translated_text = completion.choices[0].message.content.strip()
+            duration_ms = (time.time() - start_time) * 1000
+            
+            logger.info(
+                f"[Translator] Qwen API response success | Original text: '{text}' | Target language: {target_lang_qwen} | "
+                f"Translation result: '{translated_text}' | Duration: {duration_ms:.2f} ms"
+            )
+            return translated_text
+            
+        except Exception as e:
+            duration_ms = (time.time() - start_time) * 1000
+            logger.error(
+                f"[Translator] Qwen API request exception | Original text: '{text}' | Target language: {target_lang_qwen} | "
+                f"Duration: {duration_ms:.2f} ms | Error: {e}", exc_info=True
+            )
+            return None
+
+    def _translate_deepl(
+        self,
+        text: str,
+        target_lang: str,
+        source_lang: Optional[str],
+        context: Optional[str] = None,
+        prompt: Optional[str] = None
+    ) -> Optional[str]:
+        """
+        Translate using DeepL API with context and glossary support.
+        
+        Args:
+            text: Text to translate
+            target_lang: Target language code
+            source_lang: Source language code (optional)
+            context: Context hint for translation (e.g., "e-commerce product search")
+        """
+        # Map to DeepL language codes
+        target_code = self.LANG_CODE_MAP.get(target_lang, target_lang.upper())
+
+        headers = {
+            "Authorization": f"DeepL-Auth-Key {self.api_key}",
+            "Content-Type": "application/json",
+        }
+
+        # Use prompt as context parameter for DeepL API (not as text prefix)
+        # According to DeepL API: context is "Additional context that can influence a translation but is not translated itself"
+        # If prompt is provided, use it as context; otherwise use the default context
+        api_context = prompt if prompt else context
+        
+        # For e-commerce, add context words to help DeepL understand the domain
+        # This is especially important for single-word ambiguous terms like "车" (car vs rook)
+        text_to_translate, needs_extraction = self._add_ecommerce_context(text, source_lang, api_context)
+
+        payload = {
+            "text": [text_to_translate],
+            "target_lang": target_code,
+        }
+
+        if source_lang:
+            source_code = self.LANG_CODE_MAP.get(source_lang, source_lang.upper())
+            payload["source_lang"] = source_code
+
+        # Add context parameter (prompt or default context)
+        # Context influences translation but is not translated itself
+        if api_context:
+            payload["context"] = api_context
+
+        # Add glossary if configured
+        if self.glossary_id:
+            payload["glossary_id"] = self.glossary_id
+
+        # Note: DeepL API v2 supports "context" parameter for additional context
+        # that influences translation but is not translated itself.
+        # We use prompt as context parameter when provided.
+
+        try:
+            response = requests.post(
+                self.DEEPL_API_URL,
+                headers=headers,
+                json=payload,
+                timeout=self.timeout
+            )
+
+            if response.status_code == 200:
+                data = response.json()
+                if "translations" in data and len(data["translations"]) > 0:
+                    translated_text = data["translations"][0]["text"]
+                    # If we added context, extract just the term from the result
+                    if needs_extraction:
+                        translated_text = self._extract_term_from_translation(
+                            translated_text, text, target_code
+                        )
+                    logger.debug(
+                        f"[Translator] DeepL API response success | Original text: '{text}' | Target language: {target_code} | "
+                        f"Translation result: '{translated_text}'"
+                    )
+                    return translated_text
+            else:
+                logger.error(
+                    f"[Translator] DeepL API error | Original text: '{text}' | Target language: {target_code} | "
+                    f"Status code: {response.status_code} | Error message: {response.text}"
+                )
+                return None
+
+        except requests.Timeout:
+            logger.warning(
+                f"[Translator] DeepL API request timeout | Original text: '{text}' | Target language: {target_code} | "
+                f"Timeout: {self.timeout}s"
+            )
+            return None
+        except Exception as e:
+            logger.error(
+                f"[Translator] DeepL API request exception | Original text: '{text}' | Target language: {target_code} | "
+                f"Error: {e}", exc_info=True
+            )
+            return None
+
+    # NOTE: _translate_deepl_free is intentionally not implemented.
+    # We do not support automatic fallback to the free endpoint, to avoid
+    # mixing Pro keys with https://api-free.deepl.com and related 403 errors.
+
+    def translate_multi(
+        self,
+        text: str,
+        target_langs: List[str],
+        source_lang: Optional[str] = None,
+        context: Optional[str] = None,
+        async_mode: bool = True,
+        prompt: Optional[str] = None
+    ) -> Dict[str, Optional[str]]:
+        """
+        Translate text to multiple target languages.
+        
+        In async_mode=True (default):
+        - Returns cached translations immediately if available
+        - For translations that can be optimized (e.g., pure numbers, already in target language),
+          returns result immediately via synchronous call
+        - Launches async tasks for other missing translations (non-blocking)
+        - Returns None for missing translations that require async processing
+        
+        In async_mode=False:
+        - Waits for all translations to complete (blocking)
+
+        Args:
+            text: Text to translate
+            target_langs: List of target language codes
+            source_lang: Source language code (optional)
+            context: Context hint for translation (optional)
+            async_mode: If True, return cached results immediately and translate missing ones async
+            prompt: Translation prompt/instruction (optional)
+
+        Returns:
+            Dictionary mapping language code to translated text (only cached results in async mode)
+        """
+        results = {}
+        missing_langs = []
+        async_langs = []
+        
+        # First, get cached translations
+        for lang in target_langs:
+            cached = self._get_cached_translation(text, lang, source_lang, context, prompt)
+            if cached is not None:
+                results[lang] = cached
+            else:
+                missing_langs.append(lang)
+        
+        # If async mode and there are missing translations
+        if async_mode and missing_langs:
+            # Check if translation can be optimized (immediate return)
+            for lang in missing_langs:
+                target_lang = lang.lower()
+                # Check optimization conditions (same as in translate method)
+                can_optimize = False
+                if target_lang == 'en' and self._is_english_text(text):
+                    can_optimize = True
+                elif target_lang == 'zh' and (self._contains_chinese(text) or self._is_pure_number(text)):
+                    can_optimize = True
+                
+                if can_optimize:
+                    # Can be optimized, call translate synchronously for immediate result
+                    results[lang] = self.translate(text, lang, source_lang, context, prompt)
+                else:
+                    # Requires actual translation, add to async list
+                    async_langs.append(lang)
+            
+            # Launch async tasks for translations that require actual API calls
+            if async_langs:
+                for lang in async_langs:
+                    self._translate_async(text, lang, source_lang, context, prompt)
+                # Return None for async translations
+                for lang in async_langs:
+                    results[lang] = None
+        else:
+            # Synchronous mode: wait for all translations
+            for lang in missing_langs:
+                results[lang] = self.translate(text, lang, source_lang, context, prompt)
+        
+        return results
+    
+    def translate_multi_async(
+        self,
+        text: str,
+        target_langs: List[str],
+        source_lang: Optional[str] = None,
+        context: Optional[str] = None,
+        prompt: Optional[str] = None
+    ) -> Dict[str, Union[str, Future]]:
+        """
+        Translate text to multiple target languages asynchronously, returning Futures that can be awaited.
+        
+        This method returns a dictionary where:
+        - If translation is cached, the value is the translation string (immediate)
+        - If translation needs to be done, the value is a Future object that can be awaited
+        
+        Args:
+            text: Text to translate
+            target_langs: List of target language codes
+            source_lang: Source language code (optional)
+            context: Context hint for translation (optional)
+            prompt: Translation prompt/instruction (optional)
+
+        Returns:
+            Dictionary mapping language code to either translation string (cached) or Future object
+        """
+        results = {}
+        missing_langs = []
+        
+        # First, get cached translations
+        for lang in target_langs:
+            cached = self._get_cached_translation(text, lang, source_lang, context, prompt)
+            if cached is not None:
+                results[lang] = cached
+            else:
+                missing_langs.append(lang)
+        
+        # For missing translations, submit async tasks and return Futures
+        for lang in missing_langs:
+            future = self.executor.submit(
+                self.translate,
+                text,
+                lang,
+                source_lang,
+                context,
+                prompt
+            )
+            results[lang] = future
+        
+        return results
+    
+    def _get_cached_translation(
+        self,
+        text: str,
+        target_lang: str,
+        source_lang: Optional[str] = None,
+        context: Optional[str] = None,
+        prompt: Optional[str] = None
+    ) -> Optional[str]:
+        """Get translation from cache if available."""
+        if not self.redis_client:
+            return None
+        return self._get_cached_translation_redis(text, target_lang, source_lang, context, prompt)
+    
+    def _get_cached_translation_redis(
+        self,
+        text: str,
+        target_lang: str,
+        source_lang: Optional[str] = None,
+        context: Optional[str] = None,
+        prompt: Optional[str] = None
+    ) -> Optional[str]:
+        """
+        Get translation from Redis cache with sliding expiration.
+        
+        滑动过期机制：每次访问缓存时，重置过期时间为配置的过期时间（默认720天）。
+        这样缓存会在最后一次访问后的720天才过期，而不是写入后的720天。
+        这确保了常用的翻译缓存不会被过早删除。
+        """
+        if not self.redis_client:
+            return None
+        
+        try:
+            # Build cache key: prefix:target_lang:text
+            # For simplicity, we use target_lang and text as key
+            # Context and prompt are not included in key to maximize cache hits
+            cache_key = f"{self.cache_prefix}:{target_lang.upper()}:{text}"
+            value = self.redis_client.get(cache_key)
+            if value:
+                # Sliding expiration: reset expiration time on access
+                # 每次读取缓存时，重置过期时间为配置的过期时间（最后一次访问后的N天才过期）
+                try:
+                    self.redis_client.expire(cache_key, self.expire_seconds)
+                except Exception as expire_error:
+                    # 即使 expire 失败，也返回缓存值（不影响功能）
+                    logger.warning(
+                        f"[Translator] Failed to update cache expiration for key {cache_key}: {expire_error}"
+                    )
+                
+                logger.debug(
+                    f"[Translator] Redis cache hit | Original text: '{text}' | Target language: {target_lang} | "
+                    f"Cache key: {cache_key} | Translation result: '{value}' | TTL reset to {self.expire_seconds}s"
+                )
+                return value
+            logger.debug(
+                f"[Translator] Redis cache miss | Original text: '{text}' | Target language: {target_lang} | "
+                f"Cache key: {cache_key}"
+            )
+            return None
+        except Exception as e:
+            logger.error(f"[Translator] Redis error during get translation cache | Original text: '{text}' | Target language: {target_lang} | Error: {e}")
+            return None
+    
+    def _set_cached_translation_redis(
+        self,
+        text: str,
+        target_lang: str,
+        translation: str,
+        source_lang: Optional[str] = None,
+        context: Optional[str] = None,
+        prompt: Optional[str] = None
+    ) -> None:
+        """Store translation in Redis cache."""
+        if not self.redis_client:
+            return
+        
+        try:
+            cache_key = f"{self.cache_prefix}:{target_lang.upper()}:{text}"
+            self.redis_client.setex(cache_key, self.expire_seconds, translation)
+            logger.info(
+                f"[Translator] Redis cache write | Original text: '{text}' | Target language: {target_lang} | "
+                f"Cache key: {cache_key} | Translation result: '{translation}'"
+            )
+        except Exception as e:
+            logger.error(
+                f"[Translator] Redis cache write failed | Original text: '{text}' | Target language: {target_lang} | "
+                f"Error: {e}"
+            )
+    
+    def _translate_async(
+        self,
+        text: str,
+        target_lang: str,
+        source_lang: Optional[str] = None,
+        context: Optional[str] = None,
+        prompt: Optional[str] = None
+    ):
+        """Launch async translation task."""
+        def _do_translate():
+            try:
+                result = self.translate(text, target_lang, source_lang, context, prompt)
+                if result:
+                    logger.debug(f"Async translation completed: {text} -> {target_lang}: {result}")
+            except Exception as e:
+                logger.warning(f"Async translation failed: {text} -> {target_lang}: {e}")
+        
+        self.executor.submit(_do_translate)
+
+    def _add_ecommerce_context(
+        self,
+        text: str,
+        source_lang: Optional[str],
+        context: Optional[str]
+    ) -> tuple:
+        """
+        Add e-commerce context to text for better disambiguation.
+        
+        For single-word ambiguous Chinese terms, we add context words that help
+        DeepL understand this is an e-commerce/product search context.
+        
+        Args:
+            text: Original text to translate
+            source_lang: Source language code
+            context: Context hint
+            
+        Returns:
+            Tuple of (text_with_context, needs_extraction)
+            - text_with_context: Text to send to DeepL
+            - needs_extraction: Whether we need to extract the term from the result
+        """
+        # Only apply for e-commerce context and Chinese source
+        if not context or "e-commerce" not in context.lower():
+            return text, False
+            
+        if not source_lang or source_lang.lower() != 'zh':
+            return text, False
+            
+        # For single-word queries, add context to help disambiguation
+        text_stripped = text.strip()
+        if len(text_stripped.split()) == 1 and len(text_stripped) <= 2:
+            # Common ambiguous Chinese e-commerce terms like "车" (car vs rook)
+            # We add a context phrase: "购买 [term]" (buy [term]) or "商品 [term]" (product [term])
+            # This helps DeepL understand the e-commerce context
+            # We'll need to extract just the term from the translation result
+            context_phrase = f"购买 {text_stripped}"
+            return context_phrase, True
+        
+        # For multi-word queries, DeepL usually has enough context
+        return text, False
+
+    def _extract_term_from_translation(
+        self,
+        translated_text: str,
+        original_text: str,
+        target_lang_code: str
+    ) -> str:
+        """
+        Extract the actual term from a translation that included context.
+        
+        For example, if we translated "购买 车" (buy car) and got "buy car",
+        we want to extract just "car".
+        
+        Args:
+            translated_text: Full translation result
+            original_text: Original single-word query
+            target_lang_code: Target language code (EN, ZH, etc.)
+            
+        Returns:
+            Extracted term or original translation if extraction fails
+        """
+        # For English target, try to extract the last word (the actual term)
+        if target_lang_code == "EN":
+            words = translated_text.strip().split()
+            if len(words) > 1:
+                # Usually the last word is the term we want
+                # But we need to be smart - if it's "buy car", we want "car"
+                # Common context words to skip: buy, purchase, product, item, etc.
+                context_words = {"buy", "purchase", "product", "item", "commodity", "goods"}
+                # Try to find the term (not a context word)
+                for word in reversed(words):
+                    word_lower = word.lower().rstrip('.,!?;:')
+                    if word_lower not in context_words:
+                        return word_lower
+                # If all words are context words, return the last one
+                return words[-1].lower().rstrip('.,!?;:')
+        
+        # For other languages or if extraction fails, return as-is
+        # The user can configure a glossary for better results
+        return translated_text
+
+    def _shop_lang_matches(self, shop_lang_lower: str, lang_code: str) -> bool:
+        """True if shop language matches index language (use source, no translate)."""
+        if not shop_lang_lower or not lang_code:
+            return False
+        if shop_lang_lower == lang_code:
+            return True
+        if lang_code == "zh" and "zh" in shop_lang_lower:
+            return True
+        if lang_code == "en" and "en" in shop_lang_lower:
+            return True
+        return False
+
+    def translate_for_indexing(
+        self,
+        text: str,
+        shop_language: str,
+        source_lang: Optional[str] = None,
+        context: Optional[str] = None,
+        prompt: Optional[str] = None,
+        index_languages: Optional[List[str]] = None,
+    ) -> Dict[str, Optional[str]]:
+        """
+        Translate text for indexing based on shop language and tenant index_languages.
+
+        For each language in index_languages: use source text if shop language matches,
+        otherwise translate to that language.
+
+        Args:
+            text: Text to translate
+            shop_language: Shop primary language (e.g. 'zh', 'en', 'ru')
+            source_lang: Source language code (optional)
+            context: Additional context for translation (optional)
+            prompt: Translation prompt (optional)
+            index_languages: Languages to index (from tenant_config). Default ["en", "zh"].
+
+        Returns:
+            Dict keyed by each index_language with translated or source text (or None).
+        """
+        langs = index_languages if index_languages else ["en", "zh"]
+        results = {lang: None for lang in langs}
+        if not text or not text.strip():
+            return results
+        if re.match(r'^[\d\s_-]+$', text):
+            logger.info(f"[Translator] Skip translation for symbol-only query: '{text}'")
+            return results
+
+        shop_lang_lower = (shop_language or "").strip().lower()
+        targets = []
+        for lang in langs:
+            if self._shop_lang_matches(shop_lang_lower, lang):
+                results[lang] = text
+            else:
+                targets.append(lang)
+
+        for target_lang in targets:
+            cached = self._get_cached_translation_redis(text, target_lang, source_lang, context, prompt)
+            if cached:
+                results[target_lang] = cached
+                logger.debug(f"[Translator] Cache hit for indexing: '{text}' -> {target_lang}: {cached}")
+                continue
+            translated = self.translate(
+                text,
+                target_lang=target_lang,
+                source_lang=source_lang or shop_language,
+                context=context,
+                prompt=prompt,
+            )
+            results[target_lang] = translated
+        return results
+
+    def get_translation_needs(
+        self,
+        detected_lang: str,
+        supported_langs: List[str]
+    ) -> List[str]:
+        """
+        Determine which languages need translation.
+
+        Args:
+            detected_lang: Detected query language
+            supported_langs: List of supported languages
+
+        Returns:
+            List of language codes to translate to
+        """
+        # If detected language is in supported list, translate to others
+        if detected_lang in supported_langs:
+            return [lang for lang in supported_langs if detected_lang != lang]
+
+        # Otherwise, translate to all supported languages
+        return supported_langs
+    
+    def _is_english_text(self, text: str) -> bool:
+        """
+        Check if text is primarily English (ASCII letters, numbers, common punctuation).
+        
+        Args:
+            text: Text to check
+            
+        Returns:
+            True if text appears to be English
+        """
+        if not text or not text.strip():
+            return True
+        
+        # Remove whitespace and common punctuation
+        text_clean = re.sub(r'[\s\.,!?;:\-\'\"\(\)\[\]{}]', '', text)
+        if not text_clean:
+            return True
+        
+        # Check if all remaining characters are ASCII (letters, numbers)
+        # This is a simple heuristic: if most characters are ASCII, it's likely English
+        ascii_count = sum(1 for c in text_clean if ord(c) < 128)
+        ratio = ascii_count / len(text_clean) if text_clean else 0
+        
+        # If more than 80% are ASCII characters, consider it English
+        return ratio > 0.8
+    
+    def _contains_chinese(self, text: str) -> bool:
+        """
+        Check if text contains Chinese characters (Han characters).
+        
+        Args:
+            text: Text to check
+            
+        Returns:
+            True if text contains Chinese characters
+        """
+        if not text:
+            return False
+        
+        # Check for Chinese characters (Unicode range: \u4e00-\u9fff)
+        chinese_pattern = re.compile(r'[\u4e00-\u9fff]')
+        return bool(chinese_pattern.search(text))
+    
+    def _is_pure_number(self, text: str) -> bool:
+        """
+        Check if text is purely numeric (digits, possibly with spaces, dots, commas).
+        
+        Args:
+            text: Text to check
+            
+        Returns:
+            True if text is purely numeric
+        """
+        if not text or not text.strip():
+            return False
+        
+        # Remove whitespace, dots, commas (common number separators)
+        text_clean = re.sub(r'[\s\.,]', '', text.strip())
+        if not text_clean:
+            return False
+        
+        # Check if all remaining characters are digits
+        return text_clean.isdigit()
@@ -19,7 +19,7 @@ from pathlib import Path
 sys.path.insert(0, str(Path(__file__).parent.parent))
 from config import ConfigLoader
-from query.translator import Translator
+from query.qwen_mt_translate import Translator
 import logging
 # Configure logging
@@ -13,7 +13,7 @@ Supports multiple translation models:
 使用方法 (Usage):
 ```python
-from query.translator import Translator
+from query.qwen_mt_translate import Translator
 # 使用默认的 qwen 模型（推荐）
 translator = Translator()  # 默认使用 qwen 模型
-from query.translator import Translator
+from query.qwen_mt_translate import Translator
 class _RecordingRedis: