server/tests/test_intent_features.py

"""
Test Intent Classification and Query Routing

Run with: python test_intent_features.py
"""

import sys
import os
import asyncio

# Add parent directory to path
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from app.services.intent_classifier import (
    IntentClassifier,
    IntentCategory,
    classify_intent
)
from app.services.query_router import QueryRouter


def test_intent_classification():
    """Test intent classification"""
    print("=" * 60)
    print("TESTING INTENT CLASSIFICATION")
    print("=" * 60)
    
    classifier = IntentClassifier()
    
    # Test cases with expected intents
    test_cases = [
        # FAQ intents
        ("How do I reset my password?", IntentCategory.FAQ, "healthcare"),
        ("What are your office hours?", IntentCategory.FAQ, "education"),
        ("How to contact support?", IntentCategory.FAQ, None),
        
        # Creative intents
        ("Write an email to my professor", IntentCategory.CREATIVE, "education"),
        ("Generate a summary of this article", IntentCategory.CREATIVE, None),
        ("Create a report about symptoms", IntentCategory.CREATIVE, "healthcare"),
        
        # Industry knowledge
        ("What are the symptoms of flu?", IntentCategory.INDUSTRY_KNOWLEDGE, "healthcare"),
        ("What is GPA?", IntentCategory.INDUSTRY_KNOWLEDGE, "education"),
        ("Explain what diabetes is", IntentCategory.INDUSTRY_KNOWLEDGE, "healthcare"),
        
        # Business specific
        ("Who is Dr. Ali?", IntentCategory.BUSINESS_SPECIFIC, "healthcare"),
        ("What are your clinic timings?", IntentCategory.BUSINESS_SPECIFIC, "healthcare"),
        ("When does our university start?", IntentCategory.BUSINESS_SPECIFIC, "education"),
    ]
    
    print("\nTest Results:")
    correct = 0
    total = len(test_cases)
    
    for query, expected_intent, industry in test_cases:
        result = classifier.classify(query, industry=industry)
        status = "✓" if result.category == expected_intent else "✗"
        
        if result.category == expected_intent:
            correct += 1
        
        print(f"\n{status} Query: '{query}'")
        print(f"  Industry: {industry}")
        print(f"  Expected: {expected_intent.value}")
        print(f"  Detected: {result.category.value} (confidence: {result.confidence:.2f})")
        print(f"  Reasoning: {result.reasoning}")
        
        if result.alternative_intents:
            print(f"  Alternatives:")
            for alt_cat, alt_conf in result.alternative_intents[:2]:
                print(f"    - {alt_cat.value}: {alt_conf:.2f}")
    
    print(f"\n📊 Accuracy: {correct}/{total} ({100*correct/total:.1f}%)")


async def test_query_routing():
    """Test query routing"""
    print("\n" + "=" * 60)
    print("TESTING QUERY ROUTING")
    print("=" * 60)
    
    router = QueryRouter()
    
    # Test cases
    test_queries = [
        ("Hello, I need help", 1, "healthcare"),
        ("مجھے بخار ہے", 1, "healthcare"),  # "I have fever" in Urdu
        ("admission ke liye kya chahiye", 2, "education"),  # Roman Urdu
        ("What are symptoms of diabetes?", 1, "healthcare"),
        ("How to reset password?", 2, None),
    ]
    
    print("\nRouting Results:")
    
    for query, website_id, industry in test_queries:
        try:
            result = await router.route_query(
                query=query,
                website_id=website_id,
                industry=industry,
                session_context={"business_query_count": 0}
            )
            
            print(f"\n📩 Query: '{query}'")
            print(f"  Original Language: {result['original_language']}")
            print(f"  Processed Query: '{result['processed_query']}'")
            print(f"  Intent: {result['intent_category']} (conf: {result['intent_confidence']:.2f})")
            print(f"  Handler: {result['handler']}")
            print(f"  Should Translate Response: {result['should_translate_response']}")
            
            # Check if should escalate
            should_escalate = router.should_escalate_to_owner(result, 0.7)
            print(f"  Should Escalate: {should_escalate}")
            
        except Exception as e:
            print(f"\n❌ Error routing query '{query}': {e}")
            import traceback
            traceback.print_exc()


def test_multilingual_intent():
    """Test intent classification with multilingual queries"""
    print("\n" + "=" * 60)
    print("TESTING MULTILINGUAL INTENT CLASSIFICATION")
    print("=" * 60)
    
    classifier = IntentClassifier()
    
    # Same question in different languages
    multilingual_tests = [
        ("What is diabetes?", "en", IntentCategory.INDUSTRY_KNOWLEDGE),
        ("diabetes kya hai?", "ur-roman", IntentCategory.INDUSTRY_KNOWLEDGE),
        ("How do I enroll in courses?", "en", IntentCategory.FAQ),
        ("course mein kaise enroll karun?", "ur-roman", IntentCategory.FAQ),
    ]
    
    print("\nMultilingual Test Results:")
    
    for query, lang, expected in multilingual_tests:
        result = classifier.classify(query, industry="healthcare")
        status = "✓" if result.category == expected else "✗"
        
        print(f"\n{status} Query ({lang}): '{query}'")
        print(f"  Expected: {expected.value}")
        print(f"  Detected: {result.category.value} (conf: {result.confidence:.2f})")


if __name__ == "__main__":
    print("\n🚀 Starting Intent Classification Tests\n")
    
    try:
        # Synchronous tests
        test_intent_classification()
        test_multilingual_intent()
        
        # Async tests
        print("\n🔄 Running Async Tests...")
        asyncio.run(test_query_routing())
        
        print("\n" + "=" * 60)
        print("✅ ALL TESTS COMPLETED")
        print("=" * 60 + "\n")
        
    except Exception as e:
        print(f"\n❌ TEST FAILED: {e}")
        import traceback
        traceback.print_exc()