Best QuickMail 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 AutomationContact:
    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
    website: Optional[str] = None
    email_validation: Optional[Dict] = None
    phone_validation: Optional[Dict] = None
    automation_intelligence: Optional[Dict] = None
    personalization_data: Optional[Dict] = None
    deliverability_score: int = 0
    automation_score: int = 0
    sequence_priority: str = "medium"
    personalization_depth: str = "basic"
    automation_readiness: str = "pending"
    quickmail_optimization: Optional[Dict] = None
    enhancement_timestamp: str = ""

class QuickMail1lookupIntegration:
    def __init__(self, quickmail_api_key: str, lookup_api_key: str):
        self.quickmail_api_key = quickmail_api_key
        self.lookup_api_key = lookup_api_key
        self.quickmail_base_url = "https://api.quickmail.io/v1"
        self.lookup_base_url = "https://api.1lookup.app/v1"
    
    async def optimize_quickmail_campaign(self, campaign_id: str) -> Dict:
        """
        Optimize existing QuickMail campaign with comprehensive automation intelligence
        """
        try:
            # Step 1: Get campaign contacts from QuickMail
            campaign_contacts = await self._get_quickmail_campaign_contacts(campaign_id)
            
            if not campaign_contacts:
                return {"error": "No contacts found in campaign"}
            
            # Step 2: Enhance all contacts with automation intelligence
            enhanced_contacts = await self._bulk_enhance_for_automation(campaign_contacts)
            
            # Step 3: Generate automation optimization strategies
            automation_strategies = await self._generate_automation_strategies(enhanced_contacts)
            
            # Step 4: Create personalization enhancement recommendations
            personalization_enhancement = self._generate_personalization_enhancements(enhanced_contacts)
            
            # Step 5: Update QuickMail campaign with enhanced data
            update_results = await self._update_quickmail_campaign(campaign_id, enhanced_contacts)
            
            return {
                "campaign_id": campaign_id,
                "total_contacts": len(campaign_contacts),
                "enhanced_contacts": len(enhanced_contacts),
                "automation_stats": self._calculate_automation_stats(enhanced_contacts),
                "automation_strategies": automation_strategies,
                "personalization_enhancement": personalization_enhancement,
                "update_results": update_results
            }
            
        except Exception as e:
            return {"error": f"QuickMail campaign optimization failed: {e}"}
    
    async def create_enhanced_automation_sequence(self, sequence_name: str,
                                                contacts: List[Dict],
                                                sequence_config: Dict = None) -> Dict:
        """
        Create enhanced QuickMail automation sequence with comprehensive optimization
        """
        try:
            # Step 1: Enhance contacts with automation intelligence
            enhanced_contacts = await self._bulk_enhance_for_automation(contacts)
            
            # Step 2: Generate optimized sequence structure
            sequence_structure = await self._generate_automation_sequence_structure(
                enhanced_contacts, sequence_config
            )
            
            # Step 3: Create QuickMail sequence with optimization
            sequence_data = await self._create_quickmail_sequence(sequence_name, sequence_structure)
            
            if not sequence_data:
                return {"error": "Failed to create QuickMail sequence"}
            
            # Step 4: Apply automation optimizations
            optimization_results = await self._apply_automation_optimizations(
                sequence_data['id'], enhanced_contacts
            )
            
            return {
                "sequence_id": sequence_data['id'],
                "sequence_name": sequence_name,
                "total_contacts": len(enhanced_contacts),
                "automation_optimization": self._analyze_automation_optimization(enhanced_contacts),
                "optimization_results": optimization_results
            }
            
        except Exception as e:
            return {"error": f"Enhanced sequence creation failed: {e}"}
    
    async def _bulk_enhance_for_automation(self, contacts: List[Dict]) -> List[AutomationContact]:
        """Enhance multiple contacts for optimal automation campaigns"""
        enhancement_tasks = []
        
        for contact_data in contacts:
            automation_contact = AutomationContact(
                email=contact_data.get('email', ''),
                first_name=contact_data.get('first_name') or contact_data.get('firstName'),
                last_name=contact_data.get('last_name') or contact_data.get('lastName'),
                position=contact_data.get('position') or contact_data.get('title'),
                company=contact_data.get('company'),
                domain=contact_data.get('domain'),
                website=contact_data.get('website'),
                linkedin_url=contact_data.get('linkedin_url'),
                enhancement_timestamp=datetime.now().isoformat()
            )
            
            task = self._enhance_single_contact_for_automation(automation_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_automation(self, contact: AutomationContact) -> None:
        """Comprehensive single contact enhancement for automation campaigns"""
        enhancement_tasks = []
        
        # Email validation with automation optimization
        if contact.email:
            task1 = self._validate_email_for_automation(contact.email)
            enhancement_tasks.append(task1)
        
        # Phone enrichment for multi-touch automation
        if contact.first_name and contact.last_name:
            task2 = self._enrich_phone_for_automation(contact)
            enhancement_tasks.append(task2)
        
        # Automation intelligence gathering
        task3 = self._gather_automation_intelligence(contact)
        enhancement_tasks.append(task3)
        
        # Execute all 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')
        
        # Process automation intelligence
        if len(results) > 2 and not isinstance(results[2], Exception):
            contact.automation_intelligence = results[2]
        
        # Generate enhanced personalization data
        contact.personalization_data = self._generate_enhanced_personalization_data(contact)
        
        # Calculate automation scores and optimizations
        contact.automation_score = self._calculate_automation_score(contact)
        contact.sequence_priority = self._determine_sequence_priority(contact)
        contact.personalization_depth = self._assess_automation_personalization_depth(contact)
        contact.automation_readiness = self._assess_automation_readiness(contact)
        contact.quickmail_optimization = self._generate_quickmail_optimization(contact)
    
    async def _validate_email_for_automation(self, email: str) -> Dict:
        """Advanced email validation optimized for automation campaigns"""
        url = f"{self.lookup_base_url}/email/validate"
        headers = {
            'Authorization': f'Bearer {self.lookup_api_key}',
            'Content-Type': 'application/json'
        }
        
        payload = {
            'email': email,
            'automation_optimization': True,
            'cold_email_suitability': True,
            'sequence_deliverability': True,
            'automation_friendly_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 automation deliverability score
                    score = 0
                    if result.get('deliverable'): score += 35
                    if result.get('valid_format'): score += 20
                    if not result.get('spam_trap'): score += 20
                    if result.get('automation_friendly', True): score += 25
                    
                    result['deliverability_score'] = score
                    result['automation_suitability'] = 'excellent' if score >= 85 else 'good' if score >= 70 else 'fair' if score >= 50 else 'poor'
                    return result
                return {'deliverable': False, 'deliverability_score': 0, 'automation_suitability': 'poor'}
    
    async def _enrich_phone_for_automation(self, contact: AutomationContact) -> Optional[Dict]:
        """Phone enrichment optimized for multi-touch automation"""
        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,
            'automation_focus': True,
            'multi_touch_optimization': 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 automation multi-touch potential
                        if result.get('valid'):
                            result['automation_score'] = 90 if result.get('mobile') and result.get('business_phone') else 75 if result.get('mobile') or result.get('business_phone') else 60
                            result['multi_touch_capability'] = 'high' if result['automation_score'] >= 75 else 'medium'
                        else:
                            result['automation_score'] = 0
                            result['multi_touch_capability'] = 'low'
                        
                        return result
        except:
            pass
        
        return None
    
    async def _gather_automation_intelligence(self, contact: AutomationContact) -> Optional[Dict]:
        """Gather comprehensive intelligence for automation optimization"""
        url = f"{self.lookup_base_url}/enrich/automation"
        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,
            'position': contact.position,
            'website': contact.website,
            'automation_personalization_focus': True,
            'sequence_optimization': 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 _generate_enhanced_personalization_data(self, contact: AutomationContact) -> Dict:
        """Generate comprehensive personalization data for automation sequences"""
        personalization = {
            'basic_variables': {},
            'advanced_variables': {},
            'company_intelligence': {},
            'automation_hooks': [],
            'sequence_personalization': {}
        }
        
        # Basic personalization variables
        personalization['basic_variables'] = {
            'first_name': contact.first_name or 'there',
            'last_name': contact.last_name or '',
            'company': contact.company or 'your company',
            'position': contact.position or 'your role',
            'domain': contact.domain or ''
        }
        
        # Advanced personalization variables
        if contact.automation_intelligence:
            intel = contact.automation_intelligence
            personalization['advanced_variables'] = {
                'industry': intel.get('industry', ''),
                'company_size': intel.get('company_size', ''),
                'technologies': intel.get('tech_stack', []),
                'recent_funding': intel.get('funding_info', ''),
                'growth_stage': intel.get('growth_stage', '')
            }
            
            # Company intelligence for deeper personalization
            personalization['company_intelligence'] = {
                'recent_news': intel.get('company_news', []),
                'competitors': intel.get('competitors', []),
                'key_initiatives': intel.get('initiatives', []),
                'pain_points': intel.get('likely_pain_points', [])
            }
        
        # Generate automation-specific hooks
        hooks = []
        if contact.company and personalization['advanced_variables'].get('industry'):
            industry = personalization['advanced_variables']['industry']
            hooks.append(f"I noticed {contact.company} is innovating in {industry}")
        
        if contact.position:
            hooks.append(f"As a {contact.position}, you're probably dealing with...")
        
        if personalization['company_intelligence'].get('recent_news'):
            hooks.append("I saw the recent developments at your company")
        
        personalization['automation_hooks'] = hooks[:3]
        
        # Sequence-specific personalization recommendations
        personalization['sequence_personalization'] = {
            'opener_style': 'professional' if contact.position and any(title in contact.position.lower() for title in ['ceo', 'vp', 'director']) else 'friendly',
            'pain_point_focus': self._determine_pain_point_focus(contact),
            'follow_up_angle': self._determine_follow_up_angle(contact),
            'cta_approach': 'direct' if contact.automation_score >= 75 else 'soft'
        }
        
        return personalization
    
    def _calculate_automation_score(self, contact: AutomationContact) -> int:
        """Calculate comprehensive automation effectiveness score"""
        score = 0
        
        # Email deliverability (0-35 points)
        score += contact.deliverability_score * 0.35
        
        # Contact completeness (0-25 points)
        completeness = 0
        if contact.email: completeness += 8
        if contact.first_name: completeness += 5
        if contact.last_name: completeness += 4
        if contact.company: completeness += 4
        if contact.position: completeness += 4
        score += completeness
        
        # Automation intelligence (0-20 points)
        if contact.automation_intelligence:
            intelligence_quality = len([v for v in contact.automation_intelligence.values() if v])
            score += min(intelligence_quality * 2, 20)
        
        # Multi-touch capability (0-15 points)
        if contact.phone_validation and contact.phone_validation.get('valid'):
            score += contact.phone_validation.get('automation_score', 0) * 0.15
        
        # Personalization potential (0-5 points)
        if contact.personalization_data:
            hooks_count = len(contact.personalization_data.get('automation_hooks', []))
            score += min(hooks_count * 2, 5)
        
        return min(int(score), 100)
    
    def _determine_sequence_priority(self, contact: AutomationContact) -> str:
        """Determine automation sequence priority"""
        if contact.automation_score >= 80:
            return "high"
        elif contact.automation_score >= 60:
            return "medium"
        else:
            return "low"
    
    def _assess_automation_personalization_depth(self, contact: AutomationContact) -> str:
        """Assess personalization depth available for automation"""
        if not contact.personalization_data:
            return "basic"
        
        factors = 0
        
        # Count available personalization factors
        if contact.personalization_data.get('advanced_variables'):
            factors += len([v for v in contact.personalization_data['advanced_variables'].values() if v])
        
        if contact.personalization_data.get('company_intelligence'):
            factors += len([v for v in contact.personalization_data['company_intelligence'].values() if v])
        
        if contact.personalization_data.get('automation_hooks'):
            factors += len(contact.personalization_data['automation_hooks'])
        
        if factors >= 8:
            return "ultra_deep"
        elif factors >= 5:
            return "deep"
        elif factors >= 3:
            return "medium"
        else:
            return "basic"
    
    def _assess_automation_readiness(self, contact: AutomationContact) -> str:
        """Assess contact's readiness for automation campaigns"""
        readiness_factors = 0
        
        # Email deliverability
        if contact.email_validation and contact.email_validation.get('deliverable'):
            readiness_factors += 2
        
        # Automation suitability
        automation_suitability = contact.email_validation.get('automation_suitability', 'poor') if contact.email_validation else 'poor'
        if automation_suitability in ['excellent', 'good']:
            readiness_factors += 2
        
        # Personalization depth
        if contact.personalization_depth in ['ultra_deep', 'deep']:
            readiness_factors += 2
        elif contact.personalization_depth == 'medium':
            readiness_factors += 1
        
        # Multi-touch capability
        if contact.phone_validation and contact.phone_validation.get('valid'):
            readiness_factors += 1
        
        # Professional information
        if contact.position and contact.company:
            readiness_factors += 1
        
        if readiness_factors >= 7:
            return "excellent"
        elif readiness_factors >= 5:
            return "good"
        elif readiness_factors >= 3:
            return "fair"
        else:
            return "limited"
    
    def _generate_quickmail_optimization(self, contact: AutomationContact) -> Dict:
        """Generate QuickMail-specific optimization recommendations"""
        optimization = {
            'sequence_type': 'standard',
            'personalization_variables': {},
            'follow_up_strategy': 'email_sequence',
            'timing_optimization': {},
            'content_recommendations': []
        }
        
        # Sequence type based on automation score
        if contact.automation_score >= 85:
            optimization['sequence_type'] = 'premium_personalized'
        elif contact.automation_score >= 70:
            optimization['sequence_type'] = 'enhanced_automated'
        elif contact.automation_score >= 50:
            optimization['sequence_type'] = 'standard_automated'
        else:
            optimization['sequence_type'] = 'basic_sequence'
        
        # Personalization variables for QuickMail
        if contact.personalization_data:
            optimization['personalization_variables'] = {
                **contact.personalization_data.get('basic_variables', {}),
                'automation_hook': contact.personalization_data.get('automation_hooks', [''])[0],
                'industry_insight': contact.personalization_data.get('advanced_variables', {}).get('industry', ''),
                'pain_point': self._determine_pain_point_focus(contact)
            }
        
        # Follow-up strategy
        if contact.phone_validation and contact.phone_validation.get('valid'):
            optimization['follow_up_strategy'] = 'multi_touch_sequence'
        elif contact.linkedin_url:
            optimization['follow_up_strategy'] = 'email_linkedin_sequence'
        
        # Timing optimization
        optimization['timing_optimization'] = {
            'sequence_length': min(3 + (contact.automation_score // 20), 7),
            'intervals': self._calculate_optimal_intervals(contact),
            'send_times': ['morning'] if contact.automation_score >= 70 else ['morning', 'afternoon']
        }
        
        # Content recommendations
        content_recs = []
        if contact.personalization_depth in ['ultra_deep', 'deep']:
            content_recs.append('highly_personalized_opener')
            content_recs.append('company_specific_value_prop')
        
        if contact.position:
            content_recs.append('role_based_pain_point')
        
        optimization['content_recommendations'] = content_recs
        
        return optimization
    
    def _calculate_automation_stats(self, contacts: List[AutomationContact]) -> Dict:
        """Calculate comprehensive automation statistics"""
        total = len(contacts)
        if total == 0:
            return {}
        
        deliverable = sum(1 for c in contacts if c.email_validation and c.email_validation.get('deliverable'))
        automation_ready = sum(1 for c in contacts if c.automation_readiness in ['excellent', 'good'])
        high_score = sum(1 for c in contacts if c.automation_score >= 70)
        multi_touch = sum(1 for c in contacts if c.phone_validation and c.phone_validation.get('valid'))
        deep_personalization = sum(1 for c in contacts if c.personalization_depth in ['ultra_deep', 'deep'])
        
        return {
            'total_contacts': total,
            'deliverable_rate': (deliverable / total) * 100,
            'automation_ready_rate': (automation_ready / total) * 100,
            'high_score_rate': (high_score / total) * 100,
            'multi_touch_capable': (multi_touch / total) * 100,
            'deep_personalization_rate': (deep_personalization / total) * 100,
            'average_automation_score': sum(c.automation_score for c in contacts) / total,
            'high_priority_contacts': sum(1 for c in contacts if c.sequence_priority == 'high')
        }

# Example usage for QuickMail automation optimization
async def main():
    integration = QuickMail1lookupIntegration(
        quickmail_api_key="your_quickmail_api_key",
        lookup_api_key="your_1lookup_api_key"
    )
    
    # Sample contacts for automation enhancement
    sample_contacts = [
        {
            'email': 'john@example.com',
            'first_name': 'John',
            'last_name': 'Smith',
            'position': 'Marketing Director',
            'company': 'Example Corp',
            'website': 'https://example.com'
        },
        # Add more contacts...
    ]
    
    # Create enhanced automation sequence
    sequence_result = await integration.create_enhanced_automation_sequence(
        sequence_name="Enhanced Cold Outreach Q1",
        contacts=sample_contacts,
        sequence_config={
            "personalization_level": "deep",
            "multi_touch": True,
            "automation_optimization": True
        }
    )
    
    print("QuickMail Automation Enhancement Results:")
    print(f"Sequence ID: {sequence_result.get('sequence_id')}")
    print(f"Total Contacts: {sequence_result.get('total_contacts')}")
    
    automation_optimization = sequence_result.get('automation_optimization', {})
    print(f"
Automation Optimization:")
    print(f"Automation Ready Rate: {automation_optimization.get('automation_ready_rate', 0):.1f}%")
    print(f"Deep Personalization Rate: {automation_optimization.get('deep_personalization_rate', 0):.1f}%")
    print(f"Multi-Touch Capable: {automation_optimization.get('multi_touch_capable', 0):.1f}%")

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