Best Voila Norbert Phone Validation API & Email Verification Integration 2025

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

@dataclass
class EnhancedContact:
    email: str
    first_name: Optional[str] = None
    last_name: Optional[str] = None
    position: Optional[str] = None
    company: Optional[str] = None
    confidence: float = 0.0
    email_validation: Optional[Dict] = None
    phone_data: Optional[Dict] = None
    social_profiles: Optional[Dict] = None
    lead_score: int = 0
    enhancement_status: str = "pending"

class VoilaNorbert1lookupIntegration:
    def __init__(self, voila_api_key: str, lookup_api_key: str):
        self.voila_api_key = voila_api_key
        self.lookup_api_key = lookup_api_key
        self.voila_base_url = "https://api.voilanorbert.com/2018-01-08"
        self.lookup_base_url = "https://api.1lookup.app/v1"
        
    async def find_and_enhance_email(self, first_name: str, last_name: str, 
                                   domain: str, company: str = None) -> EnhancedContact:
        """
        Find email with Voila Norbert and enhance with 1lookup
        """
        try:
            # Step 1: Voila Norbert email finding
            voila_result = await self._voila_find_email(first_name, last_name, domain)
            
            if not voila_result or not voila_result.get('email'):
                return EnhancedContact(
                    email="",
                    first_name=first_name,
                    last_name=last_name,
                    company=company,
                    enhancement_status="email_not_found"
                )
            
            # Step 2: Create base contact
            contact = EnhancedContact(
                email=voila_result['email'],
                first_name=first_name,
                last_name=last_name,
                position=voila_result.get('position'),
                company=company or domain,
                confidence=voila_result.get('score', 0) / 100
            )
            
            # Step 3: 1lookup email validation
            email_validation = await self._validate_email_with_1lookup(contact.email)
            contact.email_validation = email_validation
            
            # Step 4: Phone number enrichment
            phone_data = await self._enrich_phone_data(contact)
            contact.phone_data = phone_data
            
            # Step 5: Social profile enrichment
            social_profiles = await self._enrich_social_profiles(contact)
            contact.social_profiles = social_profiles
            
            # Step 6: Calculate comprehensive lead score
            contact.lead_score = self._calculate_premium_lead_score(contact)
            contact.enhancement_status = "completed"
            
            return contact
            
        except Exception as e:
            return EnhancedContact(
                email="",
                first_name=first_name,
                last_name=last_name,
                company=company,
                enhancement_status=f"error: {str(e)}"
            )
    
    async def _voila_find_email(self, first_name: str, last_name: str, domain: str) -> Dict:
        """Voila Norbert email finding"""
        url = f"{self.voila_base_url}/search/name"
        
        payload = {
            'first_name': first_name,
            'last_name': last_name,
            'domain': domain
        }
        
        headers = {
            'X-Auth-Token': self.voila_api_key,
            'Content-Type': 'application/json'
        }
        
        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 {}
    
    async def _validate_email_with_1lookup(self, email: str) -> Dict:
        """1lookup email validation with advanced checks"""
        url = f"{self.lookup_base_url}/email/validate"
        
        payload = {
            'email': email,
            'enhanced_checks': True,
            'spam_detection': True,
            'social_validation': True
        }
        
        headers = {
            'Authorization': f'Bearer {self.lookup_api_key}',
            'Content-Type': 'application/json'
        }
        
        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, contact: EnhancedContact) -> Optional[Dict]:
        """Phone number enrichment using 1lookup"""
        url = f"{self.lookup_base_url}/enrich/phone"
        
        payload = {
            'first_name': contact.first_name,
            'last_name': contact.last_name,
            'company': contact.company,
            'email': contact.email
        }
        
        headers = {
            'Authorization': f'Bearer {self.lookup_api_key}',
            'Content-Type': 'application/json'
        }
        
        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, contact: EnhancedContact) -> Optional[Dict]:
        """Social profile enrichment"""
        url = f"{self.lookup_base_url}/enrich/social"
        
        payload = {
            'first_name': contact.first_name,
            'last_name': contact.last_name,
            'company': contact.company,
            'email': contact.email
        }
        
        headers = {
            'Authorization': f'Bearer {self.lookup_api_key}',
            'Content-Type': 'application/json'
        }
        
        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_premium_lead_score(self, contact: EnhancedContact) -> int:
        """Calculate comprehensive lead quality score"""
        score = 0
        
        # Voila Norbert confidence (0-35 points)
        voila_confidence = contact.confidence * 100
        score += int(voila_confidence * 0.35)
        
        # Email validation score (0-30 points)
        if contact.email_validation:
            if contact.email_validation.get('deliverable'):
                score += 30
            elif contact.email_validation.get('risky'):
                score += 15
        
        # Phone validation bonus (0-20 points)
        if contact.phone_data and contact.phone_data.get('valid'):
            score += 20
        
        # Social presence bonus (0-10 points)
        if contact.social_profiles and len(contact.social_profiles) > 0:
            score += min(len(contact.social_profiles) * 3, 10)
        
        # Position/seniority bonus (0-5 points)
        if contact.position:
            position_lower = contact.position.lower()
            if any(title in position_lower for title in ['ceo', 'founder', 'president']):
                score += 5
            elif any(title in position_lower for title in ['vp', 'director', 'head']):
                score += 3
        
        return min(score, 100)
    
    async def bulk_enhance_contacts(self, contact_list: List[Dict]) -> List[EnhancedContact]:
        """Process multiple contacts in parallel"""
        tasks = []
        
        for contact_data in contact_list:
            task = self.find_and_enhance_email(
                first_name=contact_data['first_name'],
                last_name=contact_data['last_name'],
                domain=contact_data['domain'],
                company=contact_data.get('company')
            )
            tasks.append(task)
        
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        # Filter out exceptions and return successful results
        enhanced_contacts = []
        for result in results:
            if isinstance(result, EnhancedContact):
                enhanced_contacts.append(result)
        
        return enhanced_contacts

# Example usage
async def main():
    integration = VoilaNorbert1lookupIntegration(
        voila_api_key="your_voila_norbert_api_key",
        lookup_api_key="your_1lookup_api_key"
    )
    
    # Single contact enhancement
    contact = await integration.find_and_enhance_email(
        first_name="John",
        last_name="Smith",
        domain="example.com",
        company="Example Corp"
    )
    
    print(f"Enhanced Contact: {contact.email}")
    print(f"Lead Score: {contact.lead_score}/100")
    print(f"Email Valid: {contact.email_validation.get('deliverable', False)}")
    print(f"Phone Available: {bool(contact.phone_data)}")
    print(f"Enhancement Status: {contact.enhancement_status}")
    
    # Bulk processing example
    contact_list = [
        {'first_name': 'Jane', 'last_name': 'Doe', 'domain': 'company1.com'},
        {'first_name': 'Bob', 'last_name': 'Johnson', 'domain': 'company2.com'},
        # Add more contacts...
    ]
    
    enhanced_contacts = await integration.bulk_enhance_contacts(contact_list)
    
    # Filter high-quality leads
    premium_leads = [c for c in enhanced_contacts if c.lead_score >= 75]
    print(f"Found {len(premium_leads)} premium leads out of {len(enhanced_contacts)} processed")

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