Best Yesware 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 ProductivityContact:
    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
    sales_intelligence: Optional[Dict] = None
    productivity_metrics: Optional[Dict] = None
    crm_enrichment: Optional[Dict] = None
    tracking_optimization: Optional[Dict] = None
    productivity_score: int = 0
    sales_priority: str = "medium"
    engagement_readiness: str = "pending"
    crm_sync_status: str = "pending"
    enhancement_timestamp: str = ""

class Yesware1lookupIntegration:
    def __init__(self, yesware_api_key: str, lookup_api_key: str):
        self.yesware_api_key = yesware_api_key
        self.lookup_api_key = lookup_api_key
        self.yesware_base_url = "https://api.yesware.com/v1"
        self.lookup_base_url = "https://api.1lookup.app/v1"
    
    async def optimize_sales_productivity(self, contact_list: List[Dict],
                                        crm_system: str = "salesforce") -> Dict:
        """
        Optimize Yesware sales productivity with comprehensive contact intelligence
        """
        try:
            # Step 1: Enhance all contacts with sales intelligence
            enhanced_contacts = await self._bulk_enhance_for_sales_productivity(contact_list)
            
            # Step 2: Generate productivity optimization strategies
            productivity_strategies = await self._generate_productivity_strategies(enhanced_contacts)
            
            # Step 3: Create CRM enrichment recommendations
            crm_optimization = self._generate_crm_optimization(enhanced_contacts, crm_system)
            
            # Step 4: Implement tracking optimization
            tracking_optimization = await self._optimize_yesware_tracking(enhanced_contacts)
            
            return {
                "total_contacts": len(contact_list),
                "enhanced_contacts": len(enhanced_contacts),
                "productivity_stats": self._calculate_sales_productivity_stats(enhanced_contacts),
                "productivity_strategies": productivity_strategies,
                "crm_optimization": crm_optimization,
                "tracking_optimization": tracking_optimization
            }
            
        except Exception as e:
            return {"error": f"Sales productivity optimization failed: {e}"}
    
    async def create_enhanced_yesware_campaign(self, campaign_name: str,
                                             contacts: List[Dict],
                                             campaign_config: Dict = None) -> Dict:
        """
        Create enhanced Yesware campaign with comprehensive productivity optimization
        """
        try:
            # Step 1: Enhance contacts with sales intelligence
            enhanced_contacts = await self._bulk_enhance_for_sales_productivity(contacts)
            
            # Step 2: Generate optimized campaign structure
            campaign_structure = await self._generate_productivity_campaign_structure(
                enhanced_contacts, campaign_config
            )
            
            # Step 3: Create Yesware campaign with optimization
            campaign_data = await self._create_yesware_campaign(campaign_name, campaign_structure)
            
            if not campaign_data:
                return {"error": "Failed to create Yesware campaign"}
            
            # Step 4: Apply productivity optimizations
            optimization_results = await self._apply_sales_productivity_optimizations(
                campaign_data['id'], enhanced_contacts
            )
            
            return {
                "campaign_id": campaign_data['id'],
                "campaign_name": campaign_name,
                "total_contacts": len(enhanced_contacts),
                "sales_productivity_optimization": self._analyze_sales_productivity_optimization(enhanced_contacts),
                "optimization_results": optimization_results
            }
            
        except Exception as e:
            return {"error": f"Enhanced campaign creation failed: {e}"}
    
    async def _bulk_enhance_for_sales_productivity(self, contacts: List[Dict]) -> List[ProductivityContact]:
        """Enhance multiple contacts for optimal sales productivity"""
        enhancement_tasks = []
        
        for contact_data in contacts:
            productivity_contact = ProductivityContact(
                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_productivity(productivity_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_productivity(self, contact: ProductivityContact) -> None:
        """Comprehensive single contact enhancement for sales productivity"""
        enhancement_tasks = []
        
        # Email validation with productivity focus
        if contact.email:
            task1 = self._validate_email_for_productivity(contact.email)
            enhancement_tasks.append(task1)
        
        # Phone enrichment for multi-channel productivity
        if contact.first_name and contact.last_name:
            task2 = self._enrich_phone_for_sales_productivity(contact)
            enhancement_tasks.append(task2)
        
        # Sales intelligence gathering
        task3 = self._gather_sales_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.tracking_optimization = results[0].get('tracking_optimization', {})
        
        # 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('business_phone')
        
        # Process sales intelligence
        if len(results) > 2 and not isinstance(results[2], Exception):
            contact.sales_intelligence = results[2]
        
        # Calculate productivity metrics
        contact.productivity_metrics = self._calculate_productivity_metrics(contact)
        contact.crm_enrichment = self._generate_crm_enrichment_data(contact)
        contact.productivity_score = self._calculate_sales_productivity_score(contact)
        contact.sales_priority = self._determine_sales_priority(contact)
        contact.engagement_readiness = self._assess_engagement_readiness(contact)
        contact.crm_sync_status = self._assess_crm_sync_readiness(contact)
    
    async def _validate_email_for_productivity(self, email: str) -> Dict:
        """Advanced email validation for sales productivity optimization"""
        url = f"{self.lookup_base_url}/email/validate"
        headers = {
            'Authorization': f'Bearer {self.lookup_api_key}',
            'Content-Type': 'application/json'
        }
        
        payload = {
            'email': email,
            'sales_productivity_focus': True,
            'tracking_optimization': True,
            'crm_enrichment': True,
            'engagement_prediction': 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()
                    
                    # Generate tracking optimization for Yesware
                    tracking_optimization = {
                        'tracking_reliability': result.get('tracking_confidence', 85),
                        'open_prediction': result.get('open_likelihood', 0),
                        'response_prediction': result.get('response_likelihood', 0),
                        'productivity_impact': result.get('productivity_score', 0),
                        'crm_sync_quality': result.get('crm_compatibility', 'good')
                    }
                    
                    result['tracking_optimization'] = tracking_optimization
                    return result
                return {'deliverable': False, 'tracking_optimization': {'tracking_reliability': 0}}
    
    async def _enrich_phone_for_sales_productivity(self, contact: ProductivityContact) -> Optional[Dict]:
        """Phone enrichment optimized for sales productivity"""
        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,
            'sales_productivity_focus': True,
            'multi_channel_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 productivity impact
                        if result.get('valid'):
                            result['productivity_impact'] = 95 if result.get('direct_dial') else 75
                            result['multi_channel_capability'] = 'high'
                            result['sales_readiness'] = 'excellent'
                        else:
                            result['productivity_impact'] = 0
                            result['multi_channel_capability'] = 'low'
                            result['sales_readiness'] = 'limited'
                        
                        return result
        except:
            pass
        
        return None
    
    async def _gather_sales_intelligence(self, contact: ProductivityContact) -> Optional[Dict]:
        """Gather comprehensive sales intelligence for productivity optimization"""
        url = f"{self.lookup_base_url}/enrich/sales_intelligence"
        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,
            'productivity_analysis': True,
            'sales_readiness_assessment': 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_productivity_metrics(self, contact: ProductivityContact) -> Dict:
        """Calculate comprehensive productivity metrics"""
        metrics = {
            'contact_completeness': 0,
            'engagement_potential': 0,
            'crm_value': 0,
            'tracking_efficiency': 0,
            'multi_channel_capability': 0
        }
        
        # Contact completeness (0-100)
        completeness = 0
        if contact.email: completeness += 25
        if contact.phone: completeness += 25
        if contact.position and contact.company: completeness += 25
        if contact.linkedin_url: completeness += 15
        if contact.sales_intelligence: completeness += 10
        metrics['contact_completeness'] = completeness
        
        # Engagement potential (0-100)
        if contact.email_validation:
            engagement = contact.email_validation.get('engagement_prediction', 0)
            if contact.tracking_optimization:
                engagement += contact.tracking_optimization.get('response_prediction', 0) * 0.3
            metrics['engagement_potential'] = min(engagement, 100)
        
        # CRM value assessment (0-100)
        crm_value = 0
        if contact.position and contact.company: crm_value += 40
        if contact.phone_validation and contact.phone_validation.get('valid'): crm_value += 30
        if contact.email_validation and contact.email_validation.get('deliverable'): crm_value += 30
        metrics['crm_value'] = crm_value
        
        # Tracking efficiency (0-100)
        if contact.tracking_optimization:
            metrics['tracking_efficiency'] = contact.tracking_optimization.get('tracking_reliability', 0)
        
        # Multi-channel capability (0-100)
        channels = 0
        if contact.email_validation and contact.email_validation.get('deliverable'): channels += 40
        if contact.phone_validation and contact.phone_validation.get('valid'): channels += 40
        if contact.linkedin_url: channels += 20
        metrics['multi_channel_capability'] = channels
        
        return metrics
    
    def _generate_crm_enrichment_data(self, contact: ProductivityContact) -> Dict:
        """Generate CRM enrichment data for contact"""
        enrichment = {
            'data_quality_score': 0,
            'sync_priority': 'low',
            'enrichment_fields': [],
            'crm_readiness': 'limited'
        }
        
        # Calculate data quality score
        quality_score = 0
        enrichment_fields = []
        
        if contact.email_validation and contact.email_validation.get('deliverable'):
            quality_score += 25
            enrichment_fields.append('validated_email')
        
        if contact.phone_validation and contact.phone_validation.get('valid'):
            quality_score += 25
            enrichment_fields.append('validated_phone')
        
        if contact.position and contact.company:
            quality_score += 20
            enrichment_fields.append('professional_info')
        
        if contact.sales_intelligence:
            quality_score += 20
            enrichment_fields.append('sales_intelligence')
        
        if contact.linkedin_url:
            quality_score += 10
            enrichment_fields.append('social_profiles')
        
        enrichment['data_quality_score'] = quality_score
        enrichment['enrichment_fields'] = enrichment_fields
        
        # Determine sync priority and CRM readiness
        if quality_score >= 80:
            enrichment['sync_priority'] = 'high'
            enrichment['crm_readiness'] = 'excellent'
        elif quality_score >= 60:
            enrichment['sync_priority'] = 'medium'
            enrichment['crm_readiness'] = 'good'
        elif quality_score >= 40:
            enrichment['sync_priority'] = 'low'
            enrichment['crm_readiness'] = 'fair'
        
        return enrichment
    
    def _calculate_sales_productivity_score(self, contact: ProductivityContact) -> int:
        """Calculate comprehensive sales productivity score"""
        score = 0
        
        # Productivity metrics contribution (0-60 points)
        if contact.productivity_metrics:
            metrics = contact.productivity_metrics
            score += int(metrics.get('contact_completeness', 0) * 0.2)
            score += int(metrics.get('engagement_potential', 0) * 0.15)
            score += int(metrics.get('crm_value', 0) * 0.15)
            score += int(metrics.get('tracking_efficiency', 0) * 0.1)
        
        # Professional level impact (0-25 points)
        if contact.position:
            position_lower = contact.position.lower()
            if any(title in position_lower for title in ['ceo', 'founder', 'president']):
                score += 25
            elif any(title in position_lower for title in ['vp', 'director', 'head']):
                score += 20
            elif any(title in position_lower for title in ['manager', 'lead', 'senior']):
                score += 15
            else:
                score += 10
        
        # Sales intelligence bonus (0-15 points)
        if contact.sales_intelligence:
            intelligence_quality = len([v for v in contact.sales_intelligence.values() if v])
            score += min(intelligence_quality * 3, 15)
        
        return min(score, 100)
    
    def _determine_sales_priority(self, contact: ProductivityContact) -> str:
        """Determine sales priority based on productivity score"""
        if contact.productivity_score >= 85:
            return "critical"
        elif contact.productivity_score >= 70:
            return "high"
        elif contact.productivity_score >= 50:
            return "medium"
        else:
            return "low"
    
    def _assess_engagement_readiness(self, contact: ProductivityContact) -> str:
        """Assess contact's readiness for sales engagement"""
        readiness_score = 0
        
        # Email quality
        if contact.email_validation and contact.email_validation.get('deliverable'):
            readiness_score += 3
        
        # Phone availability
        if contact.phone_validation and contact.phone_validation.get('valid'):
            readiness_score += 3
        
        # Professional information
        if contact.position and contact.company:
            readiness_score += 2
        
        # Sales intelligence
        if contact.sales_intelligence:
            readiness_score += 2
        
        if readiness_score >= 8:
            return "excellent"
        elif readiness_score >= 6:
            return "good"
        elif readiness_score >= 4:
            return "fair"
        else:
            return "limited"
    
    def _assess_crm_sync_readiness(self, contact: ProductivityContact) -> str:
        """Assess contact's readiness for CRM synchronization"""
        if contact.crm_enrichment:
            quality_score = contact.crm_enrichment.get('data_quality_score', 0)
            if quality_score >= 80:
                return "ready"
            elif quality_score >= 60:
                return "needs_minor_enrichment"
            elif quality_score >= 40:
                return "needs_major_enrichment"
        
        return "not_ready"
    
    def _calculate_sales_productivity_stats(self, contacts: List[ProductivityContact]) -> Dict:
        """Calculate comprehensive sales productivity statistics"""
        total = len(contacts)
        if total == 0:
            return {}
        
        high_productivity = sum(1 for c in contacts if c.productivity_score >= 70)
        crm_ready = sum(1 for c in contacts if c.crm_sync_status == "ready")
        engagement_ready = sum(1 for c in contacts if c.engagement_readiness in ['excellent', 'good'])
        multi_channel = sum(1 for c in contacts if c.productivity_metrics and c.productivity_metrics.get('multi_channel_capability', 0) >= 60)
        
        return {
            'total_contacts': total,
            'high_productivity_rate': (high_productivity / total) * 100,
            'crm_ready_rate': (crm_ready / total) * 100,
            'engagement_ready_rate': (engagement_ready / total) * 100,
            'multi_channel_capable': (multi_channel / total) * 100,
            'average_productivity_score': sum(c.productivity_score for c in contacts) / total,
            'critical_priority_contacts': sum(1 for c in contacts if c.sales_priority == 'critical')
        }

# Example usage for sales productivity optimization
async def main():
    integration = Yesware1lookupIntegration(
        yesware_api_key="your_yesware_api_key",
        lookup_api_key="your_1lookup_api_key"
    )
    
    # Sample contacts for productivity optimization
    sample_contacts = [
        {
            'email': 'john@example.com',
            'first_name': 'John',
            'last_name': 'Smith',
            'position': 'Sales Director',
            'company': 'Example Corp'
        },
        # Add more contacts...
    ]
    
    # Optimize sales productivity
    productivity_result = await integration.optimize_sales_productivity(
        contact_list=sample_contacts,
        crm_system="salesforce"
    )
    
    print("Yesware Sales Productivity Optimization Results:")
    print(f"Total Contacts: {productivity_result.get('total_contacts')}")
    print(f"Enhanced Contacts: {productivity_result.get('enhanced_contacts')}")
    
    productivity_stats = productivity_result.get('productivity_stats', {})
    print(f"
Productivity Statistics:")
    print(f"High Productivity Rate: {productivity_stats.get('high_productivity_rate', 0):.1f}%")
    print(f"CRM Ready Rate: {productivity_stats.get('crm_ready_rate', 0):.1f}%")
    print(f"Multi-Channel Capable: {productivity_stats.get('multi_channel_capable', 0):.1f}%")
    print(f"Average Productivity Score: {productivity_stats.get('average_productivity_score', 0):.1f}")

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