Best FindThatLead Phone Validation API & Email Verification Integration 2025

import requests
import json
import asyncio
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass
from datetime import datetime

@dataclass
class EnhancedProspect:
    email: str
    first_name: Optional[str] = None
    last_name: Optional[str] = None
    position: Optional[str] = None
    company: Optional[str] = None
    domain: Optional[str] = None
    phone: Optional[str] = None
    linkedin_url: Optional[str] = None
    confidence_score: float = 0.0
    email_validation: Optional[Dict] = None
    phone_validation: Optional[Dict] = None
    social_profiles: Optional[Dict] = None
    lead_score: int = 0
    enhancement_timestamp: str = ""
    source: str = "findthatlead"

class FindThatLead1lookupIntegration:
    def __init__(self, ftl_api_key: str, lookup_api_key: str):
        self.ftl_api_key = ftl_api_key
        self.lookup_api_key = lookup_api_key
        self.ftl_base_url = "https://api.findthatlead.com/v1"
        self.lookup_base_url = "https://api.1lookup.app/v1"
        
    async def enhanced_domain_search(self, domain: str, limit: int = 100) -> List[EnhancedProspect]:
        """
        Perform FindThatLead domain search with complete 1lookup enhancement
        """
        try:
            # Step 1: FindThatLead domain search
            ftl_contacts = await self._ftl_domain_search(domain, limit)
            
            if not ftl_contacts:
                return []
            
            # Step 2: Parallel enhancement of all contacts
            enhanced_prospects = await self._bulk_enhance_contacts(ftl_contacts, domain)
            
            # Step 3: Filter and sort by lead score
            qualified_prospects = [p for p in enhanced_prospects if p.lead_score >= 60]
            qualified_prospects.sort(key=lambda x: x.lead_score, reverse=True)
            
            return qualified_prospects
            
        except Exception as e:
            print(f"Error in domain search: {e}")
            return []
    
    async def enhanced_person_search(self, first_name: str, last_name: str, 
                                   domain: str, company: str = None) -> Optional[EnhancedProspect]:
        """
        Find and enhance individual prospect with comprehensive validation
        """
        try:
            # Step 1: FindThatLead person search
            ftl_result = await self._ftl_person_search(first_name, last_name, domain, company)
            
            if not ftl_result:
                return None
            
            # Step 2: Create base prospect
            prospect = EnhancedProspect(
                email=ftl_result.get('email', ''),
                first_name=first_name,
                last_name=last_name,
                position=ftl_result.get('position'),
                company=company or ftl_result.get('company'),
                domain=domain,
                linkedin_url=ftl_result.get('linkedin_url'),
                confidence_score=ftl_result.get('confidence', 0) / 100,
                enhancement_timestamp=datetime.now().isoformat()
            )
            
            # Step 3: Comprehensive enhancement
            await self._enhance_single_prospect(prospect)
            
            return prospect
            
        except Exception as e:
            print(f"Error in person search: {e}")
            return None
    
    async def _ftl_domain_search(self, domain: str, limit: int) -> List[Dict]:
        """FindThatLead domain search API call"""
        url = f"{self.ftl_base_url}/domains/{domain}/emails"
        headers = {'Authorization': f'Bearer {self.ftl_api_key}'}
        params = {'limit': limit}
        
        async with aiohttp.ClientSession() as session:
            async with session.get(url, headers=headers, params=params) as response:
                if response.status == 200:
                    data = await response.json()
                    return data.get('data', [])
                return []
    
    async def _ftl_person_search(self, first_name: str, last_name: str, 
                               domain: str, company: str = None) -> Optional[Dict]:
        """FindThatLead person search API call"""
        url = f"{self.ftl_base_url}/emails/search"
        headers = {'Authorization': f'Bearer {self.ftl_api_key}'}
        
        payload = {
            'first_name': first_name,
            'last_name': last_name,
            'domain': domain
        }
        
        if company:
            payload['company'] = company
        
        async with aiohttp.ClientSession() as session:
            async with session.post(url, headers=headers, json=payload) as response:
                if response.status == 200:
                    data = await response.json()
                    return data.get('data')
                return None
    
    async def _bulk_enhance_contacts(self, contacts: List[Dict], domain: str) -> List[EnhancedProspect]:
        """Enhance multiple contacts in parallel"""
        enhancement_tasks = []
        
        for contact_data in contacts:
            prospect = EnhancedProspect(
                email=contact_data.get('email', ''),
                first_name=contact_data.get('first_name'),
                last_name=contact_data.get('last_name'),
                position=contact_data.get('position'),
                company=contact_data.get('company'),
                domain=domain,
                linkedin_url=contact_data.get('linkedin_url'),
                confidence_score=contact_data.get('confidence', 0) / 100,
                enhancement_timestamp=datetime.now().isoformat()
            )
            
            task = self._enhance_single_prospect(prospect)
            enhancement_tasks.append(task)
        
        # Process all enhancements in parallel
        await asyncio.gather(*enhancement_tasks, return_exceptions=True)
        
        return [task.result() for task in enhancement_tasks if not isinstance(task.result(), Exception)]
    
    async def _enhance_single_prospect(self, prospect: EnhancedProspect) -> None:
        """Comprehensive single prospect enhancement"""
        enhancement_tasks = []
        
        # Email validation
        if prospect.email:
            task1 = self._validate_email_with_1lookup(prospect.email)
            enhancement_tasks.append(task1)
        
        # Phone enrichment
        if prospect.first_name and prospect.last_name:
            task2 = self._enrich_phone_data(prospect)
            enhancement_tasks.append(task2)
        
        # Social profile enrichment
        task3 = self._enrich_social_profiles(prospect)
        enhancement_tasks.append(task3)
        
        # Execute all enhancements in parallel
        results = await asyncio.gather(*enhancement_tasks, return_exceptions=True)
        
        # Process results
        if len(results) > 0 and not isinstance(results[0], Exception):
            prospect.email_validation = results[0]
        
        if len(results) > 1 and not isinstance(results[1], Exception) and results[1]:
            prospect.phone_validation = results[1]
            prospect.phone = results[1].get('phone_number')
        
        if len(results) > 2 and not isinstance(results[2], Exception):
            prospect.social_profiles = results[2]
        
        # Calculate comprehensive lead score
        prospect.lead_score = self._calculate_advanced_lead_score(prospect)
    
    async def _validate_email_with_1lookup(self, email: str) -> Dict:
        """1lookup email validation"""
        url = f"{self.lookup_base_url}/email/validate"
        headers = {
            'Authorization': f'Bearer {self.lookup_api_key}',
            'Content-Type': 'application/json'
        }
        
        payload = {
            'email': email,
            'enhanced_checks': True,
            'spam_detection': True
        }
        
        async with aiohttp.ClientSession() as session:
            async with session.post(url, json=payload, headers=headers) as response:
                if response.status == 200:
                    return await response.json()
                return {'deliverable': False, 'error': 'validation_failed'}
    
    async def _enrich_phone_data(self, prospect: EnhancedProspect) -> Optional[Dict]:
        """Phone number enrichment"""
        url = f"{self.lookup_base_url}/enrich/phone"
        headers = {
            'Authorization': f'Bearer {self.lookup_api_key}',
            'Content-Type': 'application/json'
        }
        
        payload = {
            'first_name': prospect.first_name,
            'last_name': prospect.last_name,
            'company': prospect.company,
            'domain': prospect.domain,
            'email': prospect.email
        }
        
        try:
            async with aiohttp.ClientSession() as session:
                async with session.post(url, json=payload, headers=headers) as response:
                    if response.status == 200:
                        return await response.json()
        except:
            pass
        
        return None
    
    async def _enrich_social_profiles(self, prospect: EnhancedProspect) -> Optional[Dict]:
        """Social profile enrichment"""
        url = f"{self.lookup_base_url}/enrich/social"
        headers = {
            'Authorization': f'Bearer {self.lookup_api_key}',
            'Content-Type': 'application/json'
        }
        
        payload = {
            'first_name': prospect.first_name,
            'last_name': prospect.last_name,
            'company': prospect.company,
            'domain': prospect.domain,
            'linkedin_url': prospect.linkedin_url
        }
        
        try:
            async with aiohttp.ClientSession() as session:
                async with session.post(url, json=payload, headers=headers) as response:
                    if response.status == 200:
                        return await response.json()
        except:
            pass
        
        return None
    
    def _calculate_advanced_lead_score(self, prospect: EnhancedProspect) -> int:
        """Calculate comprehensive lead quality score"""
        score = 0
        
        # FindThatLead confidence (0-25 points)
        ftl_confidence = prospect.confidence_score * 100
        score += int(ftl_confidence * 0.25)
        
        # Email validation score (0-30 points)
        if prospect.email_validation:
            if prospect.email_validation.get('deliverable'):
                score += 30
            elif prospect.email_validation.get('risky'):
                score += 15
        
        # Phone validation bonus (0-25 points)
        if prospect.phone_validation and prospect.phone_validation.get('valid'):
            score += 25
        
        # Social presence bonus (0-10 points)
        if prospect.social_profiles:
            profile_count = len([p for p in prospect.social_profiles.values() if p])
            score += min(profile_count * 3, 10)
        
        # Position/seniority scoring (0-10 points)
        if prospect.position:
            position_lower = prospect.position.lower()
            if any(title in position_lower for title in ['ceo', 'founder', 'president', 'owner']):
                score += 10
            elif any(title in position_lower for title in ['vp', 'vice president', 'director', 'head']):
                score += 8
            elif any(title in position_lower for title in ['manager', 'lead', 'senior', 'principal']):
                score += 5
        
        return min(score, 100)
    
    async def export_enhanced_prospects(self, prospects: List[EnhancedProspect], 
                                      filename: str = None) -> str:
        """Export enhanced prospects to CSV"""
        import pandas as pd
        
        if not filename:
            filename = f"findthatlead_enhanced_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
        
        # Convert to DataFrame
        data = []
        for prospect in prospects:
            row = {
                'email': prospect.email,
                'first_name': prospect.first_name,
                'last_name': prospect.last_name,
                'position': prospect.position,
                'company': prospect.company,
                'domain': prospect.domain,
                'phone': prospect.phone,
                'linkedin_url': prospect.linkedin_url,
                'lead_score': prospect.lead_score,
                'email_deliverable': prospect.email_validation.get('deliverable') if prospect.email_validation else False,
                'phone_valid': prospect.phone_validation.get('valid') if prospect.phone_validation else False,
                'social_profiles_count': len(prospect.social_profiles) if prospect.social_profiles else 0,
                'ftl_confidence': prospect.confidence_score,
                'enhancement_date': prospect.enhancement_timestamp
            }
            data.append(row)
        
        df = pd.DataFrame(data)
        df.to_csv(filename, index=False)
        
        return filename

# Example usage
async def main():
    integration = FindThatLead1lookupIntegration(
        ftl_api_key="your_findthatlead_api_key",
        lookup_api_key="your_1lookup_api_key"
    )
    
    # Domain-based prospecting
    print("Starting domain prospecting...")
    prospects = await integration.enhanced_domain_search("example.com", limit=50)
    
    print(f"Found {len(prospects)} qualified prospects")
    
    # Display top prospects
    for prospect in prospects[:5]:
        print(f"
--- Top Prospect ---")
        print(f"Name: {prospect.first_name} {prospect.last_name}")
        print(f"Email: {prospect.email}")
        print(f"Position: {prospect.position}")
        print(f"Phone: {prospect.phone or 'Not found'}")
        print(f"Lead Score: {prospect.lead_score}/100")
        print(f"Email Valid: {prospect.email_validation.get('deliverable') if prospect.email_validation else 'Unknown'}")
        print(f"Phone Valid: {prospect.phone_validation.get('valid') if prospect.phone_validation else 'Unknown'}")
    
    # Export results
    filename = await integration.export_enhanced_prospects(prospects)
    print(f"
Results exported to: {filename}")
    
    # Individual prospect search
    individual_prospect = await integration.enhanced_person_search(
        first_name="John",
        last_name="Smith",
        domain="example.com",
        company="Example Corp"
    )
    
    if individual_prospect:
        print(f"
Individual prospect found:")
        print(f"Lead Score: {individual_prospect.lead_score}/100")
        print(f"Complete Profile: Email={bool(individual_prospect.email)}, Phone={bool(individual_prospect.phone)}")

# Run the integration
if __name__ == "__main__":
    import aiohttp
    asyncio.run(main())