""" 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 使用方法 (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. """ DEEPL_API_URL = "https://api.deepl.com/v2/translate" # Pro tier QWEN_BASE_URL = "https://dashscope.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) # If still failed, return original text with warning if result is None: logger.warning( f"[Translator] Translation request | Original text: '{text}' | Target language: {target_lang} | " f"Source language: {source_lang or 'auto'} | Result: '{text}' | Status: Translation failed, returning original" ) result = text 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 result if result 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()