Best Instantly.ai 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 DeliverabilityOptimizedContact:
    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_intelligence: Optional[Dict] = None
    reputation_data: Optional[Dict] = None
    deliverability_score: int = 0
    reputation_score: int = 0
    cold_email_score: int = 0
    campaign_priority: str = "medium"
    deliverability_status: str = "unknown"
    reputation_risk: str = "low"
    instantly_optimization: Optional[Dict] = None
    enhancement_timestamp: str = ""

class InstantlyAI1lookupIntegration:
    def __init__(self, instantly_api_key: str, lookup_api_key: str):
        self.instantly_api_key = instantly_api_key
        self.lookup_api_key = lookup_api_key
        self.instantly_base_url = "https://api.instantly.ai/v1"
        self.lookup_base_url = "https://api.1lookup.app/v1"
    
    async def optimize_deliverability_campaign(self, campaign_id: str) -> Dict:
        """
        Optimize Instantly.ai campaign with comprehensive deliverability intelligence
        """
        try:
            # Step 1: Get campaign leads from Instantly.ai
            campaign_leads = await self._get_instantly_campaign_leads(campaign_id)
            
            if not campaign_leads:
                return {"error": "No leads found in campaign"}
            
            # Step 2: Enhance all leads with deliverability intelligence
            enhanced_leads = await self._bulk_enhance_for_deliverability(campaign_leads)
            
            # Step 3: Generate deliverability optimization strategies
            deliverability_strategies = await self._generate_deliverability_strategies(enhanced_leads)
            
            # Step 4: Create reputation protection recommendations
            reputation_protection = self._generate_reputation_protection(enhanced_leads)
            
            # Step 5: Update Instantly.ai campaign with enhanced data
            update_results = await self._update_instantly_campaign(campaign_id, enhanced_leads)
            
            return {
                "campaign_id": campaign_id,
                "total_leads": len(campaign_leads),
                "enhanced_leads": len(enhanced_leads),
                "deliverability_stats": self._calculate_deliverability_stats(enhanced_leads),
                "deliverability_strategies": deliverability_strategies,
                "reputation_protection": reputation_protection,
                "update_results": update_results
            }
            
        except Exception as e:
            return {"error": f"Deliverability campaign optimization failed: {e}"}
    
    async def create_perfect_deliverability_campaign(self, campaign_name: str,
                                                   leads: List[Dict],
                                                   campaign_config: Dict = None) -> Dict:
        """
        Create new Instantly.ai campaign with perfect deliverability optimization
        """
        try:
            # Step 1: Enhance leads with comprehensive deliverability intelligence
            enhanced_leads = await self._bulk_enhance_for_deliverability(leads)
            
            # Step 2: Generate optimal deliverability campaign structure
            campaign_structure = await self._generate_deliverability_campaign_structure(
                enhanced_leads, campaign_config
            )
            
            # Step 3: Create Instantly.ai campaign with deliverability focus
            campaign_data = await self._create_instantly_campaign(campaign_name, campaign_structure)
            
            if not campaign_data:
                return {"error": "Failed to create Instantly.ai campaign"}
            
            # Step 4: Apply perfect deliverability optimizations
            optimization_results = await self._apply_perfect_deliverability_optimizations(
                campaign_data['id'], enhanced_leads
            )
            
            return {
                "campaign_id": campaign_data['id'],
                "campaign_name": campaign_name,
                "total_leads": len(enhanced_leads),
                "deliverability_optimization": self._analyze_deliverability_optimization(enhanced_leads),
                "optimization_results": optimization_results
            }
            
        except Exception as e:
            return {"error": f"Perfect deliverability campaign creation failed: {e}"}
    
    async def _bulk_enhance_for_deliverability(self, leads: List[Dict]) -> List[DeliverabilityOptimizedContact]:
        """Enhance multiple leads for optimal deliverability campaigns"""
        enhancement_tasks = []
        
        for lead_data in leads:
            deliverability_contact = DeliverabilityOptimizedContact(
                email=lead_data.get('email', ''),
                first_name=lead_data.get('first_name') or lead_data.get('firstName'),
                last_name=lead_data.get('last_name') or lead_data.get('lastName'),
                position=lead_data.get('position') or lead_data.get('title'),
                company=lead_data.get('company'),
                domain=lead_data.get('domain'),
                linkedin_url=lead_data.get('linkedin_url'),
                enhancement_timestamp=datetime.now().isoformat()
            )
            
            task = self._enhance_single_contact_for_deliverability(deliverability_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_for_deliverability(self, contact: DeliverabilityOptimizedContact) -> None:
        """Comprehensive single contact enhancement for deliverability optimization"""
        enhancement_tasks = []
        
        # Email validation with deliverability focus
        if contact.email:
            task1 = self._validate_email_for_perfect_deliverability(contact.email)
            enhancement_tasks.append(task1)
        
        # Phone enrichment for campaign enhancement
        if contact.first_name and contact.last_name:
            task2 = self._enrich_phone_for_deliverability_campaigns(contact)
            enhancement_tasks.append(task2)
        
        # Deliverability intelligence gathering
        task3 = self._gather_deliverability_intelligence(contact)
        enhancement_tasks.append(task3)
        
        # Execute all enhancements
        results = await asyncio.gather(*enhancement_tasks, return_exceptions=True)
        
        # Process email validation and deliverability analysis
        if len(results) > 0 and not isinstance(results[0], Exception):
            contact.email_validation = results[0]
            contact.deliverability_score = results[0].get('deliverability_score', 0)
            contact.reputation_data = results[0].get('reputation_analysis', {})
        
        # 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')
        
        # Process deliverability intelligence
        if len(results) > 2 and not isinstance(results[2], Exception):
            contact.deliverability_intelligence = results[2]
        
        # Calculate deliverability scores and classifications
        contact.reputation_score = self._calculate_reputation_score(contact)
        contact.cold_email_score = self._calculate_cold_email_deliverability_score(contact)
        contact.campaign_priority = self._determine_deliverability_priority(contact)
        contact.deliverability_status = self._assess_deliverability_status(contact)
        contact.reputation_risk = self._assess_reputation_risk(contact)
        contact.instantly_optimization = self._generate_instantly_optimization_recommendations(contact)
    
    async def _validate_email_for_perfect_deliverability(self, email: str) -> Dict:
        """Advanced email validation optimized for perfect deliverability"""
        url = f"{self.lookup_base_url}/email/validate"
        headers = {
            'Authorization': f'Bearer {self.lookup_api_key}',
            'Content-Type': 'application/json'
        }
        
        payload = {
            'email': email,
            'perfect_deliverability_focus': True,
            'reputation_analysis': True,
            'inbox_placement_prediction': True,
            'spam_risk_assessment': True,
            'sender_reputation_impact': 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 perfect deliverability score
                    score = 0
                    if result.get('deliverable'): score += 30
                    if result.get('valid_format'): score += 15
                    if not result.get('spam_trap'): score += 20
                    if result.get('inbox_placement_likelihood', 0) > 80: score += 20
                    if result.get('reputation_safe', True): score += 15
                    
                    result['deliverability_score'] = score
                    result['perfect_deliverability_rating'] = 'perfect' if score >= 95 else 'excellent' if score >= 85 else 'good' if score >= 70 else 'needs_improvement'
                    
                    # Reputation analysis
                    result['reputation_analysis'] = {
                        'domain_reputation': result.get('domain_reputation', 'unknown'),
                        'spam_risk_level': result.get('spam_risk', 'low'),
                        'blacklist_status': result.get('blacklist_check', 'clean'),
                        'sender_impact': result.get('sender_reputation_impact', 'positive')
                    }
                    
                    return result
                return {'deliverable': False, 'deliverability_score': 0, 'perfect_deliverability_rating': 'poor'}
    
    async def _enrich_phone_for_deliverability_campaigns(self, contact: DeliverabilityOptimizedContact) -> Optional[Dict]:
        """Phone enrichment optimized for deliverability-focused 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,
            'deliverability_focus': True,
            'campaign_enhancement': True
        }
        
        try:
            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()
                        
                        # Assess deliverability campaign enhancement value
                        if result.get('valid'):
                            result['deliverability_enhancement_score'] = 90 if result.get('business_phone') else 75
                            result['campaign_enhancement_value'] = 'high'
                        else:
                            result['deliverability_enhancement_score'] = 0
                            result['campaign_enhancement_value'] = 'low'
                        
                        return result
        except:
            pass
        
        return None
    
    async def _gather_deliverability_intelligence(self, contact: DeliverabilityOptimizedContact) -> Optional[Dict]:
        """Gather comprehensive deliverability intelligence"""
        url = f"{self.lookup_base_url}/enrich/deliverability"
        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,
            'email': contact.email,
            'domain': contact.domain,
            'deliverability_optimization': True,
            'reputation_intelligence': True
        }
        
        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_reputation_score(self, contact: DeliverabilityOptimizedContact) -> int:
        """Calculate sender reputation protection score"""
        score = 100  # Start with perfect score
        
        # Deduct points for reputation risks
        if contact.reputation_data:
            if contact.reputation_data.get('spam_risk_level') == 'high':
                score -= 30
            elif contact.reputation_data.get('spam_risk_level') == 'medium':
                score -= 15
            
            if contact.reputation_data.get('blacklist_status') != 'clean':
                score -= 25
            
            if contact.reputation_data.get('domain_reputation') == 'poor':
                score -= 20
            elif contact.reputation_data.get('domain_reputation') == 'fair':
                score -= 10
        
        return max(score, 0)
    
    def _calculate_cold_email_deliverability_score(self, contact: DeliverabilityOptimizedContact) -> int:
        """Calculate comprehensive cold email deliverability score"""
        score = 0
        
        # Perfect deliverability foundation (0-50 points)
        if contact.deliverability_score:
            score += contact.deliverability_score * 0.5
        
        # Reputation protection (0-25 points)
        if contact.reputation_score:
            score += contact.reputation_score * 0.25
        
        # Contact quality for personalization (0-15 points)
        if contact.first_name and contact.last_name: score += 5
        if contact.company and contact.position: score += 5
        if contact.phone_validation and contact.phone_validation.get('valid'): score += 5
        
        # Deliverability intelligence bonus (0-10 points)
        if contact.deliverability_intelligence:
            intel_quality = len([v for v in contact.deliverability_intelligence.values() if v])
            score += min(intel_quality * 2, 10)
        
        return min(int(score), 100)
    
    def _determine_deliverability_priority(self, contact: DeliverabilityOptimizedContact) -> str:
        """Determine campaign priority based on deliverability score"""
        if contact.cold_email_score >= 90:
            return "perfect"
        elif contact.cold_email_score >= 80:
            return "excellent"
        elif contact.cold_email_score >= 65:
            return "good"
        elif contact.cold_email_score >= 50:
            return "fair"
        else:
            return "risky"
    
    def _assess_deliverability_status(self, contact: DeliverabilityOptimizedContact) -> str:
        """Assess overall deliverability status"""
        if contact.email_validation:
            rating = contact.email_validation.get('perfect_deliverability_rating', 'poor')
            return rating
        return "unknown"
    
    def _assess_reputation_risk(self, contact: DeliverabilityOptimizedContact) -> str:
        """Assess sender reputation risk level"""
        if contact.reputation_score >= 90:
            return "very_low"
        elif contact.reputation_score >= 80:
            return "low"
        elif contact.reputation_score >= 60:
            return "medium"
        elif contact.reputation_score >= 40:
            return "high"
        else:
            return "very_high"
    
    def _generate_instantly_optimization_recommendations(self, contact: DeliverabilityOptimizedContact) -> Dict:
        """Generate Instantly.ai-specific optimization recommendations"""
        optimization = {
            'warm_up_strategy': 'standard',
            'sending_schedule': 'normal',
            'reputation_protection': {},
            'personalization_settings': {},
            'campaign_settings': {},
            'deliverability_monitoring': {}
        }
        
        # Warm-up strategy based on deliverability score
        if contact.cold_email_score >= 90:
            optimization['warm_up_strategy'] = 'minimal'
        elif contact.cold_email_score >= 75:
            optimization['warm_up_strategy'] = 'standard'
        elif contact.cold_email_score >= 60:
            optimization['warm_up_strategy'] = 'extended'
        else:
            optimization['warm_up_strategy'] = 'intensive'
        
        # Sending schedule optimization
        if contact.reputation_risk in ['very_low', 'low']:
            optimization['sending_schedule'] = 'aggressive'
        elif contact.reputation_risk == 'medium':
            optimization['sending_schedule'] = 'normal'
        else:
            optimization['sending_schedule'] = 'conservative'
        
        # Reputation protection settings
        optimization['reputation_protection'] = {
            'spam_check_enabled': True,
            'domain_reputation_monitoring': True,
            'bounce_rate_alerts': True,
            'reputation_score_tracking': True,
            'risk_level': contact.reputation_risk
        }
        
        # Personalization settings for deliverability
        optimization['personalization_settings'] = {
            'personalization_level': 'high' if contact.cold_email_score >= 80 else 'medium',
            'dynamic_content': contact.cold_email_score >= 70,
            'custom_variables': bool(contact.deliverability_intelligence),
            'ai_personalization': contact.cold_email_score >= 75
        }
        
        # Campaign settings optimization
        optimization['campaign_settings'] = {
            'send_limit_per_day': self._calculate_optimal_send_limit(contact),
            'delay_between_emails': self._calculate_optimal_delay(contact),
            'follow_up_sequence_length': min(3 + (contact.cold_email_score // 25), 6),
            'deliverability_monitoring': True
        }
        
        # Deliverability monitoring configuration
        optimization['deliverability_monitoring'] = {
            'real_time_tracking': True,
            'bounce_rate_alerts': True,
            'spam_folder_monitoring': True,
            'reputation_score_alerts': contact.reputation_risk in ['medium', 'high', 'very_high'],
            'automatic_pause_triggers': contact.reputation_risk in ['high', 'very_high']
        }
        
        return optimization
    
    def _calculate_optimal_send_limit(self, contact: DeliverabilityOptimizedContact) -> int:
        """Calculate optimal daily send limit based on deliverability score"""
        base_limit = 50  # Conservative baseline
        
        if contact.cold_email_score >= 90:
            return base_limit * 3  # 150 emails/day
        elif contact.cold_email_score >= 80:
            return base_limit * 2  # 100 emails/day
        elif contact.cold_email_score >= 70:
            return int(base_limit * 1.5)  # 75 emails/day
        else:
            return base_limit  # 50 emails/day
    
    def _calculate_optimal_delay(self, contact: DeliverabilityOptimizedContact) -> int:
        """Calculate optimal delay between emails in minutes"""
        if contact.reputation_risk in ['very_low', 'low']:
            return 30  # 30 minutes
        elif contact.reputation_risk == 'medium':
            return 60  # 1 hour
        else:
            return 120  # 2 hours
    
    def _calculate_deliverability_stats(self, contacts: List[DeliverabilityOptimizedContact]) -> Dict:
        """Calculate comprehensive deliverability statistics"""
        total = len(contacts)
        if total == 0:
            return {}
        
        perfect_deliverability = sum(1 for c in contacts if c.deliverability_status == 'perfect')
        excellent_deliverability = sum(1 for c in contacts if c.deliverability_status in ['perfect', 'excellent'])
        low_reputation_risk = sum(1 for c in contacts if c.reputation_risk in ['very_low', 'low'])
        high_cold_email_score = sum(1 for c in contacts if c.cold_email_score >= 80)
        phone_enhanced = sum(1 for c in contacts if c.phone_validation and c.phone_validation.get('valid'))
        
        return {
            'total_contacts': total,
            'perfect_deliverability_rate': (perfect_deliverability / total) * 100,
            'excellent_deliverability_rate': (excellent_deliverability / total) * 100,
            'low_reputation_risk_rate': (low_reputation_risk / total) * 100,
            'high_cold_email_score_rate': (high_cold_email_score / total) * 100,
            'phone_enhanced_rate': (phone_enhanced / total) * 100,
            'average_deliverability_score': sum(c.deliverability_score for c in contacts) / total,
            'average_reputation_score': sum(c.reputation_score for c in contacts) / total,
            'average_cold_email_score': sum(c.cold_email_score for c in contacts) / total,
            'perfect_priority_contacts': sum(1 for c in contacts if c.campaign_priority == 'perfect')
        }

# Example usage for Instantly.ai perfect deliverability optimization
async def main():
    integration = InstantlyAI1lookupIntegration(
        instantly_api_key="your_instantly_ai_api_key",
        lookup_api_key="your_1lookup_api_key"
    )
    
    # Sample leads for perfect deliverability optimization
    sample_leads = [
        {
            'email': 'john@example.com',
            'first_name': 'John',
            'last_name': 'Smith',
            'position': 'Marketing Director',
            'company': 'Example Corp',
            'domain': 'example.com'
        },
        # Add more leads...
    ]
    
    # Create perfect deliverability campaign
    campaign_result = await integration.create_perfect_deliverability_campaign(
        campaign_name="Perfect Deliverability Cold Outreach",
        leads=sample_leads,
        campaign_config={
            "deliverability_focus": "maximum",
            "reputation_protection": "advanced",
            "personalization_level": "deep"
        }
    )
    
    print("Instantly.ai Perfect Deliverability Results:")
    print(f"Campaign ID: {campaign_result.get('campaign_id')}")
    print(f"Total Leads: {campaign_result.get('total_leads')}")
    
    deliverability_optimization = campaign_result.get('deliverability_optimization', {})
    print(f"
Deliverability Optimization:")
    print(f"Perfect Deliverability Rate: {deliverability_optimization.get('perfect_deliverability_rate', 0):.1f}%")
    print(f"Excellent Deliverability Rate: {deliverability_optimization.get('excellent_deliverability_rate', 0):.1f}%")
    print(f"Low Reputation Risk Rate: {deliverability_optimization.get('low_reputation_risk_rate', 0):.1f}%")
    print(f"Average Deliverability Score: {deliverability_optimization.get('average_deliverability_score', 0):.1f}")
    print(f"Average Reputation Score: {deliverability_optimization.get('average_reputation_score', 0):.1f}")

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