Blame view

query/translator.py 13.7 KB
be52af70   tangwang   first commit
1
2
3
4
5
6
7
8
9
10
  """
  Translation service for multi-language query support.
  
  Supports DeepL API for high-quality translations.
  """
  
  import requests
  from typing import Dict, List, Optional
  from utils.cache import DictCache
  
325eec03   tangwang   1. 日志、配置基础设施,使用优化
11
12
13
14
15
16
  # Try to import DEEPL_AUTH_KEY, but allow import to fail
  try:
      from config.env_config import DEEPL_AUTH_KEY
  except ImportError:
      DEEPL_AUTH_KEY = None
  
be52af70   tangwang   first commit
17
18
19
20
  
  class Translator:
      """Multi-language translator using DeepL API."""
  
16c42787   tangwang   feat: implement r...
21
      DEEPL_API_URL = "https://api.deepl.com/v2/translate"  # Pro tier
be52af70   tangwang   first commit
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
  
      # 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,
          api_key: Optional[str] = None,
          use_cache: bool = True,
522a3964   tangwang   多语言搜索翻译的优化(deepL添...
41
42
43
          timeout: int = 10,
          glossary_id: Optional[str] = None,
          translation_context: Optional[str] = None
be52af70   tangwang   first commit
44
45
46
47
48
      ):
          """
          Initialize translator.
  
          Args:
d79810d5   tangwang   first commit
49
              api_key: DeepL API key (or None to use from config/env)
be52af70   tangwang   first commit
50
51
              use_cache: Whether to cache translations
              timeout: Request timeout in seconds
522a3964   tangwang   多语言搜索翻译的优化(deepL添...
52
53
              glossary_id: DeepL glossary ID for custom terminology (optional)
              translation_context: Context hint for translation (e.g., "e-commerce", "product search")
be52af70   tangwang   first commit
54
          """
d79810d5   tangwang   first commit
55
          # Get API key from config if not provided
325eec03   tangwang   1. 日志、配置基础设施,使用优化
56
57
          if api_key is None and DEEPL_AUTH_KEY:
              api_key = DEEPL_AUTH_KEY
d79810d5   tangwang   first commit
58
  
be52af70   tangwang   first commit
59
60
61
          self.api_key = api_key
          self.timeout = timeout
          self.use_cache = use_cache
522a3964   tangwang   多语言搜索翻译的优化(deepL添...
62
63
          self.glossary_id = glossary_id
          self.translation_context = translation_context or "e-commerce product search"
be52af70   tangwang   first commit
64
65
66
67
68
69
70
71
72
73
  
          if use_cache:
              self.cache = DictCache(".cache/translations.json")
          else:
              self.cache = None
  
      def translate(
          self,
          text: str,
          target_lang: str,
522a3964   tangwang   多语言搜索翻译的优化(deepL添...
74
75
          source_lang: Optional[str] = None,
          context: Optional[str] = None
be52af70   tangwang   first commit
76
77
78
79
80
81
82
83
      ) -> Optional[str]:
          """
          Translate text to target language.
  
          Args:
              text: Text to translate
              target_lang: Target language code ('zh', 'en', 'ru', etc.)
              source_lang: Source language code (optional, auto-detect if None)
522a3964   tangwang   多语言搜索翻译的优化(deepL添...
84
              context: Additional context for translation (overrides default context)
be52af70   tangwang   first commit
85
86
87
88
89
90
91
92
93
94
95
96
  
          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()
  
522a3964   tangwang   多语言搜索翻译的优化(deepL添...
97
98
99
100
          # Use provided context or default context
          translation_context = context or self.translation_context
  
          # Check cache (include context in cache key for accuracy)
be52af70   tangwang   first commit
101
          if self.use_cache:
522a3964   tangwang   多语言搜索翻译的优化(deepL添...
102
              cache_key = f"{source_lang or 'auto'}:{target_lang}:{translation_context}:{text}"
be52af70   tangwang   first commit
103
104
105
106
107
108
109
110
111
              cached = self.cache.get(cache_key, category="translations")
              if cached:
                  return cached
  
          # If no API key, return mock translation (for testing)
          if not self.api_key:
              print(f"[Translator] No API key, returning original text (mock mode)")
              return text
  
16c42787   tangwang   feat: implement r...
112
          # Translate using DeepL with fallback
522a3964   tangwang   多语言搜索翻译的优化(deepL添...
113
          result = self._translate_deepl(text, target_lang, source_lang, translation_context)
be52af70   tangwang   first commit
114
  
16c42787   tangwang   feat: implement r...
115
116
117
          # If translation failed, try fallback to free API
          if result is None and "api.deepl.com" in self.DEEPL_API_URL:
              print(f"[Translator] Pro API failed, trying free API...")
522a3964   tangwang   多语言搜索翻译的优化(deepL添...
118
              result = self._translate_deepl_free(text, target_lang, source_lang, translation_context)
16c42787   tangwang   feat: implement r...
119
120
121
122
123
124
  
          # If still failed, return original text with warning
          if result is None:
              print(f"[Translator] Translation failed, returning original text")
              result = text
  
be52af70   tangwang   first commit
125
126
          # Cache result
          if result and self.use_cache:
522a3964   tangwang   多语言搜索翻译的优化(deepL添...
127
              cache_key = f"{source_lang or 'auto'}:{target_lang}:{translation_context}:{text}"
be52af70   tangwang   first commit
128
129
130
131
132
133
134
135
              self.cache.set(cache_key, result, category="translations")
  
          return result
  
      def _translate_deepl(
          self,
          text: str,
          target_lang: str,
522a3964   tangwang   多语言搜索翻译的优化(deepL添...
136
137
          source_lang: Optional[str],
          context: Optional[str] = None
be52af70   tangwang   first commit
138
      ) -> Optional[str]:
522a3964   tangwang   多语言搜索翻译的优化(deepL添...
139
140
141
142
143
144
145
146
147
          """
          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")
          """
be52af70   tangwang   first commit
148
149
150
151
152
153
154
155
          # 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",
          }
  
522a3964   tangwang   多语言搜索翻译的优化(deepL添...
156
157
158
159
160
          # Build text with context for better disambiguation
          # 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, context)
  
be52af70   tangwang   first commit
161
          payload = {
522a3964   tangwang   多语言搜索翻译的优化(deepL添...
162
              "text": [text_to_translate],
be52af70   tangwang   first commit
163
164
165
166
167
168
169
              "target_lang": target_code,
          }
  
          if source_lang:
              source_code = self.LANG_CODE_MAP.get(source_lang, source_lang.upper())
              payload["source_lang"] = source_code
  
522a3964   tangwang   多语言搜索翻译的优化(deepL添...
170
171
172
173
174
175
176
177
178
179
          # Add glossary if configured
          if self.glossary_id:
              payload["glossary_id"] = self.glossary_id
  
          # Note: DeepL API v2 doesn't have a direct "context" parameter,
          # but we can improve translation by:
          # 1. Using glossary for domain-specific terms (best solution)
          # 2. Adding context words to the text (for single-word queries) - implemented in _add_ecommerce_context
          # 3. Using more specific source language detection
  
be52af70   tangwang   first commit
180
181
182
183
184
185
186
187
188
189
190
          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:
522a3964   tangwang   多语言搜索翻译的优化(deepL添...
191
192
193
194
195
196
197
                      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
                          )
                      return translated_text
be52af70   tangwang   first commit
198
199
200
201
202
203
204
205
206
207
208
              else:
                  print(f"[Translator] DeepL API error: {response.status_code} - {response.text}")
                  return None
  
          except requests.Timeout:
              print(f"[Translator] Translation request timed out")
              return None
          except Exception as e:
              print(f"[Translator] Translation failed: {e}")
              return None
  
16c42787   tangwang   feat: implement r...
209
210
211
212
      def _translate_deepl_free(
          self,
          text: str,
          target_lang: str,
522a3964   tangwang   多语言搜索翻译的优化(deepL添...
213
214
          source_lang: Optional[str],
          context: Optional[str] = None
16c42787   tangwang   feat: implement r...
215
      ) -> Optional[str]:
522a3964   tangwang   多语言搜索翻译的优化(deepL添...
216
217
218
219
220
          """
          Translate using DeepL Free API.
          
          Note: Free API may not support glossary_id parameter.
          """
16c42787   tangwang   feat: implement r...
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
          # 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",
          }
  
          payload = {
              "text": [text],
              "target_lang": target_code,
          }
  
          if source_lang:
              source_code = self.LANG_CODE_MAP.get(source_lang, source_lang.upper())
              payload["source_lang"] = source_code
  
522a3964   tangwang   多语言搜索翻译的优化(deepL添...
238
239
240
          # Note: Free API typically doesn't support glossary_id
          # But we can still use context hints in the text
  
16c42787   tangwang   feat: implement r...
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
          try:
              response = requests.post(
                  "https://api-free.deepl.com/v2/translate",
                  headers=headers,
                  json=payload,
                  timeout=self.timeout
              )
  
              if response.status_code == 200:
                  data = response.json()
                  if "translations" in data and len(data["translations"]) > 0:
                      return data["translations"][0]["text"]
              else:
                  print(f"[Translator] DeepL Free API error: {response.status_code} - {response.text}")
                  return None
  
          except requests.Timeout:
              print(f"[Translator] Free API request timed out")
              return None
          except Exception as e:
              print(f"[Translator] Free API translation failed: {e}")
              return None
  
be52af70   tangwang   first commit
264
265
266
267
      def translate_multi(
          self,
          text: str,
          target_langs: List[str],
522a3964   tangwang   多语言搜索翻译的优化(deepL添...
268
269
          source_lang: Optional[str] = None,
          context: Optional[str] = None
be52af70   tangwang   first commit
270
271
272
273
274
275
276
277
      ) -> Dict[str, Optional[str]]:
          """
          Translate text to multiple target languages.
  
          Args:
              text: Text to translate
              target_langs: List of target language codes
              source_lang: Source language code (optional)
522a3964   tangwang   多语言搜索翻译的优化(deepL添...
278
              context: Context hint for translation (optional)
be52af70   tangwang   first commit
279
280
281
282
283
284
  
          Returns:
              Dictionary mapping language code to translated text
          """
          results = {}
          for lang in target_langs:
522a3964   tangwang   多语言搜索翻译的优化(deepL添...
285
              results[lang] = self.translate(text, lang, source_lang, context)
be52af70   tangwang   first commit
286
287
          return results
  
522a3964   tangwang   多语言搜索翻译的优化(deepL添...
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
      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
  
be52af70   tangwang   first commit
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
      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 lang != detected_lang]
  
          # Otherwise, translate to all supported languages
          return supported_langs