test_generation_api.py

#!/usr/bin/env python3
"""
Test generation API endpoint
"""
import requests
import time
import json

BASE_URL = "http://localhost:5002"

def test_generation():
    print("🧪 Testing Lead Generation API")
    print("="*60)
    
    # Test data
    payload = {
        "markets": ["India"],
        "cities": ["Mumbai"],
        "business_types": ["restaurant"],
        "num_leads": 5,
        "quality_threshold": 70,
        "clear_old": False
    }
    
    print(f"\n📤 Sending generation request...")
    print(f"   Country: India")
    print(f"   City: Mumbai")
    print(f"   Type: restaurant")
    print(f"   Target: 5 leads")
    
    try:
        # Start generation
        response = requests.post(
            f"{BASE_URL}/api/generate",
            json=payload,
            timeout=10
        )
        
        print(f"\n📡 Response Status: {response.status_code}")
        data = response.json()
        print(f"📡 Response Data: {json.dumps(data, indent=2)}")
        
        if not data.get('success'):
            print(f"\n❌ Generation failed to start!")
            return
        
        print(f"\n✅ Generation started!")
        print(f"\n⏳ Monitoring progress...")
        
        # Monitor progress
        for i in range(30):  # Max 30 seconds
            time.sleep(2)
            
            status_response = requests.get(f"{BASE_URL}/api/status")
            status_data = status_response.json()
            
            is_running = status_data.get('is_running', False)
            progress = status_data.get('progress', 0)
            leads_found = status_data.get('status', {}).get('leads_found', 0)
            message = status_data.get('status', {}).get('message', '')
            
            print(f"\r   Progress: {progress}% | Leads: {leads_found} | {message}", end='', flush=True)
            
            if not is_running:
                print(f"\n\n✅ Generation complete!")
                break
        
        # Check final results
        print(f"\n\n📊 Checking final results...")
        leads_response = requests.get(f"{BASE_URL}/api/leads?page=0&limit=10")
        leads_data = leads_response.json()
        
        if leads_data.get('success'):
            total = leads_data.get('total', 0)
            leads = leads_data.get('leads', [])
            
            print(f"✅ Total leads in database: {total}")
            print(f"✅ Fetched {len(leads)} leads")
            
            if leads:
                print(f"\n📋 Sample lead:")
                lead = leads[0]
                print(f"   Title: {lead.get('title', 'N/A')}")
                print(f"   Type: {lead.get('type', 'N/A')}")
                print(f"   Address: {lead.get('address', 'N/A')}")
                print(f"   Rating: {lead.get('rating', 'N/A')}")
                print(f"   Quality: {lead.get('quality_score', 'N/A')}/100")
        
    except requests.exceptions.ConnectionError:
        print(f"\n❌ Could not connect to {BASE_URL}")
        print(f"   Make sure dashboard is running:")
        print(f"   .venv/bin/python3 dashboard_ragspro.py")
    except Exception as e:
        print(f"\n❌ Error: {e}")
        import traceback
        traceback.print_exc()

if __name__ == '__main__':
    test_generation()