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

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

class ContactChannel(Enum):
    EMAIL = "email"
    PHONE = "phone"
    LINKEDIN = "linkedin"
    SMS = "sms"

@dataclass
class MultichannelContact:
    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
    linkedin_validation: Optional[Dict] = None
    channel_preferences: Optional[Dict] = None
    engagement_score: int = 0
    preferred_channels: List[str] = None
    channel_priority: Dict[str, int] = None
    multichannel_readiness: str = "pending"
    enhancement_timestamp: str = ""

class ReplyIO1lookupIntegration:
    def __init__(self, reply_api_key: str, lookup_api_key: str):
        self.reply_api_key = reply_api_key
        self.lookup_api_key = lookup_api_key
        self.reply_base_url = "https://api.reply.io/v1"
        self.lookup_base_url = "https://api.1lookup.app/v1"
    
    async def optimize_multichannel_campaign(self, campaign_id: str) -> Dict:
        """
        Optimize existing Reply.io campaign with comprehensive multichannel intelligence
        """
        try:
            # Step 1: Get campaign contacts from Reply.io
            campaign_contacts = await self._get_reply_campaign_contacts(campaign_id)
            
            if not campaign_contacts:
                return {"error": "No contacts found in campaign"}
            
            # Step 2: Enhance all contacts with multichannel intelligence
            enhanced_contacts = await self._bulk_enhance_for_multichannel(campaign_contacts)
            
            # Step 3: Optimize channel sequences based on contact intelligence
            sequence_optimization = await self._optimize_channel_sequences(campaign_id, enhanced_contacts)
            
            # Step 4: Update Reply.io campaign with enhanced data and sequences
            update_results = await self._update_reply_campaign(campaign_id, enhanced_contacts, sequence_optimization)
            
            return {
                "campaign_id": campaign_id,
                "total_contacts": len(campaign_contacts),
                "enhanced_contacts": len(enhanced_contacts),
                "multichannel_stats": self._calculate_multichannel_stats(enhanced_contacts),
                "sequence_optimization": sequence_optimization,
                "update_results": update_results
            }
            
        except Exception as e:
            return {"error": f"Campaign optimization failed: {e}"}
    
    async def create_optimized_multichannel_campaign(self, campaign_name: str, 
                                                   contacts: List[Dict],
                                                   sequence_template: str = None) -> Dict:
        """
        Create new Reply.io campaign with optimized multichannel sequences
        """
        try:
            # Step 1: Enhance all contacts with comprehensive intelligence
            enhanced_contacts = await self._bulk_enhance_for_multichannel(contacts)
            
            # Step 2: Generate optimal multichannel sequences
            optimal_sequences = await self._generate_optimal_sequences(enhanced_contacts, sequence_template)
            
            # Step 3: Create Reply.io campaign with optimized setup
            campaign_data = await self._create_reply_campaign(campaign_name, optimal_sequences)
            
            if not campaign_data:
                return {"error": "Failed to create Reply.io campaign"}
            
            # Step 4: Add enhanced contacts with channel prioritization
            await self._add_optimized_contacts_to_campaign(campaign_data['id'], enhanced_contacts)
            
            return {
                "campaign_id": campaign_data['id'],
                "campaign_name": campaign_name,
                "total_contacts": len(enhanced_contacts),
                "multichannel_optimization": self._analyze_multichannel_optimization(enhanced_contacts),
                "sequence_structure": optimal_sequences
            }
            
        except Exception as e:
            return {"error": f"Campaign creation failed: {e}"}
    
    async def _bulk_enhance_for_multichannel(self, contacts: List[Dict]) -> List[MultichannelContact]:
        """Enhance multiple contacts for optimal multichannel engagement"""
        enhancement_tasks = []
        
        for contact_data in contacts:
            multichannel_contact = MultichannelContact(
                email=contact_data.get('email', ''),
                first_name=contact_data.get('firstName') or contact_data.get('first_name'),
                last_name=contact_data.get('lastName') or contact_data.get('last_name'),
                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_for_multichannel(multichannel_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_multichannel(self, contact: MultichannelContact) -> None:
        """Comprehensive single contact enhancement for multichannel engagement"""
        enhancement_tasks = []
        
        # Email validation and deliverability analysis
        if contact.email:
            task1 = self._validate_email_for_multichannel(contact.email)
            enhancement_tasks.append(task1)
        
        # Phone enrichment and validation
        if contact.first_name and contact.last_name:
            task2 = self._enrich_and_validate_phone(contact)
            enhancement_tasks.append(task2)
        
        # LinkedIn profile validation and analysis
        if contact.linkedin_url or (contact.first_name and contact.last_name and contact.company):
            task3 = self._validate_linkedin_profile(contact)
            enhancement_tasks.append(task3)
        
        # Execute all enhancements
        results = await asyncio.gather(*enhancement_tasks, return_exceptions=True)
        
        # Process email validation results
        if len(results) > 0 and not isinstance(results[0], Exception):
            contact.email_validation = results[0]
        
        # Process phone enrichment results
        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 LinkedIn validation results
        if len(results) > 2 and not isinstance(results[2], Exception) and results[2]:
            contact.linkedin_validation = results[2]
            if not contact.linkedin_url and results[2].get('linkedin_url'):
                contact.linkedin_url = results[2]['linkedin_url']
        
        # Calculate channel preferences and priorities
        contact.channel_preferences = self._analyze_channel_preferences(contact)
        contact.preferred_channels = self._determine_preferred_channels(contact)
        contact.channel_priority = self._calculate_channel_priority(contact)
        contact.engagement_score = self._calculate_multichannel_engagement_score(contact)
        contact.multichannel_readiness = self._assess_multichannel_readiness(contact)
    
    async def _validate_email_for_multichannel(self, email: str) -> Dict:
        """Advanced email validation for multichannel campaigns"""
        url = f"{self.lookup_base_url}/email/validate"
        headers = {
            'Authorization': f'Bearer {self.lookup_api_key}',
            'Content-Type': 'application/json'
        }
        
        payload = {
            'email': email,
            'multichannel_analysis': True,
            'engagement_prediction': True,
            'deliverability_optimization': True,
            'response_likelihood': 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 email channel score
                    email_score = 0
                    if result.get('deliverable'): email_score += 40
                    if result.get('valid_format'): email_score += 20
                    if not result.get('spam_trap'): email_score += 20
                    if result.get('engagement_prediction', 0) > 70: email_score += 20
                    
                    result['channel_score'] = email_score
                    return result
                return {'deliverable': False, 'channel_score': 0}
    
    async def _enrich_and_validate_phone(self, contact: MultichannelContact) -> Optional[Dict]:
        """Phone enrichment and validation for call/SMS 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,
            'multichannel_optimization': True,
            'sms_capability_check': True,
            'call_quality_assessment': 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()
                        
                        # Calculate phone channel scores
                        phone_score = 0
                        sms_score = 0
                        
                        if result.get('valid'):
                            phone_score += 50
                            if result.get('call_quality') == 'high': phone_score += 30
                            if result.get('business_phone'): phone_score += 20
                            
                            if result.get('sms_capable'):
                                sms_score += 70
                                if result.get('sms_engagement_likelihood', 0) > 60: sms_score += 30
                        
                        result['phone_channel_score'] = phone_score
                        result['sms_channel_score'] = sms_score
                        return result
        except:
            pass
        
        return None
    
    async def _validate_linkedin_profile(self, contact: MultichannelContact) -> Optional[Dict]:
        """LinkedIn profile validation and engagement analysis"""
        url = f"{self.lookup_base_url}/enrich/linkedin"
        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,
            'linkedin_url': contact.linkedin_url,
            'engagement_analysis': True,
            'connection_likelihood': True,
            'response_prediction': 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()
                        
                        # Calculate LinkedIn channel score
                        linkedin_score = 0
                        if result.get('profile_found'):
                            linkedin_score += 30
                            if result.get('active_user'): linkedin_score += 25
                            if result.get('connection_likelihood', 0) > 70: linkedin_score += 25
                            if result.get('response_likelihood', 0) > 60: linkedin_score += 20
                        
                        result['linkedin_channel_score'] = linkedin_score
                        return result
        except:
            pass
        
        return None
    
    def _analyze_channel_preferences(self, contact: MultichannelContact) -> Dict:
        """Analyze optimal channel preferences for contact"""
        preferences = {
            'email': 0,
            'phone': 0,
            'linkedin': 0,
            'sms': 0
        }
        
        # Email preference analysis
        if contact.email_validation:
            email_score = contact.email_validation.get('channel_score', 0)
            preferences['email'] = email_score
        
        # Phone/SMS preference analysis
        if contact.phone_validation:
            phone_score = contact.phone_validation.get('phone_channel_score', 0)
            sms_score = contact.phone_validation.get('sms_channel_score', 0)
            preferences['phone'] = phone_score
            preferences['sms'] = sms_score
        
        # LinkedIn preference analysis
        if contact.linkedin_validation:
            linkedin_score = contact.linkedin_validation.get('linkedin_channel_score', 0)
            preferences['linkedin'] = linkedin_score
        
        return preferences
    
    def _determine_preferred_channels(self, contact: MultichannelContact) -> List[str]:
        """Determine ordered list of preferred channels"""
        if not contact.channel_preferences:
            return ['email']  # Default fallback
        
        # Sort channels by preference score
        sorted_channels = sorted(
            contact.channel_preferences.items(),
            key=lambda x: x[1],
            reverse=True
        )
        
        # Return channels with score > 50
        preferred = [channel for channel, score in sorted_channels if score > 50]
        
        # Ensure at least one channel
        if not preferred and sorted_channels:
            preferred = [sorted_channels[0][0]]
        
        return preferred
    
    def _calculate_channel_priority(self, contact: MultichannelContact) -> Dict[str, int]:
        """Calculate channel priority for sequence optimization"""
        if not contact.channel_preferences:
            return {'email': 1}
        
        # Assign priorities based on preference scores
        priority_map = {}
        sorted_channels = sorted(
            contact.channel_preferences.items(),
            key=lambda x: x[1],
            reverse=True
        )
        
        for i, (channel, score) in enumerate(sorted_channels):
            if score > 50:  # Only include viable channels
                priority_map[channel] = i + 1
        
        return priority_map
    
    def _calculate_multichannel_engagement_score(self, contact: MultichannelContact) -> int:
        """Calculate comprehensive multichannel engagement score"""
        score = 0
        
        # Channel availability bonus (0-40 points)
        available_channels = len([ch for ch in contact.preferred_channels or [] if ch])
        score += min(available_channels * 10, 40)
        
        # Channel quality scores (0-40 points)
        if contact.channel_preferences:
            avg_channel_score = sum(contact.channel_preferences.values()) / len(contact.channel_preferences)
            score += int(avg_channel_score * 0.4)
        
        # Contact completeness (0-20 points)
        completeness = 0
        if contact.email: completeness += 5
        if contact.phone: completeness += 5
        if contact.linkedin_url: completeness += 5
        if contact.position and contact.company: completeness += 5
        score += completeness
        
        return min(score, 100)
    
    def _assess_multichannel_readiness(self, contact: MultichannelContact) -> str:
        """Assess contact's readiness for multichannel campaigns"""
        if contact.engagement_score >= 80 and len(contact.preferred_channels or []) >= 3:
            return "excellent"
        elif contact.engagement_score >= 60 and len(contact.preferred_channels or []) >= 2:
            return "good"
        elif contact.engagement_score >= 40:
            return "fair"
        else:
            return "limited"
    
    async def _optimize_channel_sequences(self, campaign_id: str, 
                                        contacts: List[MultichannelContact]) -> Dict:
        """Optimize Reply.io sequences based on contact channel preferences"""
        sequence_optimization = {
            'excellent_contacts': [],
            'good_contacts': [],
            'fair_contacts': [],
            'limited_contacts': [],
            'sequence_recommendations': {}
        }
        
        # Segment contacts by multichannel readiness
        for contact in contacts:
            readiness = contact.multichannel_readiness
            if readiness == 'excellent':
                sequence_optimization['excellent_contacts'].append(contact)
            elif readiness == 'good':
                sequence_optimization['good_contacts'].append(contact)
            elif readiness == 'fair':
                sequence_optimization['fair_contacts'].append(contact)
            else:
                sequence_optimization['limited_contacts'].append(contact)
        
        # Generate sequence recommendations for each segment
        sequence_optimization['sequence_recommendations'] = {
            'excellent': self._generate_excellent_sequence_recommendation(),
            'good': self._generate_good_sequence_recommendation(),
            'fair': self._generate_fair_sequence_recommendation(),
            'limited': self._generate_limited_sequence_recommendation()
        }
        
        return sequence_optimization
    
    def _generate_excellent_sequence_recommendation(self) -> Dict:
        """Generate sequence for contacts with excellent multichannel readiness"""
        return {
            'sequence_type': 'aggressive_multichannel',
            'total_steps': 8,
            'channels': ['email', 'linkedin', 'phone', 'email', 'sms', 'email', 'phone', 'email'],
            'intervals': [1, 3, 2, 4, 2, 5, 3, 7],  # Days between steps
            'personalization_level': 'high'
        }
    
    def _generate_good_sequence_recommendation(self) -> Dict:
        """Generate sequence for contacts with good multichannel readiness"""
        return {
            'sequence_type': 'balanced_multichannel',
            'total_steps': 6,
            'channels': ['email', 'linkedin', 'email', 'phone', 'email', 'email'],
            'intervals': [2, 4, 3, 5, 4, 7],
            'personalization_level': 'medium'
        }
    
    def _generate_fair_sequence_recommendation(self) -> Dict:
        """Generate sequence for contacts with fair multichannel readiness"""
        return {
            'sequence_type': 'conservative_multichannel',
            'total_steps': 5,
            'channels': ['email', 'email', 'linkedin', 'email', 'email'],
            'intervals': [3, 5, 4, 6, 8],
            'personalization_level': 'medium'
        }
    
    def _generate_limited_sequence_recommendation(self) -> Dict:
        """Generate sequence for contacts with limited multichannel readiness"""
        return {
            'sequence_type': 'email_focused',
            'total_steps': 4,
            'channels': ['email', 'email', 'email', 'email'],
            'intervals': [3, 7, 7, 14],
            'personalization_level': 'basic'
        }

# Example usage for multichannel optimization
async def main():
    integration = ReplyIO1lookupIntegration(
        reply_api_key="your_reply_io_api_key",
        lookup_api_key="your_1lookup_api_key"
    )
    
    # Optimize existing campaign
    optimization_result = await integration.optimize_multichannel_campaign("your_campaign_id")
    
    print("Multichannel Optimization Results:")
    print(f"Total Contacts: {optimization_result.get('total_contacts')}")
    print(f"Enhanced Contacts: {optimization_result.get('enhanced_contacts')}")
    
    multichannel_stats = optimization_result.get('multichannel_stats', {})
    print(f"
Multichannel Statistics:")
    print(f"Excellent Readiness: {multichannel_stats.get('excellent_readiness', 0)} contacts")
    print(f"Average Engagement Score: {multichannel_stats.get('average_engagement_score', 0):.1f}")
    print(f"Multichannel Capable: {multichannel_stats.get('multichannel_capable', 0)} contacts")
    
    # Create new optimized campaign
    sample_contacts = [
        {
            'email': 'john@example.com',
            'firstName': 'John',
            'lastName': 'Smith',
            'position': 'Sales Director',
            'company': 'Example Corp'
        },
        # Add more contacts...
    ]
    
    new_campaign = await integration.create_optimized_multichannel_campaign(
        campaign_name="Optimized Multichannel Outreach",
        contacts=sample_contacts
    )
    
    print(f"
New Campaign Created:")
    print(f"Campaign ID: {new_campaign.get('campaign_id')}")
    print(f"Multichannel Optimization Applied: {len(new_campaign.get('multichannel_optimization', {}))} segments")

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