query/test_translation.py

#!/usr/bin/env python3
"""
Translation function test script.
Test content:
1. Translation prompt configuration loading
2. Synchronous translation (indexing scenario)
3. Asynchronous translation (query scenario)
4. Usage of different prompts
5. Cache functionality
6. DeepL Context parameter usage
"""
import sys
import os
from pathlib import Path
from concurrent.futures import ThreadPoolExecutor
# Add parent directory to path
sys.path.insert(0, str(Path(__file__).parent.parent))
from config import ConfigLoader
from query.qwen_mt_translate import Translator
import logging
# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
def test_config_loading():
    """Test configuration loading"""
    print("\n" + "="*60)
    print("Test 1: Configuration loading")
    print("="*60)
    
    try:
        config_loader = ConfigLoader()
        config = config_loader.load_config()
        
        print(f"✓ Configuration loaded successfully")
        print(f"  Translation service: {config.query_config.translation_service}")
        
        return config
    except Exception as e:
        print(f"✗ Configuration loading failed: {e}")
        import traceback
        traceback.print_exc()
        return None
def test_translator_sync(config):
    """Test synchronous translation (indexing scenario)"""
    print("\n" + "="*60)
    print("Test 2: Synchronous translation (indexing scenario)")
    print("="*60)
    
    if not config:
        print("✗ Skipped: Configuration not loaded")
        return None
    
    try:
        translator = Translator(
            api_key=config.query_config.translation_api_key,
            use_cache=True,
            glossary_id=config.query_config.translation_glossary_id,
            translation_context=config.query_config.translation_context
        )
        
        # 测试商品标题翻译（使用sku_name提示词）
        test_texts = [
            ("蓝牙耳机", "zh", "en", "sku_name"),
            ("Wireless Headphones", "en", "zh", "sku_name"),
        ]
        
        for text, source_lang, target_lang, scene in test_texts:
            print(f"\nTranslation test:")
            print(f"  Original text ({source_lang}): {text}")
            print(f"  Target language: {target_lang}")
            print(f"  Scene: {scene}")
            
            result = translator.translate(
                text,
                target_lang=target_lang,
                source_lang=source_lang,
                context=scene,
            )
            
            if result:
                print(f"  Result: {result}")
                print(f"  ✓ Translation successful")
            else:
                print(f"  ⚠ Translation returned None (possibly mock mode or no API key)")
        
        return translator
        
    except Exception as e:
        print(f"✗ Synchronous translation test failed: {e}")
        import traceback
        traceback.print_exc()
        return None
def test_translator_async(config, translator):
    """Test asynchronous translation (query scenario)"""
    print("\n" + "="*60)
    print("Test 3: Asynchronous translation (query scenario)")
    print("="*60)
    
    if not config or not translator:
        print("✗ Skipped: Configuration or translator not initialized")
        return
    
    try:
        query_text = "手机"
        target_langs = ['en']
        source_lang = 'zh'
        print(f"Query text: {query_text}")
        print(f"Target languages: {target_langs}")
        print("Scene: ecommerce_search_query")
        
        print(f"\nConcurrent translation via generic translate():")
        with ThreadPoolExecutor(max_workers=len(target_langs)) as executor:
            futures = {
                lang: executor.submit(
                    translator.translate,
                    query_text,
                    lang,
                    source_lang,
                    "ecommerce_search_query",
                )
                for lang in target_langs
            }
            for lang, future in futures.items():
                print(f"  {lang}: {future.result()}")
        
    except Exception as e:
        print(f"✗ Asynchronous translation test failed: {e}")
        import traceback
        traceback.print_exc()
def test_cache():
    """测试缓存功能"""
    print("\n" + "="*60)
    print("Test 4: Cache functionality")
    print("="*60)
    
    try:
        config_loader = ConfigLoader()
        config = config_loader.load_config()
        
        translator = Translator(
            api_key=config.query_config.translation_api_key,
            use_cache=True
        )
        
        test_text = "测试文本"
        target_lang = "en"
        source_lang = "zh"
        
        print(f"First translation (should call API or return mock):")
        result1 = translator.translate(test_text, target_lang, source_lang, context="default")
        print(f"  Result: {result1}")
        print(f"\nSecond translation (should use cache):")
        result2 = translator.translate(test_text, target_lang, source_lang, context="default")
        print(f"  Result: {result2}")
        if result1 == result2:
            print(f"  ✓ Cache functionality working properly")
        else:
            print(f"  ⚠ Cache might have issues")
        
    except Exception as e:
        print(f"✗ Cache test failed: {e}")
        import traceback
        traceback.print_exc()
def test_context_parameter():
    """Test DeepL Context parameter usage"""
    print("\n" + "="*60)
    print("Test 5: DeepL Context parameter")
    print("="*60)
    
    try:
        config_loader = ConfigLoader()
        config = config_loader.load_config()
        
        translator = Translator(
            api_key=config.query_config.translation_api_key,
            use_cache=False  # 禁用缓存以便测试
        )
        
        # 测试带context和不带context的翻译
        text = "手机"
        
        print(f"Test text: {text}")
        print("Scene: ecommerce_search_query")
        
        # 带context的翻译
        result_with_context = translator.translate(
            text,
            target_lang='en',
            source_lang='zh',
            context="ecommerce_search_query",
        )
        print(f"\nTranslation result with context: {result_with_context}")
        
        # 不带context的翻译
        result_without_context = translator.translate(
            text,
            target_lang='en',
            source_lang='zh',
            prompt=None
        )
        print(f"Translation result without context: {result_without_context}")
        
        print(f"\n✓ Context parameter test completed")
        print(f"  Note: According to DeepL API, context parameter affects translation but does not participate in translation itself")
        
    except Exception as e:
        print(f"✗ Context parameter test failed: {e}")
        import traceback
        traceback.print_exc()
def main():
    """Main test function"""
    print("="*60)
    print("Translation function test")
    print("="*60)
    
    # 测试1: 配置加载
    config = test_config_loading()
    
    # 测试2: 同步翻译
    translator = test_translator_sync(config)
    
    # 测试3: 异步翻译
    test_translator_async(config, translator)
    
    # 测试4: 缓存功能
    test_cache()
    
    # 测试5: Context参数
    test_context_parameter()
    
    print("\n" + "="*60)
    print("Test completed")
    print("="*60)
if __name__ == '__main__':
    main()