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

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

@dataclass
class EnhancedCampaignContact:
    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
    email_validation: Optional[Dict] = None
    phone_validation: Optional[Dict] = None
    deliverability_score: int = 0
    lead_score: int = 0
    campaign_readiness: str = "pending"
    enhancement_timestamp: str = ""

class SnovIO1lookupIntegration:
    def __init__(self, snov_user_id: str, snov_api_key: str, lookup_api_key: str):
        self.snov_user_id = snov_user_id
        self.snov_api_key = snov_api_key
        self.lookup_api_key = lookup_api_key
        self.snov_base_url = "https://api.snov.io/v1"
        self.lookup_base_url = "https://api.1lookup.app/v1"
    
    async def enhance_campaign_list(self, campaign_id: str) -> Dict[str, any]:
        """
        Enhance entire Snov.io campaign list with comprehensive validation
        """
        try:
            # Step 1: Get campaign contacts from Snov.io
            campaign_contacts = await self._get_snov_campaign_contacts(campaign_id)
            
            if not campaign_contacts:
                return {"error": "No contacts found in campaign", "enhanced_contacts": []}
            
            # Step 2: Enhance all contacts in parallel
            enhanced_contacts = await self._bulk_enhance_contacts(campaign_contacts)
            
            # Step 3: Filter campaign-ready contacts
            campaign_ready = [c for c in enhanced_contacts if c.campaign_readiness == "ready"]
            
            # Step 4: Update Snov.io campaign with enhanced data
            await self._update_snov_campaign(campaign_id, enhanced_contacts)
            
            return {
                "campaign_id": campaign_id,
                "total_contacts": len(campaign_contacts),
                "enhanced_contacts": enhanced_contacts,
                "campaign_ready": len(campaign_ready),
                "enhancement_stats": self._calculate_enhancement_stats(enhanced_contacts)
            }
            
        except Exception as e:
            return {"error": f"Campaign enhancement failed: {e}"}
    
    async def optimize_email_sequence(self, campaign_id: str, sequence_data: Dict) -> Dict:
        """
        Optimize Snov.io email sequence with deliverability intelligence
        """
        try:
            # Get campaign contacts
            contacts = await self._get_snov_campaign_contacts(campaign_id)
            
            # Validate all email addresses
            validation_tasks = [
                self._validate_email_deliverability(contact['email'])
                for contact in contacts if contact.get('email')
            ]
            
            validation_results = await asyncio.gather(*validation_tasks)
            
            # Segment contacts by deliverability
            high_deliverability = []
            medium_deliverability = []
            low_deliverability = []
            
            for i, result in enumerate(validation_results):
                contact = contacts[i]
                if result.get('deliverability_score', 0) >= 80:
                    high_deliverability.append(contact)
                elif result.get('deliverability_score', 0) >= 60:
                    medium_deliverability.append(contact)
                else:
                    low_deliverability.append(contact)
            
            # Create optimized sequences for each segment
            optimized_sequences = {
                "high_deliverability": {
                    "contacts": high_deliverability,
                    "sequence_timing": "aggressive",  # Shorter intervals
                    "personalization_level": "high"
                },
                "medium_deliverability": {
                    "contacts": medium_deliverability,
                    "sequence_timing": "moderate",  # Standard intervals
                    "personalization_level": "medium"
                },
                "low_deliverability": {
                    "contacts": low_deliverability,
                    "sequence_timing": "conservative",  # Longer intervals
                    "personalization_level": "basic",
                    "recommendation": "Consider phone/LinkedIn outreach"
                }
            }
            
            return {
                "campaign_id": campaign_id,
                "total_contacts": len(contacts),
                "segmentation": {
                    "high_deliverability": len(high_deliverability),
                    "medium_deliverability": len(medium_deliverability),
                    "low_deliverability": len(low_deliverability)
                },
                "optimized_sequences": optimized_sequences
            }
            
        except Exception as e:
            return {"error": f"Sequence optimization failed: {e}"}
    
    async def _get_snov_campaign_contacts(self, campaign_id: str) -> List[Dict]:
        """Get contacts from Snov.io campaign"""
        url = f"{self.snov_base_url}/get-prospects-from-campaign"
        
        payload = {
            'userId': self.snov_user_id,
            'apiKey': self.snov_api_key,
            'campaignId': campaign_id
        }
        
        async with aiohttp.ClientSession() as session:
            async with session.post(url, json=payload) as response:
                if response.status == 200:
                    data = await response.json()
                    return data.get('prospects', [])
                return []
    
    async def _bulk_enhance_contacts(self, contacts: List[Dict]) -> List[EnhancedCampaignContact]:
        """Enhance multiple contacts in parallel"""
        enhancement_tasks = []
        
        for contact_data in contacts:
            enhanced_contact = EnhancedCampaignContact(
                email=contact_data.get('email', ''),
                first_name=contact_data.get('firstName'),
                last_name=contact_data.get('lastName'),
                position=contact_data.get('position'),
                company=contact_data.get('company'),
                domain=contact_data.get('domain'),
                linkedin_url=contact_data.get('linkedinUrl'),
                enhancement_timestamp=datetime.now().isoformat()
            )
            
            task = self._enhance_single_contact(enhanced_contact)
            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_contact(self, contact: EnhancedCampaignContact) -> None:
        """Comprehensive single contact enhancement"""
        enhancement_tasks = []
        
        # Email validation
        if contact.email:
            task1 = self._validate_email_deliverability(contact.email)
            enhancement_tasks.append(task1)
        
        # Phone enrichment
        if contact.first_name and contact.last_name:
            task2 = self._enrich_phone_data(contact)
            enhancement_tasks.append(task2)
        
        # Execute enhancements
        results = await asyncio.gather(*enhancement_tasks, return_exceptions=True)
        
        # Process email validation
        if len(results) > 0 and not isinstance(results[0], Exception):
            contact.email_validation = results[0]
            contact.deliverability_score = results[0].get('deliverability_score', 0)
        
        # Process phone enrichment
        if len(results) > 1 and not isinstance(results[1], Exception) and results[1]:
            contact.phone_validation = results[1]
            contact.phone = results[1].get('phone_number')
        
        # Calculate lead score and campaign readiness
        contact.lead_score = self._calculate_campaign_lead_score(contact)
        contact.campaign_readiness = self._determine_campaign_readiness(contact)
    
    async def _validate_email_deliverability(self, email: str) -> Dict:
        """Advanced email deliverability validation"""
        url = f"{self.lookup_base_url}/email/validate"
        headers = {
            'Authorization': f'Bearer {self.lookup_api_key}',
            'Content-Type': 'application/json'
        }
        
        payload = {
            'email': email,
            'deliverability_check': True,
            'spam_trap_detection': True,
            'bounce_risk_assessment': True,
            'reputation_check': True
        }
        
        async with aiohttp.ClientSession() as session:
            async with session.post(url, json=payload, headers=headers) as response:
                if response.status == 200:
                    result = await response.json()
                    # Calculate deliverability score
                    score = 0
                    if result.get('deliverable'):
                        score += 40
                    if result.get('valid_format'):
                        score += 20
                    if not result.get('spam_trap'):
                        score += 20
                    if result.get('reputation') == 'good':
                        score += 20
                    
                    result['deliverability_score'] = score
                    return result
                return {'deliverable': False, 'deliverability_score': 0}
    
    async def _enrich_phone_data(self, contact: EnhancedCampaignContact) -> Optional[Dict]:
        """Phone number enrichment for multi-channel campaigns"""
        url = f"{self.lookup_base_url}/enrich/phone"
        headers = {
            'Authorization': f'Bearer {self.lookup_api_key}',
            'Content-Type': 'application/json'
        }
        
        payload = {
            'first_name': contact.first_name,
            'last_name': contact.last_name,
            'company': contact.company,
            'domain': contact.domain,
            'email': contact.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
    
    def _calculate_campaign_lead_score(self, contact: EnhancedCampaignContact) -> int:
        """Calculate campaign-optimized lead score"""
        score = 0
        
        # Email deliverability (0-40 points)
        score += contact.deliverability_score * 0.4
        
        # Contact completeness (0-30 points)
        completeness = 0
        if contact.email: completeness += 10
        if contact.phone: completeness += 10
        if contact.position: completeness += 5
        if contact.company: completeness += 5
        score += completeness
        
        # Phone validation bonus (0-20 points)
        if contact.phone_validation and contact.phone_validation.get('valid'):
            score += 20
        
        # Position/seniority bonus (0-10 points)
        if contact.position:
            position_lower = contact.position.lower()
            if any(title in position_lower for title in ['ceo', 'founder', 'president']):
                score += 10
            elif any(title in position_lower for title in ['vp', 'director', 'head']):
                score += 7
            elif any(title in position_lower for title in ['manager', 'lead', 'senior']):
                score += 4
        
        return min(int(score), 100)
    
    def _determine_campaign_readiness(self, contact: EnhancedCampaignContact) -> str:
        """Determine if contact is ready for campaign"""
        if contact.deliverability_score >= 80 and contact.email:
            return "ready"
        elif contact.deliverability_score >= 60:
            return "caution"
        else:
            return "not_ready"
    
    async def _update_snov_campaign(self, campaign_id: str, contacts: List[EnhancedCampaignContact]):
        """Update Snov.io campaign with enhanced contact data"""
        url = f"{self.snov_base_url}/update-prospects-in-campaign"
        
        # Prepare contacts for Snov.io format
        snov_contacts = []
        for contact in contacts:
            snov_contact = {
                'email': contact.email,
                'firstName': contact.first_name,
                'lastName': contact.last_name,
                'position': contact.position,
                'company': contact.company,
                'phone': contact.phone,
                'leadScore': contact.lead_score,
                'deliverabilityScore': contact.deliverability_score,
                'campaignReadiness': contact.campaign_readiness
            }
            snov_contacts.append(snov_contact)
        
        payload = {
            'userId': self.snov_user_id,
            'apiKey': self.snov_api_key,
            'campaignId': campaign_id,
            'prospects': snov_contacts
        }
        
        async with aiohttp.ClientSession() as session:
            async with session.post(url, json=payload) as response:
                return response.status == 200
    
    def _calculate_enhancement_stats(self, contacts: List[EnhancedCampaignContact]) -> Dict:
        """Calculate enhancement statistics"""
        total = len(contacts)
        if total == 0:
            return {}
        
        deliverable_emails = sum(1 for c in contacts if c.deliverability_score >= 80)
        valid_phones = sum(1 for c in contacts if c.phone_validation and c.phone_validation.get('valid'))
        high_score_leads = sum(1 for c in contacts if c.lead_score >= 70)
        campaign_ready = sum(1 for c in contacts if c.campaign_readiness == "ready")
        
        return {
            'total_contacts': total,
            'deliverable_email_rate': (deliverable_emails / total) * 100,
            'phone_enrichment_rate': (valid_phones / total) * 100,
            'high_quality_leads_rate': (high_score_leads / total) * 100,
            'campaign_ready_rate': (campaign_ready / total) * 100,
            'average_lead_score': sum(c.lead_score for c in contacts) / total
        }

# Example usage
async def main():
    integration = SnovIO1lookupIntegration(
        snov_user_id="your_snov_user_id",
        snov_api_key="your_snov_api_key",
        lookup_api_key="your_1lookup_api_key"
    )
    
    # Enhance existing campaign
    campaign_result = await integration.enhance_campaign_list("your_campaign_id")
    
    print(f"Campaign Enhancement Results:")
    print(f"Total Contacts: {campaign_result['total_contacts']}")
    print(f"Campaign Ready: {campaign_result['campaign_ready']}")
    
    stats = campaign_result.get('enhancement_stats', {})
    print(f"
Enhancement Statistics:")
    print(f"Deliverable Email Rate: {stats.get('deliverable_email_rate', 0):.1f}%")
    print(f"Phone Enrichment Rate: {stats.get('phone_enrichment_rate', 0):.1f}%")
    print(f"Campaign Ready Rate: {stats.get('campaign_ready_rate', 0):.1f}%")
    print(f"Average Lead Score: {stats.get('average_lead_score', 0):.1f}")
    
    # Optimize email sequence
    sequence_result = await integration.optimize_email_sequence("your_campaign_id", {})
    
    print(f"
Sequence Optimization:")
    segmentation = sequence_result.get('segmentation', {})
    print(f"High Deliverability: {segmentation.get('high_deliverability', 0)} contacts")
    print(f"Medium Deliverability: {segmentation.get('medium_deliverability', 0)} contacts")
    print(f"Low Deliverability: {segmentation.get('low_deliverability', 0)} contacts")

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