Best GMass 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
import gspread
from google.oauth2.service_account import Credentials

@dataclass
class GmailMarketingContact:
    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
    gmail_intelligence: Optional[Dict] = None
    sheets_data: Optional[Dict] = None
    deliverability_score: int = 0
    gmail_marketing_score: int = 0
    campaign_priority: str = "medium"
    personalization_level: str = "basic"
    gmass_readiness: str = "pending"
    sheets_row_id: Optional[int] = None
    enhancement_timestamp: str = ""

class GMass1lookupIntegration:
    def __init__(self, lookup_api_key: str, google_credentials_path: str):
        self.lookup_api_key = lookup_api_key
        self.lookup_base_url = "https://api.1lookup.app/v1"
        
        # Initialize Google Sheets API
        self.google_creds = Credentials.from_service_account_file(
            google_credentials_path,
            scopes=['https://www.googleapis.com/auth/spreadsheets', 
                   'https://www.googleapis.com/auth/drive']
        )
        self.sheets_client = gspread.authorize(self.google_creds)
    
    async def enhance_google_sheet_for_gmass(self, sheet_id: str, 
                                           worksheet_name: str = "Sheet1") -> Dict:
        """
        Enhance Google Sheet with comprehensive contact validation for GMass campaigns
        """
        try:
            # Step 1: Load data from Google Sheet
            worksheet_data = self._load_google_sheet_data(sheet_id, worksheet_name)
            
            if not worksheet_data:
                return {"error": "No data found in Google Sheet"}
            
            # Step 2: Enhance all contacts with Gmail marketing intelligence
            enhanced_contacts = await self._bulk_enhance_for_gmail_marketing(worksheet_data)
            
            # Step 3: Update Google Sheet with enhanced data
            update_results = await self._update_google_sheet_with_enhancements(
                sheet_id, worksheet_name, enhanced_contacts
            )
            
            # Step 4: Generate GMass campaign recommendations
            campaign_recommendations = self._generate_gmass_campaign_recommendations(enhanced_contacts)
            
            return {
                "sheet_id": sheet_id,
                "worksheet_name": worksheet_name,
                "total_contacts": len(worksheet_data),
                "enhanced_contacts": len(enhanced_contacts),
                "gmail_marketing_stats": self._calculate_gmail_marketing_stats(enhanced_contacts),
                "update_results": update_results,
                "campaign_recommendations": campaign_recommendations
            }
            
        except Exception as e:
            return {"error": f"Google Sheet enhancement failed: {e}"}
    
    async def optimize_gmass_campaign_list(self, contact_list: List[Dict],
                                         campaign_config: Dict = None) -> Dict:
        """
        Optimize contact list for GMass campaigns with comprehensive intelligence
        """
        try:
            # Step 1: Enhance contacts with Gmail marketing intelligence
            enhanced_contacts = await self._bulk_enhance_for_gmail_marketing(contact_list)
            
            # Step 2: Generate Gmail deliverability optimization
            deliverability_optimization = await self._optimize_gmail_deliverability(enhanced_contacts)
            
            # Step 3: Create personalization recommendations
            personalization_recommendations = self._generate_personalization_recommendations(enhanced_contacts)
            
            # Step 4: Generate campaign structure optimization
            campaign_optimization = self._generate_gmail_campaign_optimization(
                enhanced_contacts, campaign_config
            )
            
            return {
                "total_contacts": len(contact_list),
                "enhanced_contacts": len(enhanced_contacts),
                "gmail_marketing_optimization": self._analyze_gmail_marketing_optimization(enhanced_contacts),
                "deliverability_optimization": deliverability_optimization,
                "personalization_recommendations": personalization_recommendations,
                "campaign_optimization": campaign_optimization
            }
            
        except Exception as e:
            return {"error": f"GMass campaign optimization failed: {e}"}
    
    def _load_google_sheet_data(self, sheet_id: str, worksheet_name: str) -> List[Dict]:
        """Load data from Google Sheet"""
        try:
            sheet = self.sheets_client.open_by_key(sheet_id)
            worksheet = sheet.worksheet(worksheet_name)
            records = worksheet.get_all_records()
            
            # Add row IDs for tracking
            for i, record in enumerate(records):
                record['sheets_row_id'] = i + 2  # +2 because sheets are 1-indexed and have header
            
            return records
        except Exception as e:
            print(f"Error loading Google Sheet: {e}")
            return []
    
    async def _bulk_enhance_for_gmail_marketing(self, contacts: List[Dict]) -> List[GmailMarketingContact]:
        """Enhance multiple contacts for optimal Gmail marketing campaigns"""
        enhancement_tasks = []
        
        for contact_data in contacts:
            gmail_contact = GmailMarketingContact(
                email=contact_data.get('email', ''),
                first_name=contact_data.get('first_name') or contact_data.get('firstName') or contact_data.get('First Name'),
                last_name=contact_data.get('last_name') or contact_data.get('lastName') or contact_data.get('Last Name'),
                position=contact_data.get('position') or contact_data.get('title') or contact_data.get('Position'),
                company=contact_data.get('company') or contact_data.get('Company'),
                domain=contact_data.get('domain') or contact_data.get('Domain'),
                linkedin_url=contact_data.get('linkedin_url') or contact_data.get('LinkedIn'),
                sheets_row_id=contact_data.get('sheets_row_id'),
                sheets_data=contact_data,
                enhancement_timestamp=datetime.now().isoformat()
            )
            
            task = self._enhance_single_contact_for_gmail_marketing(gmail_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_gmail_marketing(self, contact: GmailMarketingContact) -> None:
        """Comprehensive single contact enhancement for Gmail marketing"""
        enhancement_tasks = []
        
        # Email validation with Gmail optimization
        if contact.email:
            task1 = self._validate_email_for_gmail_marketing(contact.email)
            enhancement_tasks.append(task1)
        
        # Phone enrichment for multi-channel follow-up
        if contact.first_name and contact.last_name:
            task2 = self._enrich_phone_for_gmail_campaigns(contact)
            enhancement_tasks.append(task2)
        
        # Gmail marketing intelligence gathering
        task3 = self._gather_gmail_marketing_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 Gmail marketing intelligence
        if len(results) > 2 and not isinstance(results[2], Exception):
            contact.gmail_intelligence = results[2]
        
        # Calculate Gmail marketing scores and classifications
        contact.gmail_marketing_score = self._calculate_gmail_marketing_score(contact)
        contact.campaign_priority = self._determine_gmail_campaign_priority(contact)
        contact.personalization_level = self._assess_gmail_personalization_level(contact)
        contact.gmass_readiness = self._assess_gmass_campaign_readiness(contact)
    
    async def _validate_email_for_gmail_marketing(self, email: str) -> Dict:
        """Advanced email validation optimized for Gmail marketing campaigns"""
        url = f"{self.lookup_base_url}/email/validate"
        headers = {
            'Authorization': f'Bearer {self.lookup_api_key}',
            'Content-Type': 'application/json'
        }
        
        payload = {
            'email': email,
            'gmail_optimization': True,
            'mass_email_suitability': True,
            'deliverability_focus': True,
            'gmail_reputation_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 Gmail marketing 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('gmail_friendly', True): score += 25
                    
                    result['deliverability_score'] = score
                    result['gmail_marketing_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, 'gmail_marketing_suitability': 'poor'}
    
    async def _enrich_phone_for_gmail_campaigns(self, contact: GmailMarketingContact) -> Optional[Dict]:
        """Phone enrichment optimized for Gmail marketing 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,
            'gmail_marketing_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 Gmail marketing multi-channel potential
                        if result.get('valid'):
                            result['gmail_followup_score'] = 85 if result.get('mobile') else 65
                            result['multi_channel_potential'] = 'high'
                        else:
                            result['gmail_followup_score'] = 0
                            result['multi_channel_potential'] = 'low'
                        
                        return result
        except:
            pass
        
        return None
    
    async def _gather_gmail_marketing_intelligence(self, contact: GmailMarketingContact) -> Optional[Dict]:
        """Gather comprehensive intelligence for Gmail marketing optimization"""
        url = f"{self.lookup_base_url}/enrich/gmail_marketing"
        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,
            'gmail_personalization_focus': True,
            'mass_email_intelligence': True
        }
        
        try:
            async with aiohttp.ClientSession() as session:
                async with session.post(url, json=payload, headers=headers) as response:
                    if response.status == 200:
                        return await response.json()
        except:
            pass
        
        return None
    
    def _calculate_gmail_marketing_score(self, contact: GmailMarketingContact) -> int:
        """Calculate comprehensive Gmail marketing effectiveness score"""
        score = 0
        
        # Email deliverability (0-40 points)
        score += contact.deliverability_score * 0.4
        
        # Contact completeness (0-30 points)
        completeness = 0
        if contact.email: completeness += 12
        if contact.first_name: completeness += 6
        if contact.last_name: completeness += 4
        if contact.company: completeness += 4
        if contact.position: completeness += 4
        score += completeness
        
        # Multi-channel capability (0-15 points)
        if contact.phone_validation and contact.phone_validation.get('valid'):
            score += 15
        elif contact.linkedin_url:
            score += 8
        
        # Gmail intelligence bonus (0-10 points)
        if contact.gmail_intelligence:
            intelligence_quality = len([v for v in contact.gmail_intelligence.values() if v])
            score += min(intelligence_quality * 2, 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']):
                score += 5
            elif any(title in position_lower for title in ['manager', 'vp', 'head']):
                score += 3
        
        return min(int(score), 100)
    
    def _determine_gmail_campaign_priority(self, contact: GmailMarketingContact) -> str:
        """Determine Gmail campaign priority based on marketing score"""
        if contact.gmail_marketing_score >= 80:
            return "high"
        elif contact.gmail_marketing_score >= 60:
            return "medium"
        else:
            return "low"
    
    def _assess_gmail_personalization_level(self, contact: GmailMarketingContact) -> str:
        """Assess available personalization depth for Gmail campaigns"""
        personalization_factors = 0
        
        if contact.first_name: personalization_factors += 1
        if contact.company: personalization_factors += 1
        if contact.position: personalization_factors += 1
        if contact.gmail_intelligence: personalization_factors += 2
        if contact.linkedin_url: personalization_factors += 1
        
        if personalization_factors >= 5:
            return "ultra_deep"
        elif personalization_factors >= 4:
            return "deep"
        elif personalization_factors >= 2:
            return "medium"
        else:
            return "basic"
    
    def _assess_gmass_campaign_readiness(self, contact: GmailMarketingContact) -> str:
        """Assess contact's readiness for GMass campaigns"""
        readiness_score = 0
        
        # Email deliverability
        if contact.email_validation and contact.email_validation.get('deliverable'):
            readiness_score += 3
        
        # Gmail marketing suitability
        gmail_suitability = contact.email_validation.get('gmail_marketing_suitability', 'poor') if contact.email_validation else 'poor'
        if gmail_suitability == 'excellent':
            readiness_score += 3
        elif gmail_suitability == 'good':
            readiness_score += 2
        
        # Personalization potential
        if contact.personalization_level in ['ultra_deep', 'deep']:
            readiness_score += 2
        elif contact.personalization_level == 'medium':
            readiness_score += 1
        
        if readiness_score >= 7:
            return "excellent"
        elif readiness_score >= 5:
            return "good"
        elif readiness_score >= 3:
            return "fair"
        else:
            return "limited"
    
    async def _update_google_sheet_with_enhancements(self, sheet_id: str, worksheet_name: str, 
                                                   contacts: List[GmailMarketingContact]) -> Dict:
        """Update Google Sheet with enhancement data"""
        try:
            sheet = self.sheets_client.open_by_key(sheet_id)
            worksheet = sheet.worksheet(worksheet_name)
            
            # Get current headers and add new ones if needed
            headers = worksheet.row_values(1)
            new_headers = [
                'Email_Valid', 'Phone_Validated', 'Gmail_Marketing_Score', 
                'Campaign_Priority', 'Personalization_Level', 'GMass_Readiness',
                'Deliverability_Score', 'Enhancement_Timestamp'
            ]
            
            # Add new headers if they don't exist
            for header in new_headers:
                if header not in headers:
                    headers.append(header)
            
            # Update headers
            worksheet.update('1:1', [headers])
            
            # Update contact data
            updates = []
            for contact in contacts:
                if contact.sheets_row_id:
                    row_data = []
                    for header in headers:
                        if header == 'Email_Valid':
                            row_data.append(contact.email_validation.get('deliverable', False) if contact.email_validation else False)
                        elif header == 'Phone_Validated':
                            row_data.append(contact.phone_validation.get('valid', False) if contact.phone_validation else False)
                        elif header == 'Gmail_Marketing_Score':
                            row_data.append(contact.gmail_marketing_score)
                        elif header == 'Campaign_Priority':
                            row_data.append(contact.campaign_priority)
                        elif header == 'Personalization_Level':
                            row_data.append(contact.personalization_level)
                        elif header == 'GMass_Readiness':
                            row_data.append(contact.gmass_readiness)
                        elif header == 'Deliverability_Score':
                            row_data.append(contact.deliverability_score)
                        elif header == 'Enhancement_Timestamp':
                            row_data.append(contact.enhancement_timestamp)
                        else:
                            # Keep existing data
                            original_value = contact.sheets_data.get(header, '')
                            row_data.append(original_value)
                    
                    updates.append({
                        'range': f'{contact.sheets_row_id}:{contact.sheets_row_id}',
                        'values': [row_data]
                    })
            
            # Batch update the sheet
            if updates:
                worksheet.batch_update(updates)
            
            return {
                'success': True,
                'updated_rows': len(updates),
                'new_headers_added': len([h for h in new_headers if h not in worksheet.row_values(1)])
            }
            
        except Exception as e:
            return {'success': False, 'error': str(e)}
    
    def _calculate_gmail_marketing_stats(self, contacts: List[GmailMarketingContact]) -> Dict:
        """Calculate comprehensive Gmail marketing 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.gmail_marketing_score >= 70)
        phone_enhanced = sum(1 for c in contacts if c.phone_validation and c.phone_validation.get('valid'))
        gmass_ready = sum(1 for c in contacts if c.gmass_readiness in ['excellent', 'good'])
        deep_personalization = sum(1 for c in contacts if c.personalization_level in ['ultra_deep', 'deep'])
        
        return {
            'total_contacts': total,
            'deliverable_rate': (deliverable / total) * 100,
            'high_score_rate': (high_score / total) * 100,
            'phone_enhanced_rate': (phone_enhanced / total) * 100,
            'gmass_ready_rate': (gmass_ready / total) * 100,
            'deep_personalization_rate': (deep_personalization / total) * 100,
            'average_gmail_marketing_score': sum(c.gmail_marketing_score for c in contacts) / total,
            'high_priority_contacts': sum(1 for c in contacts if c.campaign_priority == 'high')
        }

# Example usage for GMass Gmail marketing optimization
async def main():
    integration = GMass1lookupIntegration(
        lookup_api_key="your_1lookup_api_key",
        google_credentials_path="path/to/google_credentials.json"
    )
    
    # Enhance Google Sheet for GMass campaigns
    sheet_result = await integration.enhance_google_sheet_for_gmass(
        sheet_id="your_google_sheet_id",
        worksheet_name="GMass Contacts"
    )
    
    print("GMass Google Sheet Enhancement Results:")
    print(f"Sheet ID: {sheet_result.get('sheet_id')}")
    print(f"Total Contacts: {sheet_result.get('total_contacts')}")
    print(f"Enhanced Contacts: {sheet_result.get('enhanced_contacts')}")
    
    stats = sheet_result.get('gmail_marketing_stats', {})
    print(f"
Gmail Marketing Statistics:")
    print(f"Deliverable Rate: {stats.get('deliverable_rate', 0):.1f}%")
    print(f"GMass Ready Rate: {stats.get('gmass_ready_rate', 0):.1f}%")
    print(f"Deep Personalization Rate: {stats.get('deep_personalization_rate', 0):.1f}%")
    print(f"Average Gmail Marketing Score: {stats.get('average_gmail_marketing_score', 0):.1f}")

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