Best SalesHandy Phone Validation API & Email Verification Integration 2025

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

@dataclass
class ColdEmailContact:
    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
    cold_email_intelligence: Optional[Dict] = None
    personalization_data: Optional[Dict] = None
    deliverability_score: int = 0
    cold_email_score: int = 0
    outreach_priority: str = "medium"
    personalization_level: str = "basic"
    campaign_readiness: str = "pending"
    saleshandy_optimization: Optional[Dict] = None
    enhancement_timestamp: str = ""

class SalesHandy1lookupIntegration:
    def __init__(self, saleshandy_api_key: str, lookup_api_key: str):
        self.saleshandy_api_key = saleshandy_api_key
        self.lookup_api_key = lookup_api_key
        self.saleshandy_base_url = "https://app.saleshandy.com/api/v1"
        self.lookup_base_url = "https://api.1lookup.app/v1"
    
    async def optimize_cold_email_campaign(self, campaign_id: str) -> Dict:
        """
        Optimize existing SalesHandy cold email campaign with comprehensive intelligence
        """
        try:
            # Step 1: Get campaign contacts from SalesHandy
            campaign_contacts = await self._get_saleshandy_campaign_contacts(campaign_id)
            
            if not campaign_contacts:
                return {"error": "No contacts found in campaign"}
            
            # Step 2: Enhance all contacts with cold email intelligence
            enhanced_contacts = await self._bulk_enhance_for_cold_email(campaign_contacts)
            
            # Step 3: Generate cold email optimization strategies
            optimization_strategies = await self._generate_cold_email_strategies(enhanced_contacts)
            
            # Step 4: Update SalesHandy campaign with enhanced data
            update_results = await self._update_saleshandy_campaign(campaign_id, enhanced_contacts)
            
            return {
                "campaign_id": campaign_id,
                "total_contacts": len(campaign_contacts),
                "enhanced_contacts": len(enhanced_contacts),
                "cold_email_stats": self._calculate_cold_email_stats(enhanced_contacts),
                "optimization_strategies": optimization_strategies,
                "update_results": update_results
            }
            
        except Exception as e:
            return {"error": f"Cold email campaign optimization failed: {e}"}
    
    async def create_optimized_cold_email_campaign(self, campaign_name: str,
                                                 contacts: List[Dict],
                                                 campaign_config: Dict = None) -> Dict:
        """
        Create new SalesHandy campaign with cold email optimization
        """
        try:
            # Step 1: Enhance contacts with cold email intelligence
            enhanced_contacts = await self._bulk_enhance_for_cold_email(contacts)
            
            # Step 2: Generate optimal cold email campaign structure
            campaign_structure = await self._generate_cold_email_campaign_structure(
                enhanced_contacts, campaign_config
            )
            
            # Step 3: Create SalesHandy campaign with optimization
            campaign_data = await self._create_saleshandy_campaign(campaign_name, campaign_structure)
            
            if not campaign_data:
                return {"error": "Failed to create SalesHandy campaign"}
            
            # Step 4: Apply cold email optimizations
            optimization_results = await self._apply_cold_email_optimizations(
                campaign_data['id'], enhanced_contacts
            )
            
            return {
                "campaign_id": campaign_data['id'],
                "campaign_name": campaign_name,
                "total_contacts": len(enhanced_contacts),
                "cold_email_optimization": self._analyze_cold_email_optimization(enhanced_contacts),
                "optimization_results": optimization_results
            }
            
        except Exception as e:
            return {"error": f"Optimized campaign creation failed: {e}"}
    
    async def process_mail_merge_with_validation(self, csv_file_path: str,
                                               template_config: Dict) -> Dict:
        """
        Process mail merge CSV with comprehensive validation and enhancement
        """
        try:
            # Step 1: Load and parse CSV file
            contacts = await self._load_csv_contacts(csv_file_path)
            
            # Step 2: Enhance contacts with validation
            enhanced_contacts = await self._bulk_enhance_for_cold_email(contacts)
            
            # Step 3: Generate enhanced mail merge data
            mail_merge_data = self._generate_enhanced_mail_merge_data(enhanced_contacts, template_config)
            
            # Step 4: Export validated and enriched CSV
            output_file = await self._export_enhanced_csv(enhanced_contacts, csv_file_path)
            
            return {
                "input_file": csv_file_path,
                "output_file": output_file,
                "total_contacts": len(contacts),
                "validated_contacts": len([c for c in enhanced_contacts if c.email_validation and c.email_validation.get('deliverable')]),
                "enhancement_stats": self._calculate_cold_email_stats(enhanced_contacts),
                "mail_merge_data": mail_merge_data
            }
            
        except Exception as e:
            return {"error": f"Mail merge processing failed: {e}"}
    
    async def _bulk_enhance_for_cold_email(self, contacts: List[Dict]) -> List[ColdEmailContact]:
        """Enhance multiple contacts for optimal cold email campaigns"""
        enhancement_tasks = []
        
        for contact_data in contacts:
            cold_email_contact = ColdEmailContact(
                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'),
                linkedin_url=contact_data.get('linkedin_url'),
                enhancement_timestamp=datetime.now().isoformat()
            )
            
            task = self._enhance_single_contact_for_cold_email(cold_email_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_cold_email(self, contact: ColdEmailContact) -> None:
        """Comprehensive single contact enhancement for cold email campaigns"""
        enhancement_tasks = []
        
        # Email validation with cold email focus
        if contact.email:
            task1 = self._validate_email_for_cold_outreach(contact.email)
            enhancement_tasks.append(task1)
        
        # Phone enrichment for follow-up campaigns
        if contact.first_name and contact.last_name:
            task2 = self._enrich_phone_for_cold_email_followup(contact)
            enhancement_tasks.append(task2)
        
        # Cold email intelligence gathering
        task3 = self._gather_cold_email_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 cold email intelligence
        if len(results) > 2 and not isinstance(results[2], Exception):
            contact.cold_email_intelligence = results[2]
        
        # Generate personalization data
        contact.personalization_data = self._generate_personalization_data(contact)
        
        # Calculate scores and optimizations
        contact.cold_email_score = self._calculate_cold_email_score(contact)
        contact.outreach_priority = self._determine_outreach_priority(contact)
        contact.personalization_level = self._assess_personalization_level(contact)
        contact.campaign_readiness = self._assess_campaign_readiness(contact)
        contact.saleshandy_optimization = self._generate_saleshandy_optimization(contact)
    
    async def _validate_email_for_cold_outreach(self, email: str) -> Dict:
        """Advanced email validation optimized for cold email outreach"""
        url = f"{self.lookup_base_url}/email/validate"
        headers = {
            'Authorization': f'Bearer {self.lookup_api_key}',
            'Content-Type': 'application/json'
        }
        
        payload = {
            'email': email,
            'cold_email_optimization': True,
            'deliverability_focus': True,
            'spam_risk_assessment': True,
            'reply_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 cold email 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('cold_email_friendly', True): score += 20
                    
                    result['deliverability_score'] = score
                    result['cold_email_suitability'] = 'excellent' if score >= 80 else 'good' if score >= 60 else 'fair' if score >= 40 else 'poor'
                    return result
                return {'deliverable': False, 'deliverability_score': 0, 'cold_email_suitability': 'poor'}
    
    async def _enrich_phone_for_cold_email_followup(self, contact: ColdEmailContact) -> Optional[Dict]:
        """Phone enrichment optimized for cold email follow-up 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,
            'cold_outreach_focus': True,
            'follow_up_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 cold email follow-up potential
                        if result.get('valid'):
                            result['cold_email_followup_score'] = 90 if result.get('business_phone') else 70
                            result['multi_touch_potential'] = 'high'
                        else:
                            result['cold_email_followup_score'] = 0
                            result['multi_touch_potential'] = 'low'
                        
                        return result
        except:
            pass
        
        return None
    
    async def _gather_cold_email_intelligence(self, contact: ColdEmailContact) -> Optional[Dict]:
        """Gather comprehensive intelligence for cold email personalization"""
        url = f"{self.lookup_base_url}/enrich/cold_email"
        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,
            'linkedin_url': contact.linkedin_url,
            'personalization_focus': True,
            'cold_outreach_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 _generate_personalization_data(self, contact: ColdEmailContact) -> Dict:
        """Generate comprehensive personalization data for cold email templates"""
        personalization = {
            'basic_fields': {},
            'company_intelligence': {},
            'social_intelligence': {},
            'personalization_hooks': [],
            'icebreaker_suggestions': []
        }
        
        # Basic personalization fields
        personalization['basic_fields'] = {
            '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'
        }
        
        # Company intelligence
        if contact.cold_email_intelligence:
            company_data = contact.cold_email_intelligence.get('company_data', {})
            personalization['company_intelligence'] = {
                'industry': company_data.get('industry', ''),
                'company_size': company_data.get('size', ''),
                'recent_news': company_data.get('recent_news', []),
                'technologies': company_data.get('tech_stack', [])
            }
        
        # Social intelligence
        if contact.cold_email_intelligence:
            social_data = contact.cold_email_intelligence.get('social_data', {})
            personalization['social_intelligence'] = {
                'linkedin_activity': social_data.get('linkedin_posts', []),
                'professional_interests': social_data.get('interests', []),
                'mutual_connections': social_data.get('mutual_connections', 0)
            }
        
        # Generate personalization hooks
        hooks = []
        if contact.company:
            hooks.append(f"I noticed {contact.company} is doing interesting work in...")
        if contact.position:
            hooks.append(f"As a {contact.position}, you probably face challenges with...")
        if personalization['company_intelligence'].get('recent_news'):
            hooks.append("I saw the recent news about your company...")
        
        personalization['personalization_hooks'] = hooks[:3]  # Limit to top 3
        
        # Generate icebreaker suggestions
        icebreakers = []
        if contact.linkedin_url:
            icebreakers.append("I found your LinkedIn profile and was impressed by...")
        if personalization['company_intelligence'].get('industry'):
            industry = personalization['company_intelligence']['industry']
            icebreakers.append(f"I've been following trends in {industry} and noticed...")
        
        personalization['icebreaker_suggestions'] = icebreakers[:2]  # Limit to top 2
        
        return personalization
    
    def _calculate_cold_email_score(self, contact: ColdEmailContact) -> int:
        """Calculate comprehensive cold email effectiveness score"""
        score = 0
        
        # Email deliverability (0-40 points)
        score += contact.deliverability_score * 0.4
        
        # Contact completeness (0-25 points)
        completeness = 0
        if contact.email: completeness += 10
        if contact.first_name: completeness += 5
        if contact.last_name: completeness += 5
        if contact.company: completeness += 3
        if contact.position: completeness += 2
        score += completeness
        
        # Personalization potential (0-20 points)
        if contact.personalization_data:
            hooks_count = len(contact.personalization_data.get('personalization_hooks', []))
            company_intel = len([v for v in contact.personalization_data.get('company_intelligence', {}).values() if v])
            score += min((hooks_count * 5) + (company_intel * 2), 20)
        
        # Phone follow-up capability (0-10 points)
        if contact.phone_validation and contact.phone_validation.get('valid'):
            score += 10
        
        # Professional relevance (0-5 points)
        if contact.position:
            position_lower = contact.position.lower()
            if any(title in position_lower for title in ['ceo', 'founder', 'director', 'vp']):
                score += 5
            elif any(title in position_lower for title in ['manager', 'lead', 'head']):
                score += 3
        
        return min(int(score), 100)
    
    def _determine_outreach_priority(self, contact: ColdEmailContact) -> str:
        """Determine cold email outreach priority"""
        if contact.cold_email_score >= 80:
            return "high"
        elif contact.cold_email_score >= 60:
            return "medium"
        else:
            return "low"
    
    def _assess_personalization_level(self, contact: ColdEmailContact) -> str:
        """Assess available personalization depth for cold email"""
        if not contact.personalization_data:
            return "basic"
        
        hooks_count = len(contact.personalization_data.get('personalization_hooks', []))
        company_intel_count = len([v for v in contact.personalization_data.get('company_intelligence', {}).values() if v])
        
        if hooks_count >= 3 and company_intel_count >= 3:
            return "ultra_deep"
        elif hooks_count >= 2 or company_intel_count >= 2:
            return "deep"
        elif hooks_count >= 1 or company_intel_count >= 1:
            return "medium"
        else:
            return "basic"
    
    def _assess_campaign_readiness(self, contact: ColdEmailContact) -> str:
        """Assess contact's readiness for cold email campaigns"""
        readiness_score = 0
        
        # Email quality
        if contact.email_validation and contact.email_validation.get('deliverable'):
            readiness_score += 3
        
        # Personalization potential
        if contact.personalization_level in ['ultra_deep', 'deep']:
            readiness_score += 3
        elif contact.personalization_level == 'medium':
            readiness_score += 2
        
        # Professional information
        if contact.position and contact.company:
            readiness_score += 2
        
        if readiness_score >= 7:
            return "excellent"
        elif readiness_score >= 5:
            return "good"
        elif readiness_score >= 3:
            return "fair"
        else:
            return "limited"
    
    def _generate_saleshandy_optimization(self, contact: ColdEmailContact) -> Dict:
        """Generate SalesHandy-specific optimization recommendations"""
        optimization = {
            'sequence_type': 'standard',
            'personalization_variables': {},
            'follow_up_strategy': 'email_only',
            'template_recommendations': [],
            'timing_optimization': {}
        }
        
        # Sequence type based on cold email score
        if contact.cold_email_score >= 80:
            optimization['sequence_type'] = 'premium_personalized'
        elif contact.cold_email_score >= 60:
            optimization['sequence_type'] = 'enhanced_standard'
        else:
            optimization['sequence_type'] = 'basic_outreach'
        
        # Personalization variables for SalesHandy templates
        if contact.personalization_data:
            optimization['personalization_variables'] = {
                'first_name': contact.first_name or 'there',
                'company': contact.company or 'your company',
                'position': contact.position or 'your role',
                'personalization_hook': contact.personalization_data.get('personalization_hooks', [''])[0],
                'company_industry': contact.personalization_data.get('company_intelligence', {}).get('industry', '')
            }
        
        # Follow-up strategy
        if contact.phone_validation and contact.phone_validation.get('valid'):
            optimization['follow_up_strategy'] = 'multi_channel'
        elif contact.linkedin_url:
            optimization['follow_up_strategy'] = 'email_linkedin'
        
        # Template recommendations
        templates = []
        if contact.personalization_level in ['ultra_deep', 'deep']:
            templates.append('highly_personalized_opener')
        if contact.position:
            templates.append('role_specific_value_prop')
        if contact.company:
            templates.append('company_relevant_case_study')
        
        optimization['template_recommendations'] = templates[:2]  # Limit to top 2
        
        # Timing optimization
        optimization['timing_optimization'] = {
            'send_time': 'morning',  # Default for cold email
            'follow_up_intervals': [3, 7, 14, 21],  # Days between follow-ups
            'sequence_length': min(4 + (contact.cold_email_score // 20), 7)  # 4-7 emails based on score
        }
        
        return optimization
    
    def _calculate_cold_email_stats(self, contacts: List[ColdEmailContact]) -> Dict:
        """Calculate comprehensive cold email 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'))
        high_score = sum(1 for c in contacts if c.cold_email_score >= 70)
        deep_personalization = sum(1 for c in contacts if c.personalization_level in ['ultra_deep', 'deep'])
        phone_capable = sum(1 for c in contacts if c.phone_validation and c.phone_validation.get('valid'))
        campaign_ready = sum(1 for c in contacts if c.campaign_readiness in ['excellent', 'good'])
        
        return {
            'total_contacts': total,
            'deliverable_rate': (deliverable / total) * 100,
            'high_score_rate': (high_score / total) * 100,
            'deep_personalization_rate': (deep_personalization / total) * 100,
            'phone_capable_rate': (phone_capable / total) * 100,
            'campaign_ready_rate': (campaign_ready / total) * 100,
            'average_cold_email_score': sum(c.cold_email_score for c in contacts) / total,
            'high_priority_contacts': sum(1 for c in contacts if c.outreach_priority == 'high')
        }

# Example usage for cold email optimization
async def main():
    integration = SalesHandy1lookupIntegration(
        saleshandy_api_key="your_saleshandy_api_key",
        lookup_api_key="your_1lookup_api_key"
    )
    
    # Process mail merge with validation
    mail_merge_result = await integration.process_mail_merge_with_validation(
        csv_file_path="cold_email_prospects.csv",
        template_config={
            "personalization_level": "deep",
            "follow_up_sequence": True
        }
    )
    
    print("SalesHandy Mail Merge Enhancement Results:")
    print(f"Input File: {mail_merge_result.get('input_file')}")
    print(f"Output File: {mail_merge_result.get('output_file')}")
    print(f"Total Contacts: {mail_merge_result.get('total_contacts')}")
    print(f"Validated Contacts: {mail_merge_result.get('validated_contacts')}")
    
    stats = mail_merge_result.get('enhancement_stats', {})
    print(f"
Cold Email Enhancement Statistics:")
    print(f"Deliverable Rate: {stats.get('deliverable_rate', 0):.1f}%")
    print(f"Deep Personalization Rate: {stats.get('deep_personalization_rate', 0):.1f}%")
    print(f"Campaign Ready Rate: {stats.get('campaign_ready_rate', 0):.1f}%")

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