Best Hunter.io Phone Validation API & Email Verification Integration 2025

const express = require('express');
const axios = require('axios');

const app = express();
app.use(express.json());

// Webhook endpoint for Hunter.io integration
app.post('/webhook/hunter-validation', async (req, res) => {
    const { emails, source_domain } = req.body;
    
    const validationResults = await Promise.all(
        emails.map(async (email) => {
            try {
                const response = await axios.post(
                    'https://api.1lookup.app/v1/email/validate',
                    { email: email.value },
                    {
                        headers: {
                            'Authorization': `Bearer ${process.env.LOOKUP_API_KEY}`,
                            'Content-Type': 'application/json'
                        }
                    }
                );
                
                return {
                    email: email.value,
                    hunter_confidence: email.confidence,
                    validation: response.data,
                    deliverable: response.data.deliverable,
                    risk_level: response.data.risk || 'unknown'
                };
            } catch (error) {
                return {
                    email: email.value,
                    error: error.message
                };
            }
        })
    );
    
    // Filter for high-quality leads
    const qualifiedLeads = validationResults.filter(result => 
        result.deliverable && 
        result.hunter_confidence >= 75 &&
        result.risk_level !== 'high'
    );
    
    // Send to your CRM or campaign tool
    await sendToCRM(qualifiedLeads, source_domain);
    
    res.json({
        processed: validationResults.length,
        qualified: qualifiedLeads.length,
        success: true
    });
});

async function sendToCRM(leads, domain) {
    // Integration with your CRM system
    console.log(`Sending ${leads.length} qualified leads from ${domain} to CRM`);
}

import pandas as pd
import asyncio
import aiohttp
from typing import List, Dict

async def process_hunter_csv(csv_file: str, batch_size: int = 100):
    """
    Process Hunter.io CSV export with 1lookup validation
    """
    # Load Hunter.io CSV export
    df = pd.read_csv(csv_file)
    
    # Prepare for batch processing
    total_rows = len(df)
    processed = 0
    results = []
    
    # Process in batches
    for i in range(0, total_rows, batch_size):
        batch = df.iloc[i:i+batch_size]
        batch_results = await validate_email_batch(batch.to_dict('records'))
        results.extend(batch_results)
        processed += len(batch)
        
        print(f"Processed {processed}/{total_rows} contacts")
    
    # Create results DataFrame
    results_df = pd.DataFrame(results)
    
    # Add lead scoring
    results_df['lead_score'] = results_df.apply(calculate_comprehensive_score, axis=1)
    
    # Filter high-quality leads
    high_quality = results_df[results_df['lead_score'] >= 70]
    
    # Export results
    results_df.to_csv('hunter_validated_all.csv', index=False)
    high_quality.to_csv('hunter_qualified_leads.csv', index=False)
    
    return {
        'total_processed': len(results_df),
        'high_quality_leads': len(high_quality),
        'average_score': results_df['lead_score'].mean(),
        'deliverable_rate': (results_df['deliverable'] == True).sum() / len(results_df)
    }

async def validate_email_batch(contacts: List[Dict]) -> List[Dict]:
    """Validate batch of emails with 1lookup API"""
    async with aiohttp.ClientSession() as session:
        tasks = []
        
        for contact in contacts:
            task = validate_single_contact(session, contact)
            tasks.append(task)
        
        return await asyncio.gather(*tasks)

async def validate_single_contact(session: aiohttp.ClientSession, contact: Dict) -> Dict:
    """Validate single contact"""
    email = contact.get('email', '')
    
    try:
        # Email validation
        async with session.post(
            'https://api.1lookup.app/v1/email/validate',
            json={'email': email},
            headers={
                'Authorization': f'Bearer {LOOKUP_API_KEY}',
                'Content-Type': 'application/json'
            }
        ) as response:
            email_result = await response.json()
        
        # Phone enrichment if first/last name available
        phone_result = None
        if contact.get('first_name') and contact.get('last_name'):
            phone_result = await enrich_phone(session, contact)
        
        return {
            **contact,
            'email_validation': email_result,
            'phone_data': phone_result,
            'deliverable': email_result.get('deliverable', False),
            'risk_level': email_result.get('risk', 'unknown')
        }
    
    except Exception as e:
        return {
            **contact,
            'error': str(e),
            'deliverable': False
        }

def calculate_comprehensive_score(row) -> int:
    """Calculate comprehensive lead score"""
    score = 0
    
    # Email deliverability (40 points)
    if row.get('deliverable'):
        score += 40
    elif row.get('risk_level') == 'low':
        score += 30
    elif row.get('risk_level') == 'medium':
        score += 15
    
    # Hunter confidence (30 points)
    hunter_conf = row.get('confidence', 0)
    score += int(hunter_conf * 0.3)
    
    # Phone validation bonus (20 points)
    phone_data = row.get('phone_data')
    if phone_data and phone_data.get('valid'):
        score += 20
    
    # Position/seniority (10 points)
    position = str(row.get('position', '')).lower()
    if any(title in position for title in ['ceo', 'founder', 'cto', 'cfo']):
        score += 10
    elif any(title in position for title in ['vp', 'director', 'head']):
        score += 7
    elif any(title in position for title in ['manager', 'lead', 'senior']):
        score += 4
    
    return min(score, 100)

# Run batch processing
results = asyncio.run(process_hunter_csv('hunter_export.csv'))
print(f"Processed {results['total_processed']} contacts")
print(f"Found {results['high_quality_leads']} high-quality leads")