test_ml_final.py

#!/usr/bin/env python3
"""
test_ml_final.py - Test the trained ML models
Run: python test_ml_final.py
"""

import sys
import asyncio
import json
sys.path.insert(0, '.')

from app.ml.models.ml_listing_extractor import get_ml_extractor

async def main():
    """Test ML model functionality"""
    
    print("\n" + "="*70)
    print("🧪 TESTING TRAINED ML MODELS")
    print("="*70 + "\n")
    
    extractor = get_ml_extractor()
    
    # Test 1: Location extraction
    print("Test 1: Location Extraction")
    print("-" * 70)
    
    test_locations = [
        "Akpkpa Kpondehou",
        "Victoria Island, Lagos",
        "Kilimani, Nairobi",
        "Downtown Dubai",
        "Chelsea, London",
    ]
    
    for location in test_locations:
        try:
            city, info = await extractor.extract_location_from_address(location)
            print(f"✅ '{location}' → {city}")
        except Exception as e:
            print(f"❌ '{location}' failed: {e}")
    
    print("\n")
    
    # Test 2: Listing Type Inference
    print("Test 2: Listing Type Inference")
    print("-" * 70)
    
    test_cases = [
        {
            "name": "Monthly rent",
            "state": {"price_type": "monthly"},
            "user_role": "landlord",
            "message": "monthly rent"
        },
        {
            "name": "Nightly (short-stay)",
            "state": {"price_type": "nightly"},
            "user_role": "landlord",
            "message": "nightly"
        },
        {
            "name": "Sale keyword",
            "state": {},
            "user_role": "landlord",
            "message": "I want to SELL my property"
        },
        {
            "name": "Renter listing",
            "state": {},
            "user_role": "renter",
            "message": "list a space"
        },
    ]
    
    for test in test_cases:
        try:
            listing_type, conf = extractor.infer_listing_type(
                test["state"],
                user_role=test["user_role"],
                user_message=test["message"]
            )
            print(f"✅ {test['name']}: {listing_type} ({conf:.0%})")
        except Exception as e:
            print(f"❌ {test['name']} failed: {e}")
    
    print("\n")
    
    # Test 3: Currency Inference
    print("Test 3: Currency Inference")
    print("-" * 70)
    
    test_currencies = [
        {"location": "cotonou", "expected": "XOF"},
        {"location": "lagos", "expected": "NGN"},
        {"location": "nairobi", "expected": "KES"},
        {"location": "dubai", "expected": "AED"},
        {"location": "london", "expected": "GBP"},
    ]
    
    for test in test_currencies:
        try:
            state = {"location": test["location"]}
            currency, city, conf = await extractor.infer_currency(state)
            status = "✅" if currency == test["expected"] else "⚠️"
            print(f"{status} {test['location']}: {currency} ({conf:.0%})")
        except Exception as e:
            print(f"❌ {test['location']} failed: {e}")
    
    print("\n")
    
    # Test 4: Field Validation
    print("Test 4: Field Validation")
    print("-" * 70)
    
    validation_tests = [
        {"field": "bedrooms", "value": 2, "expected": True},
        {"field": "bedrooms", "value": 0, "expected": False},
        {"field": "price", "value": 500000, "expected": True},
        {"field": "price", "value": -100, "expected": False},
        {"field": "price_type", "value": "monthly", "expected": True},
        {"field": "price_type", "value": "invalid", "expected": False},
    ]
    
    for test in validation_tests:
        try:
            result = extractor.validate_field(
                test["field"],
                test["value"],
                f"test {test['field']}",
                "test_user"
            )
            status = "✅" if result["is_valid"] == test["expected"] else "⚠️"
            print(f"{status} {test['field']}={test['value']}: {result['is_valid']}")
        except Exception as e:
            print(f"❌ {test['field']}={test['value']} failed: {e}")
    
    print("\n")
    
    # Test 5: Full Flow
    print("Test 5: Complete Flow")
    print("-" * 70)
    
    try:
        state = {
            "location": "Cotonou, Benin",  # Use more specific location
            "bedrooms": 2,
            "bathrooms": 1,
            "price": 500000,
            "price_type": "monthly",
            "user_id": "test_user",
        }
        
        # Extract location
        city, info = await extractor.extract_location_from_address(state["location"])
        if city:
            state["location"] = city
            print(f"✅ Location extracted: {city}")
        else:
            state["location"] = "cotonou"  # Fallback
            print(f"⚠️ Location extraction fallback: cotonou")
        
        # Infer currency
        currency, _, curr_conf = await extractor.infer_currency(state)
        if currency:
            state["currency"] = currency
            print(f"✅ Currency inferred: {currency}")
        else:
            state["currency"] = "XOF"  # Fallback
            print(f"⚠️ Currency fallback: XOF")
        
        # Infer listing type
        listing_type, type_conf = extractor.infer_listing_type(
            state,
            user_role="landlord",
            user_message="monthly rental"
        )
        state["listing_type"] = listing_type
        print(f"✅ Listing type inferred: {listing_type}")
        
        # Validate all fields
        validation = extractor.validate_all_fields(state, "test_user")
        print(f"✅ Validation: {validation['all_valid']}")
        
        # Get display price
        if state.get("currency"):
            price_display = await extractor.get_display_price(state, user_currency="USD")
            if price_display:
                print(f"✅ Display price: {price_display['formatted']}")
            else:
                print(f"⚠️ Price display failed")
        
        print("\n✅ Complete flow PASSED!")
        
    except Exception as e:
        print(f"❌ Complete flow failed: {e}")
        import traceback
        traceback.print_exc()
    
    print("\n" + "="*70)
    print("✅ ALL TESTS COMPLETE!")
    print("="*70 + "\n")


if __name__ == "__main__":
    asyncio.run(main())