Enrich HubSpot companies with technographic data from BuiltWith using code

# Test the BuiltWith API
curl "https://api.builtwith.com/v21/api.json?KEY=$BUILTWITH_API_KEY&LOOKUP=hubspot.com" \
  | python3 -m json.tool | head -50

import requests
import os
import time
from datetime import datetime
 
HUBSPOT_ACCESS_TOKEN = os.environ["HUBSPOT_ACCESS_TOKEN"]
BUILTWITH_API_KEY = os.environ["BUILTWITH_API_KEY"]
HS_HEADERS = {"Authorization": f"Bearer {HUBSPOT_ACCESS_TOKEN}", "Content-Type": "application/json"}
 
def get_companies_without_tech(limit=50):
    companies = []
    after = 0
 
    while len(companies) < limit:
        resp = requests.post(
            "https://api.hubapi.com/crm/v3/objects/companies/search",
            headers=HS_HEADERS,
            json={
                "filterGroups": [{"filters": [{
                    "propertyName": "tech_stack_crm",
                    "operator": "NOT_HAS_PROPERTY"
                }]}],
                "properties": ["domain", "name"],
                "limit": min(100, limit - len(companies)),
                "after": after
            }
        )
        resp.raise_for_status()
        data = resp.json()
        companies.extend(data["results"])
 
        if data.get("paging", {}).get("next"):
            after = data["paging"]["next"]["after"]
        else:
            break
 
    return companies

CATEGORY_MAP = {
    "crm": ["crm"],
    "marketing": ["marketing-automation", "email", "marketing"],
    "analytics": ["analytics", "web-analytics"],
    "ecommerce": ["ecommerce", "shopping-cart", "payment"],
    "hosting": ["hosting", "cdn", "cloud-paas"],
}
 
def lookup_tech_stack(domain):
    """Call BuiltWith and categorize technologies."""
    resp = requests.get(
        "https://api.builtwith.com/v21/api.json",
        params={"KEY": BUILTWITH_API_KEY, "LOOKUP": domain}
    )
    resp.raise_for_status()
    data = resp.json()
 
    # Extract technologies from the nested response
    results = data.get("Results", [])
    if not results:
        return None
 
    paths = results[0].get("Result", {}).get("Paths", [])
    if not paths:
        return None
 
    technologies = paths[0].get("Technologies", [])
    if not technologies:
        return None
 
    # Categorize
    categorized = {cat: [] for cat in CATEGORY_MAP}
    for tech in technologies:
        tag = (tech.get("Tag") or "").lower()
        cats = [c.lower() for c in (tech.get("Categories") or [])]
        for category, keywords in CATEGORY_MAP.items():
            if any(kw in tag for kw in keywords) or any(kw in c for kw in keywords for c in cats):
                categorized[category].append(tech["Name"])
                break
 
    return {
        "tech_stack_crm": ", ".join(categorized["crm"]) or "None detected",
        "tech_stack_marketing": ", ".join(categorized["marketing"]) or "None detected",
        "tech_stack_analytics": ", ".join(categorized["analytics"]) or "None detected",
        "tech_stack_all": ", ".join(t["Name"] for t in technologies),
        "tech_count": len(technologies),
    }

def update_company_tech(company_id, tech_data):
    """Write tech stack data to HubSpot company."""
    properties = {
        **tech_data,
        "tech_enrichment_date": datetime.now().strftime("%Y-%m-%d"),
    }
    # Remove tech_count from HubSpot update (internal metric only)
    properties.pop("tech_count", None)
 
    resp = requests.patch(
        f"https://api.hubapi.com/crm/v3/objects/companies/{company_id}",
        headers=HS_HEADERS,
        json={"properties": properties}
    )
    resp.raise_for_status()
 
def main():
    companies = get_companies_without_tech(limit=50)
    print(f"Found {len(companies)} companies to enrich\n")
 
    enriched = 0
    skipped = 0
 
    for company in companies:
        domain = company["properties"].get("domain")
        name = company["properties"].get("name", "Unknown")
 
        if not domain:
            print(f"  {name} — no domain, skipping")
            skipped += 1
            continue
 
        tech_data = lookup_tech_stack(domain)
        if not tech_data:
            print(f"  {name} ({domain}) — no tech data found")
            skipped += 1
            time.sleep(2)
            continue
 
        update_company_tech(company["id"], tech_data)
        enriched += 1
        print(f"  {name} ({domain}) — {tech_data['tech_count']} technologies")
        print(f"    CRM: {tech_data['tech_stack_crm']}")
        print(f"    Marketing: {tech_data['tech_stack_marketing']}")
        print(f"    Analytics: {tech_data['tech_stack_analytics']}")
 
        time.sleep(2)  # BuiltWith rate limit
 
    print(f"\nDone. Enriched: {enriched}, Skipped: {skipped}")
 
if __name__ == "__main__":
    main()