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
# Add parent directory to path
sys.path.insert(0, str(Path(__file__).parent.parent))
from config import ConfigLoader
from query.translator 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}")
        print(f"  Translation prompt configuration:")
        for key, value in config.query_config.translation_prompts.items():
            print(f"    {key}: {value[:60]}..." if len(value) > 60 else f"    {key}: {value}")
        
        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
        )
        
        # 测试商品标题翻译（使用product_title提示词）
        test_texts = [
            ("蓝牙耳机", "zh", "en", "product_title"),
            ("Wireless Headphones", "en", "zh", "product_title"),
        ]
        
        for text, source_lang, target_lang, prompt_type in test_texts:
            if prompt_type == "product_title":
                if target_lang == "zh":
                    prompt = config.query_config.translation_prompts.get('product_title_zh')
                else:
                    prompt = config.query_config.translation_prompts.get('product_title_en')
            else:
                if target_lang == "zh":
                    prompt = config.query_config.translation_prompts.get('default_zh')
                else:
                    prompt = config.query_config.translation_prompts.get('default_en')
            
            print(f"\nTranslation test:")
            print(f"  Original text ({source_lang}): {text}")
            print(f"  Target language: {target_lang}")
            print(f"  Prompt: {prompt[:50] if prompt else 'None'}...")
            
            result = translator.translate(
                text,
                target_lang=target_lang,
                source_lang=source_lang,
                prompt=prompt
            )
            
            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'
        
        query_prompt = config.query_config.translation_prompts.get('query_zh')
        
        print(f"Query text: {query_text}")
        print(f"Target languages: {target_langs}")
        print(f"Prompt: {query_prompt}")
        
        # 异步模式（立即返回，后台翻译）
        results = translator.translate_multi(
            query_text,
            target_langs,
            source_lang=source_lang,
            context=config.query_config.translation_context,
            async_mode=True,
            prompt=query_prompt
        )
        
        print(f"\nAsynchronous translation results:")
        for lang, translation in results.items():
            if translation:
                print(f"  {lang}: {translation} (cache hit)")
            else:
                print(f"  {lang}: None (translating in background...)")
        
        # 同步模式（等待完成）
        print(f"\nSynchronous translation (waiting for completion):")
        results_sync = translator.translate_multi(
            query_text,
            target_langs,
            source_lang=source_lang,
            context=config.query_config.translation_context,
            async_mode=False,
            prompt=query_prompt
        )
        
        for lang, translation in results_sync.items():
            print(f"  {lang}: {translation}")
        
    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"
        prompt = config.query_config.translation_prompts.get('default_zh')
        
        print(f"First translation (should call API or return mock):")
        result1 = translator.translate(test_text, target_lang, source_lang, prompt=prompt)
        print(f"  Result: {result1}")
        print(f"\nSecond translation (should use cache):")
        result2 = translator.translate(test_text, target_lang, source_lang, prompt=prompt)
        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 = "手机"
        prompt = config.query_config.translation_prompts.get('query_zh')
        
        print(f"Test text: {text}")
        print(f"Prompt (as context): {prompt}")
        
        # 带context的翻译
        result_with_context = translator.translate(
            text,
            target_lang='en',
            source_lang='zh',
            prompt=prompt
        )
        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()