Batch enrich HubSpot contacts missing job title or company size using code

# Verify HubSpot search with NOT_HAS_PROPERTY
curl -s -X POST "https://api.hubapi.com/crm/v3/objects/contacts/search" \
  -H "Authorization: Bearer $HUBSPOT_ACCESS_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "filterGroups": [{"filters": [{"propertyName": "jobtitle", "operator": "NOT_HAS_PROPERTY"}]}],
    "properties": ["email", "jobtitle"],
    "limit": 5
  }' | python3 -m json.tool

import requests
import os
import time
from datetime import datetime
 
HUBSPOT_ACCESS_TOKEN = os.environ["HUBSPOT_ACCESS_TOKEN"]
APOLLO_API_KEY = os.environ["APOLLO_API_KEY"]
HS_HEADERS = {"Authorization": f"Bearer {HUBSPOT_ACCESS_TOKEN}", "Content-Type": "application/json"}
 
CRITICAL_FIELDS = ["jobtitle", "company", "phone", "industry"]
 
def get_contacts_missing_fields(field="jobtitle", limit=200):
    """Find contacts missing a specific field using NOT_HAS_PROPERTY."""
    contacts = []
    after = 0
 
    while len(contacts) < limit:
        resp = requests.post(
            "https://api.hubapi.com/crm/v3/objects/contacts/search",
            headers=HS_HEADERS,
            json={
                "filterGroups": [{"filters": [{
                    "propertyName": field,
                    "operator": "NOT_HAS_PROPERTY"
                }]}],
                "properties": ["email", "firstname", "lastname"] + CRITICAL_FIELDS,
                "limit": min(100, limit - len(contacts)),
                "after": after
            }
        )
        resp.raise_for_status()
        data = resp.json()
        contacts.extend(data["results"])
 
        if data.get("paging", {}).get("next"):
            after = data["paging"]["next"]["after"]
            time.sleep(0.2)  # HubSpot Search rate limit
        else:
            break
 
    return contacts

def bulk_enrich_apollo(contacts_batch):
    """Enrich up to 10 contacts in a single Apollo API call."""
    details = []
    for contact in contacts_batch:
        email = contact["properties"].get("email")
        if email:
            details.append({"email": email})
 
    if not details:
        return []
 
    resp = requests.post(
        "https://api.apollo.io/api/v1/people/bulk_match",
        headers={"x-api-key": APOLLO_API_KEY, "Content-Type": "application/json"},
        json={"details": details}
    )
    resp.raise_for_status()
    return resp.json().get("matches", [])
 
def enrich_batch(contacts):
    """Process contacts in batches of 10 using Apollo bulk endpoint."""
    results = []
    for i in range(0, len(contacts), 10):
        batch = contacts[i:i+10]
        matches = bulk_enrich_apollo(batch)
 
        for contact, match in zip(batch, matches):
            if match:
                results.append({"contact": contact, "match": match})
 
        time.sleep(1)  # rate limit between bulk calls
        print(f"  Processed {min(i+10, len(contacts))}/{len(contacts)}")
 
    return results

def update_contact_fields(contact_id, existing_props, apollo_match):
    """Write only fields that are currently empty."""
    properties = {}
 
    field_map = {
        "jobtitle": lambda m: m.get("title"),
        "company": lambda m: m.get("organization", {}).get("name"),
        "phone": lambda m: (m.get("phone_numbers") or [{}])[0].get("sanitized_number"),
        "linkedin_url": lambda m: m.get("linkedin_url"),
        "industry": lambda m: m.get("organization", {}).get("industry"),
    }
 
    for hs_field, extractor in field_map.items():
        existing_value = existing_props.get(hs_field)
        if not existing_value:  # only fill if empty
            new_value = extractor(apollo_match)
            if new_value:
                properties[hs_field] = new_value
 
    if not properties:
        return 0
 
    properties["enrichment_date"] = datetime.now().strftime("%Y-%m-%d")
    properties["enrichment_source"] = "apollo-batch"
 
    resp = requests.patch(
        f"https://api.hubapi.com/crm/v3/objects/contacts/{contact_id}",
        headers=HS_HEADERS,
        json={"properties": properties}
    )
    resp.raise_for_status()
    return len(properties) - 2  # subtract enrichment_date and enrichment_source

def main():
    print(f"[{datetime.now().isoformat()}] Starting batch enrichment...")
 
    contacts = get_contacts_missing_fields(field="jobtitle", limit=200)
    print(f"Found {len(contacts)} contacts missing job title")
 
    # Filter out personal emails
    business_contacts = [
        c for c in contacts
        if c["properties"].get("email") and
        c["properties"]["email"].split("@")[-1].lower() not in
        ("gmail.com", "yahoo.com", "hotmail.com", "outlook.com", "aol.com")
    ]
    print(f"After filtering personal emails: {len(business_contacts)} contacts")
 
    enriched_results = enrich_batch(business_contacts)
    print(f"Apollo matched {len(enriched_results)} contacts")
 
    fields_filled = 0
    contacts_updated = 0
    for item in enriched_results:
        count = update_contact_fields(
            item["contact"]["id"],
            item["contact"]["properties"],
            item["match"]
        )
        if count > 0:
            contacts_updated += 1
            fields_filled += count
 
    print(f"\nDone. Updated {contacts_updated} contacts, filled {fields_filled} fields.")
 
if __name__ == "__main__":
    main()