删除无邮箱联系人

目的

在HubSpot营销中心实例中，没有邮箱地址的联系人没有任何功能用途。他们无法接收营销邮件、销售序列或事务性消息。他们只会增加计费的联系人数量。此技能通过API识别并删除这些联系人。

前提条件

• - 具有crm.objects.contacts.read和crm.objects.contacts.write权限的HubSpot私有应用访问令牌
• Python 3.10+，使用uv进行包管理
• 包含HUBSPOTACCESSTOKEN的.env文件

执行模式

此技能遵循4阶段执行模式：计划 -> 之前状态 -> 执行 -> 之后状态。

阶段1：计划

在编写任何代码之前，请与用户确认以下事项：

1. 1. 根本原因：询问是否有任何集成（CRM同步、表单工具、导入流程）在有意创建无邮箱联系人。如果是，请先修复数据流入问题。
2. 阈值：默认安全中止阈值为500个联系人。如果用户预期更多，请在执行脚本中调整阈值。
3. 恢复窗口：确认用户了解已删除的联系人可在90天内通过HubSpot设置 > 数据管理 > 已删除对象进行恢复。

阶段2：之前状态

运行计数查询以建立基线。保存结果以供比较。

python

之前状态：统计无邮箱地址的联系人数量。

import os
import json
import requests
from dotenv import load_dotenv

load_dotenv()

TOKEN = os.environ[HUBSPOTACCESSTOKEN]
BASE = https://api.hubapi.com
headers = {
Authorization: fBearer {TOKEN},
Content-Type: application/json,
}

search_payload = {
filterGroups: [
{
filters: [
{
propertyName: email,
operator: NOTHASPROPERTY,
}
]
}
],
properties: [firstname, lastname, createdate, hsobjectid],
limit: 1, # 只需要总数
}

url = f{BASE}/crm/v3/objects/contacts/search
response = requests.post(url, headers=headers, json=search_payload)
response.raiseforstatus()

data = response.json()
total = data.get(total, 0)

print(f之前状态：存在 {total} 个无邮箱地址的联系人。)

if total > 0 and data.get(results):
sample = data[results][0]
props = sample.get(properties, {})
print(f 示例：ID {sample[id]}，
f{props.get(firstname, (空))} {props.get(lastname, (空))}，
f创建于 {props.get(createdate, (未知))})

预期输出：无邮箱联系人的数量以及用于合理性检查的示例记录。

在继续之前向用户展示结果。请求明确的确认以继续。

阶段3：执行

通过分页搜索收集所有联系人ID，导出CSV审计追踪，然后批量删除。

python

执行：删除所有无邮箱地址的联系人。
步骤：
1. 分页搜索收集所有联系人ID
2. 在删除前导出CSV审计日志
3. 按每批100个进行批量归档

import os
import csv
import time
import requests
from dotenv import load_dotenv

load_dotenv()

TOKEN = os.environ[HUBSPOTACCESSTOKEN]
BASE = https://api.hubapi.com
headers = {
Authorization: fBearer {TOKEN},
Content-Type: application/json,
}

--- 步骤1：收集所有联系人ID ---

all_contacts = [] after = None

search_payload = {
filterGroups: [
{
filters: [
{
propertyName: email,
operator: NOTHASPROPERTY,
}
]
}
],
properties: [firstname, lastname, createdate, hsobjectid],
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.raiseforstatus()
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) # 速率限制

print(f待删除的联系人总数：{len(all_contacts)})

--- 步骤2：安全检查 ---

ABORT_THRESHOLD = 500 if len(allcontacts) > ABORTTHRESHOLD: print(f安全中止：发现 {len(all_contacts)} 个联系人， f超过阈值 {ABORT_THRESHOLD}。) print(请审查数据，如果这是预期情况，请调整阈值。) exit(1)

--- 步骤3：导出CSV审计追踪 ---

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审计日志已保存：{csvpath}（{len(allcontacts)} 条记录）)

--- 步骤4：批量删除 ---

allids = [c[id] for c in allcontacts] BATCH_SIZE = 100 deleted_count = 0 failed_ids = []

for i in range(0, len(allids), BATCHSIZE):
batch = allids[i : i + BATCHSIZE]
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 批次 {i // BATCH_SIZE + 1}：已删除 {len(batch)} 个联系人)
else:
failed_ids.extend(batch)
print(f 批次失败：{resp.status_code} — {resp.text[:200]})

time.sleep(0.5) # 批次间的速率限制

print(f\n已删除：{deletedcount}，失败：{len(failedids)})

关键API详情：

• - POST /crm/v3/objects/contacts/search，使用email字段的NOTHASPROPERTY过滤器
• 使用after游标进行分页，每页100条结果
• POST /crm/v3/objects/contacts/batch/archive每次调用最多接受100个ID
• 成功归档返回HTTP 204（无内容）

阶段4：之后状态

重新运行之前状态的查询，确认剩余联系人为零。

python

之后状态：验证没有无邮箱的联系人剩余。

（与之前状态相同的搜索负载）

response = requests.post(url, headers=headers, json=search_payload) response.raiseforstatus() total = response.json().get(total, 0)

if total == 0:
print(成功：剩余0个无邮箱联系人。)
else:
print(f警告：仍存在 {total} 个无邮箱联系人。)
print(自删除以来可能创建了新的联系人。请调查原因。)

向用户展示结果。 如果出现了新的联系人，请调查来源（表单提交、集成、导入）。

安全机制

机制	详情
中止阈值	默认硬编码为500个联系人。如果搜索返回更多，脚本将退出而不删除任何内容。仅在获得用户明确确认后调整。
CSV审计追踪

在进行任何删除之前，每个联系人的ID、姓名和创建日期都会导出到CSV。 | | 确认提示 | 始终向用户展示之前状态计数，并在运行执行前等待明确确认。 | | 90天恢复 | 已删除的联系人可在90天内通过HubSpot设置 > 数据管理 > 已删除对象进行恢复。 | | 已归档联系人审计 | 删除后，您