Find and verify emails for HubSpot prospects using Apollo and Hunter with code

# Test Apollo people match
curl -X POST "https://api.apollo.io/api/v1/people/match" \
  -H "x-api-key: $APOLLO_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"first_name": "Tim", "last_name": "Cook", "organization_name": "Apple"}'
 
# Test Hunter email verification
curl "https://api.hunter.io/v2/email-verifier?email=test@example.com&api_key=$HUNTER_API_KEY"

import requests
import os
import time
 
HUBSPOT_ACCESS_TOKEN = os.environ["HUBSPOT_ACCESS_TOKEN"]
APOLLO_API_KEY = os.environ["APOLLO_API_KEY"]
HUNTER_API_KEY = os.environ["HUNTER_API_KEY"]
 
HS_HEADERS = {"Authorization": f"Bearer {HUBSPOT_ACCESS_TOKEN}", "Content-Type": "application/json"}
 
def get_contacts_without_email(limit=50):
    all_contacts = []
    after = 0
 
    while len(all_contacts) < limit:
        resp = requests.post(
            "https://api.hubapi.com/crm/v3/objects/contacts/search",
            headers=HS_HEADERS,
            json={
                "filterGroups": [{"filters": [
                    {"propertyName": "email", "operator": "NOT_HAS_PROPERTY"},
                    {"propertyName": "firstname", "operator": "HAS_PROPERTY"},
                    {"propertyName": "company", "operator": "HAS_PROPERTY"},
                ]}],
                "properties": ["firstname", "lastname", "company", "domain"],
                "limit": min(100, limit - len(all_contacts)),
                "after": after
            }
        )
        resp.raise_for_status()
        data = resp.json()
        all_contacts.extend(data["results"])
 
        if data.get("paging", {}).get("next"):
            after = data["paging"]["next"]["after"]
        else:
            break
 
    return all_contacts

def find_email_apollo(first_name, last_name, company):
    """Find a person's email via Apollo People Match."""
    resp = requests.post(
        "https://api.apollo.io/api/v1/people/match",
        headers={"x-api-key": APOLLO_API_KEY, "Content-Type": "application/json"},
        json={
            "first_name": first_name,
            "last_name": last_name,
            "organization_name": company
        }
    )
    resp.raise_for_status()
    person = resp.json().get("person")
    if not person or not person.get("email"):
        return None, None
 
    return person["email"], person.get("email_status", "unknown")

def verify_email_hunter(email):
    """Verify an email via Hunter. Returns 'deliverable', 'risky', 'undeliverable', or 'unknown'."""
    resp = requests.get(
        "https://api.hunter.io/v2/email-verifier",
        params={"email": email, "api_key": HUNTER_API_KEY}
    )
    resp.raise_for_status()
    return resp.json()["data"]["result"]
 
def find_and_verify(first_name, last_name, company):
    """Find email via Apollo, verify via Hunter if needed."""
    email, apollo_status = find_email_apollo(first_name, last_name, company)
 
    if not email:
        return {"email": None, "status": "not_found", "source": "apollo"}
 
    # Skip Hunter if Apollo already verified
    if apollo_status == "verified":
        return {"email": email, "status": "verified", "source": "apollo"}
 
    # Verify with Hunter
    hunter_result = verify_email_hunter(email)
    return {
        "email": email,
        "status": hunter_result,
        "source": "apollo+hunter"
    }

def update_contact_email(contact_id, email, status, source):
    """Write email and verification status to HubSpot."""
    properties = {"email_source": source, "email_verification_status": status}
 
    if status == "deliverable" or status == "verified":
        properties["email"] = email
 
    resp = requests.patch(
        f"https://api.hubapi.com/crm/v3/objects/contacts/{contact_id}",
        headers=HS_HEADERS,
        json={"properties": properties}
    )
    resp.raise_for_status()
 
def main():
    contacts = get_contacts_without_email(limit=50)
    print(f"Found {len(contacts)} contacts without email\n")
 
    stats = {"found": 0, "verified": 0, "risky": 0, "not_found": 0}
 
    for contact in contacts:
        props = contact["properties"]
        name = f"{props.get('firstname', '')} {props.get('lastname', '')}".strip()
        company = props.get("company", "")
 
        result = find_and_verify(
            props.get("firstname", ""),
            props.get("lastname", ""),
            company
        )
 
        if result["status"] in ("deliverable", "verified"):
            update_contact_email(contact["id"], result["email"], result["status"], result["source"])
            stats["verified"] += 1
            print(f"  {name} @ {company} -> {result['email']} ({result['source']})")
        elif result["status"] == "risky":
            update_contact_email(contact["id"], result["email"], "risky", result["source"])
            stats["risky"] += 1
            print(f"  {name} @ {company} -> {result['email']} (RISKY)")
        else:
            stats["not_found"] += 1
            print(f"  {name} @ {company} -> not found")
 
        time.sleep(0.5)  # rate limit
 
    print(f"\nDone. Verified: {stats['verified']}, Risky: {stats['risky']}, Not found: {stats['not_found']}")
 
if __name__ == "__main__":
    main()