tomgranot/hubspot-admin-skills/delete-no-email-contacts

"""
Before State: Count contacts with no email address.
"""
import os
import json
import requests
from dotenv import load_dotenv
load_dotenv()
TOKEN = os.environ["HUBSPOT_ACCESS_TOKEN"]
BASE = "https://api.hubapi.com"
headers = {
    "Authorization": f"Bearer {TOKEN}",
    "Content-Type": "application/json",
}
search_payload = {
    "filterGroups": [
        {
            "filters": [
                {
                    "propertyName": "email",
                    "operator": "NOT_HAS_PROPERTY",
                }
            ]
        }
    ],
    "properties": ["firstname", "lastname", "createdate", "hs_object_id"],
    "limit": 1,  # Only need the total count
}
url = f"{BASE}/crm/v3/objects/contacts/search"
response = requests.post(url, headers=headers, json=search_payload)
response.raise_for_status()
data = response.json()
total = data.get("total", 0)
print(f"BEFORE STATE: {total} contacts exist with no email address.")
if total > 0 and data.get("results"):
    sample = data["results"][0]
    props = sample.get("properties", {})
    print(f"  Sample: ID {sample['id']}, "
          f"{props.get('firstname', '(empty)')} {props.get('lastname', '(empty)')}, "
          f"created {props.get('createdate', '(unknown)')}")

"""
Execute: Delete all contacts with no email address.
Steps:
  1. Paginated search to collect all contact IDs
  2. Export CSV audit log before deletion
  3. Batch archive in groups of 100
"""
import os
import csv
import time
import requests
from dotenv import load_dotenv
load_dotenv()
TOKEN = os.environ["HUBSPOT_ACCESS_TOKEN"]
BASE = "https://api.hubapi.com"
headers = {
    "Authorization": f"Bearer {TOKEN}",
    "Content-Type": "application/json",
}
# --- Step 1: Collect all contact IDs ---
all_contacts = []
after = None
search_payload = {
    "filterGroups": [
        {
            "filters": [
                {
                    "propertyName": "email",
                    "operator": "NOT_HAS_PROPERTY",
                }
            ]
        }
    ],
    "properties": ["firstname", "lastname", "createdate", "hs_object_id"],
    "limit": 100,
}
while True:
    payload = search_payload.copy()
    if after:
        payload["after"] = after
    resp = requests.post(
        f"{BASE}/crm/v3/objects/contacts/search",
        headers=headers, json=payload,
    )
    resp.raise_for_status()
    data = resp.json()
    for contact in data.get("results", []):
        props = contact.get("properties", {})
        all_contacts.append({
            "id": contact["id"],
            "firstname": props.get("firstname", ""),
            "lastname": props.get("lastname", ""),
            "createdate": props.get("createdate", ""),
        })
    paging = data.get("paging", {})
    after = paging.get("next", {}).get("after")
    if not after:
        break
    time.sleep(0.2)  # Rate limiting
print(f"Total contacts to delete: {len(all_contacts)}")
# --- Step 2: SAFETY CHECK ---
ABORT_THRESHOLD = 500
if len(all_contacts) > ABORT_THRESHOLD:
    print(f"SAFETY ABORT: Found {len(all_contacts)} contacts, "
          f"exceeds threshold of {ABORT_THRESHOLD}.")
    print("Review the data and adjust the threshold if this is expected.")
    exit(1)
# --- Step 3: Export CSV audit trail ---
os.makedirs("data/audit-logs", exist_ok=True)
csv_path = "data/audit-logs/deleted-no-email-contacts.csv"
with open(csv_path, "w", newline="") as f:
    writer = csv.DictWriter(f, fieldnames=["id", "firstname", "lastname", "createdate"])
    writer.writeheader()
    writer.writerows(all_contacts)
print(f"Audit log saved: {csv_path} ({len(all_contacts)} records)")
# --- Step 4: Batch delete ---
all_ids = [c["id"] for c in all_contacts]
BATCH_SIZE = 100
deleted_count = 0
failed_ids = []
for i in range(0, len(all_ids), BATCH_SIZE):
    batch = all_ids[i : i + BATCH_SIZE]
    delete_payload = {"inputs": [{"id": cid} for cid in batch]}
    resp = requests.post(
        f"{BASE}/crm/v3/objects/contacts/batch/archive",
        headers=headers, json=delete_payload,
    )
    if resp.status_code == 204:
        deleted_count += len(batch)
        print(f"  Batch {i // BATCH_SIZE + 1}: deleted {len(batch)} contacts")
    else:
        failed_ids.extend(batch)
        print(f"  Batch FAILED: {resp.status_code} — {resp.text[:200]}")
    time.sleep(0.5)  # Rate limiting between batches
print(f"\nDeleted: {deleted_count}, Failed: {len(failed_ids)}")

"""
After State: Verify no contacts with missing email remain.
"""
# (Same search payload as Before State)
response = requests.post(url, headers=headers, json=search_payload)
response.raise_for_status()
total = response.json().get("total", 0)
if total == 0:
    print("SUCCESS: 0 contacts with no email remain.")
else:
    print(f"WARNING: {total} contacts with no email still exist.")
    print("New contacts may have been created since deletion. Investigate.")

HUBSPOT_ACCESS_TOKEN=pat-na1-xxxxxxxx

uv run scripts/before.py