Identify website visitors with Clearbit Reveal and create HubSpot companies using an agent skill

mkdir -p .claude/skills/identify-visitors/scripts

---
name: identify-visitors
description: Resolve website visitor IPs to companies via Clearbit Reveal and create matching companies in HubSpot.
disable-model-invocation: true
allowed-tools: Bash(python *)
---
 
Identify companies from website visitor IPs and add them to HubSpot.
 
Usage: /identify-visitors <path_to_log_or_csv>
 
1. Run: `python $SKILL_DIR/scripts/reveal_visitors.py --input "$1"`
2. Review the output showing resolved companies and ICP matches
3. Confirm new companies were created in HubSpot

#!/usr/bin/env python3
"""
Visitor Identification: Clearbit Reveal → HubSpot
Resolves visitor IPs to companies, filters by ICP, creates HubSpot records.
"""
import argparse
import csv
import os
import re
import sys
import time
 
try:
    import requests
except ImportError:
    os.system("pip install requests -q")
    import requests
 
# --- Config ---
CLEARBIT_API_KEY = os.environ.get("CLEARBIT_API_KEY")
HUBSPOT_TOKEN = os.environ.get("HUBSPOT_TOKEN")
 
if not all([CLEARBIT_API_KEY, HUBSPOT_TOKEN]):
    print("ERROR: Set CLEARBIT_API_KEY and HUBSPOT_TOKEN environment variables")
    sys.exit(1)
 
HS_HEADERS = {"Authorization": f"Bearer {HUBSPOT_TOKEN}", "Content-Type": "application/json"}
CB_HEADERS = {"Authorization": f"Bearer {CLEARBIT_API_KEY}"}
 
MIN_EMPLOYEES = 50  # Adjust to your ICP
 
# --- Functions ---
def extract_ips(input_path):
    """Extract unique IPs from a log file or CSV."""
    ips = set()
    with open(input_path) as f:
        for line in f:
            # Match IPv4 addresses
            found = re.findall(r'\b(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})\b', line)
            ips.update(found)
 
    # Filter out private/reserved IPs
    public_ips = set()
    for ip in ips:
        octets = [int(o) for o in ip.split(".")]
        if octets[0] == 10:
            continue
        if octets[0] == 172 and 16 <= octets[1] <= 31:
            continue
        if octets[0] == 192 and octets[1] == 168:
            continue
        if octets[0] == 127:
            continue
        public_ips.add(ip)
 
    return public_ips
 
 
def reveal_company(ip):
    """Resolve IP to company via Clearbit Reveal."""
    resp = requests.get(
        "https://reveal.clearbit.com/v1/companies/find",
        params={"ip": ip},
        headers=CB_HEADERS,
    )
    if resp.status_code == 404:
        return None
    resp.raise_for_status()
 
    company = resp.json().get("company")
    if not company or company.get("type") != "company":
        return None
 
    return {
        "domain": company.get("domain"),
        "name": company.get("name"),
        "industry": company.get("category", {}).get("industry"),
        "employees": company.get("metrics", {}).get("employees"),
        "city": company.get("geo", {}).get("city"),
        "state": company.get("geo", {}).get("state"),
        "country": company.get("geo", {}).get("country"),
    }
 
 
def company_exists(domain):
    """Check if company exists in HubSpot by domain."""
    resp = requests.post(
        "https://api.hubapi.com/crm/v3/objects/companies/search",
        headers=HS_HEADERS,
        json={"filterGroups": [{"filters": [{
            "propertyName": "domain", "operator": "EQ", "value": domain,
        }]}]},
    )
    resp.raise_for_status()
    return len(resp.json().get("results", [])) > 0
 
 
def create_company(company):
    """Create company in HubSpot."""
    resp = requests.post(
        "https://api.hubapi.com/crm/v3/objects/companies",
        headers=HS_HEADERS,
        json={"properties": {
            "domain": company["domain"],
            "name": company["name"],
            "industry": company.get("industry", ""),
            "numberofemployees": str(company.get("employees", "")),
            "city": company.get("city", ""),
            "state": company.get("state", ""),
            "country": company.get("country", ""),
        }},
    )
    resp.raise_for_status()
    return resp.json()["id"]
 
 
# --- Main ---
parser = argparse.ArgumentParser()
parser.add_argument("--input", required=True, help="Path to log file or CSV with visitor IPs")
parser.add_argument("--min-employees", type=int, default=50, help="Minimum employee count for ICP")
args = parser.parse_args()
 
MIN_EMPLOYEES = args.min_employees
 
ips = extract_ips(args.input)
print(f"Found {len(ips)} unique public IPs\n")
 
created = 0
already_exists = 0
no_match = 0
unresolved = 0
seen_domains = set()
 
for ip in sorted(ips):
    company = reveal_company(ip)
    time.sleep(0.1)  # Respect rate limits
 
    if not company:
        unresolved += 1
        continue
 
    domain = company["domain"]
    if domain in seen_domains:
        continue
    seen_domains.add(domain)
 
    employees = company.get("employees") or 0
    if employees < MIN_EMPLOYEES:
        no_match += 1
        continue
 
    if company_exists(domain):
        print(f"  EXISTS: {company['name']} ({domain})")
        already_exists += 1
        continue
 
    company_id = create_company(company)
    print(f"  CREATED: {company['name']} — {employees} employees — ID: {company_id}")
    created += 1
 
print(f"\n--- Summary ---")
print(f"IPs processed: {len(ips)}")
print(f"Companies created: {created}")
print(f"Already in HubSpot: {already_exists}")
print(f"Below ICP threshold: {no_match}")
print(f"Unresolved IPs: {unresolved}")

# Process yesterday's access log
/identify-visitors /var/log/nginx/access.log
 
# Process a CSV export from your analytics tool
/identify-visitors ~/downloads/visitor-ips.csv
 
# With a custom employee threshold
python .claude/skills/identify-visitors/scripts/reveal_visitors.py \
  --input /var/log/nginx/access.log --min-employees 100

# crontab — run daily at 7 AM, process previous day's log
0 7 * * * cd /path/to/project && python .claude/skills/identify-visitors/scripts/reveal_visitors.py \
  --input /var/log/nginx/access.log