如何大规模爬虫 ZoomInfo

了解如何爬虫 ZoomInfo 公司页面、将 CRM 域名与企业资料匹配,并使用 Python 和 Bright Data 的 ZoomInfo 爬虫 API 验证记录。
7 分钟阅读
如何抓取 Zoominfo

爬虫 B2B 数据平台 ZoomInfo,是指无需登录即可采集其展示的公司资料。销售和数据团队使用 ZoomInfo 公司数据,以行业、员工人数、营收和技术等字段丰富 CRM 账户。

如果没有来自其他网站的 referrer,我们从新的住宅 IP 发出的前 32 个首次请求全部遇到了 ZoomInfo 的“Press & Hold”验证页面。在测试中,成功获取第一个页面的 IP 在爬虫中位数为 3 个页面后被封锁。

本指南将介绍如何大规模获取公司页面、将账户与资料匹配、提取数据,以及何时应获取ZoomInfo 爬虫工具。

简要总结

  • 我们测得每个 IP 大约可获取 3 个页面,因此采集 10 万个 ZoomInfo 页面需要超过 3.3 万个会话。到第 5 个请求时,14 个会话中有 12 个收到了 Cloudflare 429。
  • ZoomInfo 存在重复的公司资料。搜索一个域名返回了 3 个公司 ID,其中一份重复资料列出了不同的员工人数和营收。
  • ZoomInfo 的嵌入式 JSON 包含未取整的数字。网络解锁器返回的 Aviagen 页面在 JSON 中显示有 6,776 名员工,而资料页面显示的是 5K 至 10K。
  • Bright Data 的 ZoomInfo 爬虫 API 会为你处理解锁。它从 1 个目录 URL 返回了 45 条结构化公司记录。

ZoomInfo 在无需登录时显示哪些内容

ZoomInfo 的公开页面提供详细的公司数据和基本的人员数据。电子邮件、直拨电话号码、完整联系人列表和完整技术列表需要账户才能查看。我们在未登录状态下使用爬虫工具采集了行业和员工人数等企业统计数据。该公开网站主要有以下页面类型:

页面类型 URL 格式 可显示的内容
公司资料 /c/{slug}/{id} 行业、员工人数、营收估算、总部、电话、网站、SIC 和 NAICS 代码、部分技术、融资情况,以及按管理层级划分的员工人数
公司目录 /companies-search/{filters} 每页 9 家公司,可按地点和行业筛选
个人资料 /p/{name}/{id} 姓名、职位和雇主
其他页面 /pic/、/hq/、/tech/、/executives/、/competitors/ 员工目录、总部详情、使用某项技术的公司、高管和竞争对手

个人页面、员工目录以及公司页面中的人员列表均包含个人数据,可能受到 GDPR 约束。登录后的页面受 ZoomInfo 服务条款约束,账户持有人需接受这些条款。我们的脚本仅采集公开的公司记录,并丢弃人员字段。

为什么你对 ZoomInfo 的首次请求会被封锁

ZoomInfo 在其页面前部署了 2 套反机器人系统,每个被封锁的响应都会表明由哪套系统发出。位于你的抓取工具与 ZoomInfo 之间的代理也可能改变这些系统看到的 TLS 指纹。在我们的测试中,会话的首次请求需要外部 referrer,但随后携带该会话 cookie 的请求不再需要。

每套反机器人系统都有自己的特征

对 zoominfo.com 页面(包括目录页面)的请求会经由 Cloudflare 路由并由 HUMAN Security 检查,后者于 2022 年与 PerimeterX 合并。两个系统都会在其封锁响应中留下不同的特征,真实公司页面也有自己的特征:

响应来源 状态和页面 用于分支判断的信号
HUMAN Security 403,“Access to this page has been denied”,并带有“Press & Hold”按钮 403 状态以及响应正文中的 px-captcha
Cloudflare 429,“Just a moment…” cf-mitigated: challenge 响应标头
ZoomInfo 源站 200,公司页面 包含公司数据的 <script id="ng-state">

应根据这些信号进行分支判断,而不是根据页面标题。公司页面也包含 _pxAppId因此它的存在并不表示 HUMAN 封锁了请求。Cloudflare 文档说明,每种验证页面都会带有 cf-mitigated: challenge 标头。

ng-state 脚本中的嵌入式 JSON 是识别真实公司页面最可靠的标志。仅当响应包含此 JSON 时,才将其视为成功。

代理如何影响你的 TLS 指纹

我们使用了 Bright Data 住宅代理区域,即一项具名代理配置。该区域会自行与网站建立 TLS 连接。我们使用 3 个客户端向 TLS 指纹检查器 tls.peet.ws 发送请求,先不使用代理,再通过该区域发送:

客户端 不使用代理时的 JA4 通过代理时的 JA4
模拟 Chrome 150 的 curl_cffi t13d1516h2_8daaf6152771_806a8c22fdea t13d1516h2_8daaf6152771_d8a2da3f94cd
模拟 Firefox 147 的 curl_cffi t13d1717h2_5b57614c22b0_3cbfd9057e0d t13d1516h2_8daaf6152771_02713d6af862
系统 curl 8.7.1(LibreSSL) t13d497h2_0d8feac7bc37_7395dae3b2f3 t13d1516h2_8daaf6152771_02713d6af862

通过代理时,全部 3 个 JA4 指纹都以 t13d1516h2_8daaf6152771 开头,前两部分与模拟 Chrome 的 curl_cffi 相同。这些部分表示 TLS 1.3、15 个密码套件、16 个扩展和 HTTP/2。因此,该代理为全部 3 个客户端呈现了类似 Chrome 的 TLS 指纹。Chrome 模拟还通过代理保留了其 HTTP/2 指纹,因此可使用它让 TLS 和 HTTP/2 指纹都呈现为 Chrome 的特征。

在花时间模拟 TLS 之前,先检查哪个节点终止 TLS。如果代理只是通过隧道转发连接,网站会看到客户端自身的握手。在这两种情况下,TCP 连接都从出口节点发起,其 SYN 数据包携带自身的 TTL、窗口大小和选项顺序。macOS User-Agent 可能通过 Windows TCP 栈到达,而读取 TCP 指纹的反机器人系统可能会标记这种不匹配。

ZoomInfo 如何判断首次请求

HUMAN 的验证页面会在浏览器中运行传感器脚本,该脚本会设置 _px3 等风险 cookie。爬虫的首次请求不携带任何此类 cookie,因此网站只能根据请求及其连接进行判断。

我们从新的美国住宅 IP 发出了 32 个不带 cookie、无 referrer 的公司页面请求,4 个客户端各发送 8 个。客户端包括系统 curl、带完整 Chrome 标头的 curl、模拟 Chrome 的 curl_cffi以及通过 HTTP/2 连接的 httpx。全部 32 个请求均收到 HUMAN 的 403 和以下“Press & Hold”页面:

ZoomInfo 的 HUMAN Security 验证页面,其中包含 ZoomInfo 徽标、文字“Press & Hold to confirm you are a human (and not a bot)”以及一个空白的长按按钮。

ZoomInfo 向不带 cookie 且没有 referrer 的请求返回了此验证页面。

通过代理区域时,Cloudflare 看到的都是代理针对 4 个客户端执行的 TLS 握手。通过家庭网络连接时,Cloudflare 看到的是各客户端自身的握手。我们在该连接上测试的全部 6 个客户端,其首次请求都收到了相同的 403。因此,对于无 referrer 的首次请求,TLS 握手并未改变结果。

外部 referrer 改变了结果。当 Referer 标头指向另一个网站时,模拟 Chrome 的 curl_cffi 发出的不带 cookie 请求有 18 次中的 16 次成功。referrer 包括 Google、Bing、DuckDuckGo、LinkedIn 和 example.com。在 2 个失败请求中,有 1 个收到 HTTP 502。

带有 sec-fetch-site: cross-site 标头但没有 Referer 标头的不带 cookie 请求,3 次全部被封锁。带有同源 referrer 的不带 cookie 请求也连续 3 次被封锁。在我们的测试中,首次请求需要看起来像是从其他网站(例如搜索结果)点击进入。

使用 Google referrer 时,成功率还会随并发量和 User-Agent 而变化:

  • 通过动态 IP 并发发送 6 个请求时,45 个 curl_cffi 请求中有 14 个成功。
  • 每次只发送 1 个请求,并在每个请求后暂停 6 秒时,20 个请求中有 13 个成功,另有 3 个发生重定向。
  • 使用 curl 或 python-requests User-Agent 的请求,29 次全部被封锁。
  • 并发发送 6 个请求时,使用 Chrome User-Agent 的 15 个系统 curl 请求中有 4 个成功。

即使每次只发送 1 个请求,20 个请求中仍有 7 个失败,因此自建抓取工具仍需要重试、新会话和监控。

每个 IP 能获取多少个 ZoomInfo 页面

会话成功获取第一个页面后,ZoomInfo 只会向同一 IP 再提供少量页面。每个会话都通过代理的粘性会话设置保持 1 个住宅 IP,并在首次请求中发送 Google referrer。然后,我们统计在首次请求被封锁前成功获取的公司页面数量。

在成功获取第一个页面的 14 个会话中,每个会话在首次请求被封锁前加载公司页面数量的中位数为 3。到第 5 个请求时,14 个会话中有 12 个已收到 Cloudflare 429。第一次收到 429 后,该会话中的所有后续请求也都会收到 429。我们检查的所有 429 均未携带 Retry-After 标头。

在以下会话中,页面之间延迟 2 秒或 15 秒并未改变可获取的数量:

客户端 页面之间的延迟 各次请求的结果
HTTP 客户端 15 秒 成功、成功,之后 8 次均为 429
HTTP 客户端 15 秒 成功、成功、成功,之后 7 次均为 429
HTTP 客户端 2 秒 成功、成功、成功,之后 7 次均为 429
真实 Chrome 页面加载时间 成功、成功、成功,之后 5 次均显示“Just a moment…”

在 Chrome 会话中,有界面的 Chrome 在隐藏 navigator.webdriver 的情况下运行,并使用单独的粘性会话。前三个页面各自在 6 至 9 秒内加载完成。之后 5 个页面均在大约 15 秒后仍显示 Cloudflare 验证尽管浏览器已有放行 cookie cf_clearance。

在保留首次响应 cookie 的会话中,后续请求不再需要外部 referrer。后续请求可以像浏览器一样,将前一个 ZoomInfo 页面指定为 referrer。

清除 cookie 并不会重置计数。在每次请求前都清除 cookie 并每次发送 Google referrer 的 4 个会话中,有 3 个在抓取 3 个页面后收到 429。因此,该限制与 IP 绑定,而不是与 cookie 绑定。

论坛帖子称反机器人 cookie 与出口 IP 绑定,但我们无法使用 HTTP 客户端收到的 cookie 复现这一情况。我们将包含 _pxhd、_cfuvid 和 contextual_signup_test 的 cookie jar 从一个 IP 移到新 IP,5 个请求中有 4 个成功。与此同时,来自其他新 IP 的不带 cookie 请求在 5 次中有 2 次成功。两组请求都发送了 Google referrer。

此图展示了从抓取工具到公司页面的路径,以及每套反机器人系统返回的响应:

从抓取程序到 ZoomInfo 公司页面的路径流程图。请求经由可选的代理出口节点发送至 2 个反机器人系统。Cloudflare 在每个 IP 访问约 3 个页面后返回 429;当首次请求没有外部引荐来源时,HUMAN Security 返回 403。ZoomInfo 的源站返回 200,公司数据嵌入在 JSON 中。

被封锁的请求不会返回公司数据,因此封锁率越高,获取每条记录所需的请求就越多。

按每个 IP 大约 3 个页面计算,每月采集 10 万个公司页面需要超过 3.3 万个能成功获取首个页面的会话。每个被封锁的首次请求都意味着还要再启动一个会话。Bright Data 的网络解锁器会为你处理这些会话。

从 ZoomInfo 页面提取公司数据

ZoomInfo 的公司页面是服务端渲染的 Angular 应用。服务器将页面数据写入 <script id="ng-state" type="application/json">。该脚本来自 Angular 的 TransferState API,可让浏览器复用数据,无需再次获取。对于 Walmart,该脚本包含 84 KB 的 JSON,其 pageData 对象有 42 个键。

嵌入式 JSON 包含的数字比可见页面更精确。Aviagen 的资料页面将员工人数显示为范围,并将营收四舍五入到小数点后 1 位:

Aviagen 的 ZoomInfo 公司资料。页眉显示 Agriculture、Alabama、United States 和 5K - 10K Employees;概览中显示总部、电话号码、网站、24 亿美元营收、行业标签以及 65 的 ZI Rank。

Aviagen 的可见资料显示员工人数为“5K – 10K”,营收为“$2.4 Billion”。

同一页面的 JSON 包含未取整的数字:

<script id="ng-state" type="application/json">
{"pageData": {
  "companyId": 9047889,
  "name": "Aviagen",
  "numberOfEmployees": "6776",
  "about": {"revenue": "2413885", ...},
  ...
}}
</script>

ZoomInfo 以千美元为单位存储营收,因此 Aviagen 的 2413885 表示 2,413,885,000 美元。请将一家公司的嵌入式 JSON 与可见资料进行比较,以确认换算倍数。

其他公司页面也显示范围,而其嵌入式 JSON 则包含未取整的数字。Walmart 页面上的 FAQ 区块显示员工人数为“10K+”,而 JSON 中为 2,100,000。对于 5 家营收显示为“<$5 Million”的小型公司,JSON 中的估算值介于 366,000 至 4,474,000 美元之间。

即使公司数据保持不变,某些字段也会在不同抓取请求之间发生变化。两次抓取 Walmart 页面相隔数分钟,恰好有 3 个 pageData 键不同:

  • pic 包含一组部分不同的 9 名员工。
  • copilotContent 包含相同的 10 个营销模板,但顺序不同。
  • intents 包含 5 个条目,其中 2 个带有 randomLocation 值,而且两个值均发生变化。

如果页面中包含这些字段,对整个页面进行哈希可能会让未变化的公司在每次爬取时看起来都发生了变化。以下解析器保存为 parse_zoominfo.py它会读取嵌入式 JSON、转换单位、将 ID 保留为字符串,并仅对公司字段进行哈希:

import hashlib
import json
import re
from urllib.parse import urljoin

STATE = re.compile(r'<script[^>]*id="ng-state"[^>]*>(.*?)</script>', re.S)
LD_JSON = re.compile(
    r'<script[^>]*type="application/ld\+json"[^>]*>(.*?)</script>', re.S
)


def _state(html: str) -> dict | None:
    match = STATE.search(html)
    return json.loads(match.group(1)) if match else None


def _int(value) -> int | None:
    return int(value) if value and str(value).isdigit() else None


def _zi_rank(html: str) -> int | None:
    for block in LD_JSON.findall(html):
        data = json.loads(block)
        if isinstance(data, dict) and data.get("@type") == "Organization":
            for prop in data.get("additionalProperty", []):
                if prop.get("name") == "ZI Rank":
                    return prop.get("value")
    return None


def parse_company(html: str) -> dict | None:
    state = _state(html)
    page = (state or {}).get("pageData") or {}
    if not page.get("companyId"):
        return None  # a challenge page, an empty shell, or another template
    about = page.get("about") or {}
    overview = page.get("overview") or {}
    revenue_k = _int(about.get("revenue"))  # stored in thousands of USD
    domain = (page.get("url") or "").lower().removeprefix("www.")
    return {
        "id": str(page["companyId"]),  # strings: person IDs overflow int32
        "name": page.get("name"),
        "domain": domain or None,
        "employees": _int(page.get("numberOfEmployees")),  # unrounded count
        "revenue_usd": revenue_k * 1000 if revenue_k is not None else None,
        "founded": _int(page.get("foundingYear")),
        "ticker": overview.get("ticker"),
        "sic": overview.get("sic") or [],
        "naics": overview.get("naics") or [],
        "industries": [i["name"] for i in about.get("industries") or []],
        "hq": page.get("address"),
        "phone": about.get("phone"),
        "tech_used": sorted(t["name"] for t in page.get("techUsed") or []),
        "management_levels": {
            m["title"]: m["value"] for m in page.get("mgtl") or []
        },
        "zi_rank": _zi_rank(html),
        # Person fields (executives, pic, hierarchyOrgChart) are excluded.
    }


# intents, pic, and copilotContent changed between 2 fetches of the
# same page, so the hash excludes them.
STABLE = ("name", "domain", "employees", "revenue_usd", "industries",
          "hq", "phone", "sic", "naics", "ticker", "tech_used")


def content_hash(record: dict) -> str:
    stable = json.dumps({k: record.get(k) for k in STABLE}, sort_keys=True)
    return hashlib.sha256(stable.encode()).hexdigest()[:16]

对于任何不含公司数据的页面(例如验证页面或空页面)parse_company 都会返回 None。对于两次 Walmart 抓取结果content_hash 返回了相同的值 b4ffe1810bbb1c8c而其原始 HTML 的 SHA-256 不同。

大规模查找 ZoomInfo 公司 URL

ZoomInfo 的 robots.txt 列出了 1 个站点地图,其中仅包含营销页面。我们没有找到公司或个人页面的站点地图,因此你可以通过另外 3 个来源构建 URL 列表。

目录页面

每个 /companies-search/ URL 都会列出符合地点和行业筛选条件的公司pageNum 查询参数用于选择页码。阿拉巴马州农业列表显示共有 1,962 家公司。第 1 至 5 页均返回了我们请求的页面。请求第 6 页和第 10 页时,HTTP 200 响应中返回的却是第 1 页的公司,嵌入式 JSON 中的值为 pageNum: 1。

以下是该列表的第一页:

ZoomInfo 中 Alabama 农业公司的目录结果。表格列出了 9 家公司,包括 Peco Foods 和 Aviagen,并包含行业、地点、营收和员工列;分页显示第 1 至第 5 页。

分页只显示 5 页。营收列以千美元为单位,因此 Aviagen 的 2.4M 与其资料中的 24 亿美元相符。

仅检查状态码的分页爬虫工具会再次将第 1 页的公司采集为新结果。请将以下防护逻辑添加到 parse_zoominfo.py。它会将嵌入式 JSON 中的页码与你请求的页码进行比较:

def directory_page(html: str, requested: int) -> list[str] | None:
    state = _state(html) or {}
    if state.get("pageNum") != requested:
        return None  # past the reachable pages, ZoomInfo serves page 1 again
    rows = (state.get("aiSearchResults") or {}).get("data") or []
    return [urljoin("https://www.zoominfo.com", row["companyUrl"])
            for row in rows]

directory_page 每页返回 9 个公司 URL,并对第 6 页返回 None。因此,该筛选条件下只有 5 页可访问,其中包含约 45 家公司,而总数为 1,962 家。要覆盖更多公司,可按美国州或细分行业拆分大型筛选条件,直到每个较小筛选条件包含的公司不超过 45 家。Bright Data 的 ZoomInfo 爬虫 API 也可以爬取 ZoomInfo 目录页面并将其中列出的公司作为结构化记录返回。

搜索你已经知道的公司

通过 Bright Data 的搜索引擎 API发送 site:zoominfo.com/c example.com 之类的查询,可以查找某个域名对应的候选资料。对于 brightdata.com,该查询返回了 3 个不同的公司 ID。保留网站与所搜索域名匹配的候选资料。

许多资料页面都会列出相似公司,Bright Data 的 ZoomInfo 爬虫 API 会在 similar_companies 字段中返回这些公司。爬虫可以从种子 URL 列表开始,并跟随这些链接发现更多公司。

重复资料和稳定 ID

ZoomInfo 自身的数据中存在重复项,因此上述 3 个来源都可能返回重复资料。对于 brightdata.com,其中一个额外资料列出了主资料的街道地址,但网站域名不同。其员工人数和营收估算也与主资料不同。

按域名关联会漏掉该重复项,按名称关联也可能匹配到无关公司。搜索 minimax.com 返回了 9 个名称中包含 Minimax 的资料,其中数个属于无关企业。应根据域名或总部街道地址进行实体匹配,并仅审核只匹配其中一项的资料。使用 ZoomInfo 的数字 ID 作为主键。

在公司 URL 中,slug 可能发生变化,但我们追踪的重定向保留了数字 ID。在一个样本中,20 个公司 URL 中有 3 个返回 301 重定向,因此每个 URL 都需要通过相同的反机器人系统再发送一次请求。对于输入 /c/walmart-inc/155353090爬虫 API 返回了规范 URL /c/walmart/155353090其中保留了相同 ID。网络解锁器跟随了同一重定向并返回 Walmart 页面。

大规模采集 ZoomInfo 数据的 3 种方式

按每月 10 万个公司页面、每个 IP 大约 3 个页面计算,自建流程需要超过 3.3 万个会话,并且需要持续投入工程工作。Bright Data 的 ZoomInfo 数据提取方案可处理不同程度的工作,从已解锁页面到现成记录:

网络解锁器 ZoomInfo 爬虫 API ZoomInfo 数据集
你发送 页面 URL 公司或目录 URL 筛选条件
你获得 完整页面 HTML 结构化 JSON 记录 文件中的现成记录
在我们的测试中 返回的公司页面完整保留嵌入式 JSON 返回了公司记录,目录 URL 的分页也由系统代为处理 示例文件展示了完整 schema
交付方式 同步或异步请求 通过 API、云存储或 webhook 提供快照 按计划刷新并交付文件
解析 由你负责 已包含 已包含

合适的方案取决于你需要的输出:

  • 为你自己的 URL 列表获取结构化记录。使用爬虫 API。
  • 获取未取整的员工人数和营收。使用带 parse_company 的网络解锁器。
  • 获取已完成采集的广泛覆盖数据。使用数据集。

对于需要点击和滚动的页面,Bright Data 的浏览器 API会运行一个托管 Chrome,你可以通过 Playwright 或 Puppeteer 控制它。公司页面不需要点击,因此网络解锁器是更简单的选择,因为它会直接返回 HTML,无需控制浏览器。对于 AI 智能体,Bright Data 的 ZoomInfo MCP 服务器可让 Claude Desktop 和 Cursor 等客户端采集相同的公开公司资料。

首先,创建 Bright Data 账户和 API 密钥。对于网络解锁器,还需要在控制面板中创建区域即具名的网络解锁器配置。新账户会获得免费额度用于测试 Bright Data 的 API。

这些脚本需要 Python 3.10 或更高版本。完整文件位于 GitHub 上 Bright Data 的 zoominfo-爬虫工具代码库中。只需设置一次密钥和区域:

pip install httpx
export BRIGHTDATA_API_KEY="your_api_key"
export BRIGHTDATA_UNLOCKER_ZONE="your_web_unlocker_zone"

每个脚本都会在启动时读取密钥,因此缺少密钥的脚本会在发送任何请求前停止。

网络解锁器返回由你解析的页面

网络解锁器是 Bright Data 的解锁 API。你发送 URL,网络解锁器会先处理代理、浏览器渲染和验证,再返回页面。在我们的测试中,它成功解锁了 ZoomInfo 公司页面并返回完整保留嵌入式 JSON 的 HTML。因此parse_company 无需修改即可处理该 HTML。

网络解锁器通过 x-brd-status-code 响应标头报告目标网站自身的状态,并在 x-brd-error 中提供详细信息因此流程可以根据结果进行分支。默认情况下,它仅对成功请求计费文档列出了会改变此默认行为的设置。

以下抓取程序保存为 fetch_pages.py每次处理 5 个 URL,每个 URL 最多尝试 3 次,并将记录保存到 companies.jsonl:

import json
import os
import sys
import time
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timezone

import httpx

from parse_zoominfo import content_hash, parse_company

API = "https://api.brightdata.com/request"
KEY = os.environ["BRIGHTDATA_API_KEY"]
ZONE = os.environ["BRIGHTDATA_UNLOCKER_ZONE"]


def fetch(url: str, attempts: int = 3) -> dict:
    error = None
    for attempt in range(1, attempts + 1):
        try:
            resp = httpx.post(
                API,
                headers={"Authorization": f"Bearer {KEY}"},
                json={"zone": ZONE, "url": url, "format": "raw"},
                timeout=180,
            )
            # The API reports the target's own result in these 2 headers.
            target = resp.headers.get("x-brd-status-code")
            record = parse_company(resp.text) if target == "200" else None
        except (httpx.HTTPError, ValueError) as exc:  # timeout or bad JSON
            error = f"{type(exc).__name__}: {exc}"
        else:
            if record:
                record["fetched_at"] = datetime.now(timezone.utc).isoformat()
                record["hash"] = content_hash(record)
                return {"url": url, "ok": True, "attempts": attempt,
                        "record": record}
            error = resp.headers.get("x-brd-error") or f"target {target}"
        if attempt < attempts:
            time.sleep(2 ** attempt)
    return {"url": url, "ok": False, "attempts": attempts, "error": error}


if __name__ == "__main__":
    urls = [line.strip() for line in open(sys.argv[1]) if line.strip()]
    with ThreadPoolExecutor(max_workers=5) as pool, \
            open("companies.jsonl", "a") as good, \
            open("failures.jsonl", "a") as bad:
        for result in pool.map(fetch, urls):
            (good if result["ok"] else bad).write(json.dumps(result) + "\n")
            print("saved" if result["ok"] else "not saved", result["url"])

当 urls.txt 中逐行列出 3 个公司 URL 时python fetch_pages.py urls.txt 输出:

saved https://www.zoominfo.com/c/bright-data/430599389
saved https://www.zoominfo.com/c/peco-foods-inc/82428957
saved https://www.zoominfo.com/c/aviagen-group/9047889

全部 3 家公司均已保存。以下是 companies.jsonl 中 Aviagen 条目的公司字段:

{
  "id": "9047889",
  "name": "Aviagen",
  "domain": "aviagen.com",
  "employees": 6776,
  "revenue_usd": 2413885000,
  "founded": 1923,
  "ticker": null,
  "sic": ["02", "025"],
  "naics": ["11", "112"],
  "industries": ["Agriculture", "Animals & Livestock"],
  "hq": {
    "street": "920 Explorer Blvd NW",
    "city": "Huntsville",
    "state": "Alabama",
    "country": "United States",
    "zip": "35806"
  },
  "phone": "(256) 890-3800",
  "tech_used": ["ClickDimensions", "Freshservice", "PHP", "Zingfit"],
  "management_levels": {
    "C-Level": 14,
    "VP-Level": 28,
    "Director": 108,
    "Manager": 998,
    "Non-Manager": 1229,
    "Top Contacts": 1316
  },
  "zi_rank": 65
}

Aviagen 资料页面在标题和概览中将这些 employees 和 revenue_usd 值显示为“5K – 10K”和“$2.4 Billion”。

ZoomInfo 爬虫 API 返回记录

Bright Data 的 ZoomInfo 爬虫 API 同时处理解锁和解析。其 Collect by URL 端点接收公司 URL,而 Discover by search_url 接收目录 URL,并采集这些页面中列出的公司。控制面板列出了两个端点,并提供同步和异步模式:

用于 Zoominfo companies information 抓取工具的 Bright Data 控制面板。Available Endpoints 下已选择 Discover by search_url,并输入了 4 个 ZoomInfo 目录 URL,其下方显示同步和异步抓取模式。

Discover by search_url 接收此类目录 URL,并返回其页面中列出的公司。

对于批量任务,使用异步 /trigger 端点。它会在后台运行任务并返回快照 ID。同步 /scrape 端点会直接返回小型测试的记录;对于耗时较长的任务,则会在达到响应时间限制后将其转为快照。

在包含 40 个 URL 的批次中,首批记录在一分钟内返回。对于大型批次,可让 API 将结果发送到云存储或 webhook。它会在最后一个输入处理完毕时发送每个快照,因此你无需轮询 API。

Discover by search_url 会自动处理分页。仅通过阿拉巴马州目录 URL,它就返回了 45 条公司记录。每条记录都包含其来源目录 URL。对于更大的筛选条件,可为每个美国州或细分行业发送一个目录 URL。

scraper_api.py 客户端会触发任务并轮询,直到快照准备就绪,然后下载快照。其 clean 函数会将员工人数范围转换为 employees_min 和 employees_max并标记“<$5 Million”等营收范围。它还会删除包含人员数据的领导层、组织结构图和 CEO 字段。使用 --discover 时,客户端会将目录 URL 发送到 Discover by search_url:

import json
import os
import re
import sys
import time

import httpx

BASE = "https://api.brightdata.com/datasets/v3"
DATASET_ID = "gd_m0ci4a4ivx3j5l6nx"  # Zoominfo companies information
HEADERS = {"Authorization": f"Bearer {os.environ['BRIGHTDATA_API_KEY']}"}


def trigger(urls: list[str], discover: bool = False, limit: int = 50) -> str:
    params = {"dataset_id": DATASET_ID, "include_errors": "true"}
    if discover:  # directory URLs: the job handles their pagination
        params |= {"type": "discover_new", "discover_by": "search_url",
                   "limit_per_input": limit}
    resp = httpx.post(f"{BASE}/trigger", params=params, headers=HEADERS,
                      json=[{"url": u} for u in urls], timeout=60)
    resp.raise_for_status()
    return resp.json()["snapshot_id"]


def wait(snapshot_id: str, every: int = 30) -> dict:
    while True:
        try:
            progress = httpx.get(f"{BASE}/progress/{snapshot_id}",
                                 headers=HEADERS, timeout=60).json()
        except (httpx.HTTPError, ValueError):  # retry after a network error
            progress = {}
        if progress.get("status") in ("ready", "failed"):
            return progress
        time.sleep(every)


def download(snapshot_id: str, tries: int = 30) -> list[dict]:
    for _ in range(tries):
        try:
            resp = httpx.get(f"{BASE}/snapshot/{snapshot_id}",
                             params={"format": "json"}, headers=HEADERS,
                             timeout=300)
            if resp.status_code == 200 and resp.content.strip():
                return resp.json()
        except (httpx.HTTPError, ValueError):  # timeout or a partial file
            pass
        time.sleep(10)  # wait until the snapshot file is available
    raise RuntimeError(f"snapshot {snapshot_id} is not available yet")


def employee_range(text: str | None) -> tuple[int | None, int | None]:
    # "11-50" -> (11, 50), "1K - 5K" -> (1000, 5000), "10K+" -> (10000, None)
    if not text:
        return None, None
    nums = [int(float(n) * (1000 if k else 1)) for n, k
            in re.findall(r"([\d.]+)\s*(K?)", text.replace(",", ""))]
    if not nums:
        return None, None
    if len(nums) == 1:
        return nums[0], None if "+" in text else nums[0]
    return nums[0], nums[1]


def clean(row: dict) -> dict:
    row["employees_min"], row["employees_max"] = employee_range(
        row.get("employees_text"))
    # Headcount comes from the employees_text range.
    row["employees"] = row["total_employees"] = None
    row["revenue_is_range"] = (row.get("revenue_text") or "").startswith("<")
    row["stock_symbol"] = (row.get("stock_symbol") or "").strip() or None
    row["id"] = str(row["id"])
    for field in ("leadership", "org_chart", "ceo"):  # data about people
        row.pop(field, None)
    return row


if __name__ == "__main__":
    urls = [a for a in sys.argv[1:] if a.startswith("http")]
    snapshot = trigger(urls, discover="--discover" in sys.argv)
    progress = wait(snapshot)
    summary = {k: progress.get(k) for k in ("status", "records", "errors")}
    print(json.dumps(summary))
    rows = download(snapshot)
    # Keep rows that carry an id and no error.
    good = [clean(r) for r in rows if r.get("id") and not r.get("error")]
    bad = [r for r in rows if r.get("error") or not r.get("id")]
    json.dump(good, open("records.json", "w"), indent=2)
    json.dump(bad, open("errors.json", "w"), indent=2)
    print(f"{len(good)} records saved to records.json")

以 urls.txt 中的 3 个公司 URL 作为参数时,客户端输出:

{"status": "ready", "records": 3, "errors": 0}
3 records saved to records.json

快照包含全部 3 条记录,客户端将其保存到 records.json。每条记录都包含营收、行业、技术、网站、总部和电话号码等公司字段。

ZoomInfo 数据集交付已采集的记录

ZoomInfo 数据集提供包含相同字段、已完成采集并按计划刷新的记录。获取前,请查看数据集页面了解当前记录数量、刷新频率和定价,并检查其样本。要在该数据集与 ZoomInfo 订阅之间作出选择,请阅读我们对 Bright Data 与 ZoomInfo 的比较。

获取前先筛选。数据集的筛选器可让你只选择所需记录,例如 URL 中包含 /c/ 的公司页面。除非你有合法依据处理人员数据,否则请删除包含此类数据的领导层和组织结构图列。

数据集保存的是采集时的记录,每次刷新都会更新这些记录。要选择刷新频率,可对同一组公司样本采集两次,两次间隔一个月,然后统计发生变化的记录。

将 CRM 域名与 ZoomInfo 资料匹配

CRM 保存的是域名,而不是 ZoomInfo URL。以下脚本保存为 find_profiles.py会通过 Bright Data 在 Google 上搜索每个域名。它使用 fetch 从 fetch_pages.py 获取排名前三的资料,并保留 domain 与 CRM 域名匹配的第一条记录。如果不带引号的搜索没有找到匹配项,脚本会将域名加上引号再次搜索:

import json
import os
import re
import sys
from concurrent.futures import ThreadPoolExecutor
from urllib.parse import quote_plus

import httpx

from fetch_pages import fetch

API = "https://api.brightdata.com/request"
KEY = os.environ["BRIGHTDATA_API_KEY"]
# A SERP API zone if you have one, otherwise the Web Unlocker zone.
ZONE = (os.environ.get("BRIGHTDATA_SERP_ZONE")
        or os.environ["BRIGHTDATA_UNLOCKER_ZONE"])
PROFILE = re.compile(r"https://www\.zoominfo\.com/c/[^/?#]+/\d+")
# Each query found profiles that the other missed, so try both.
QUERIES = ("site:zoominfo.com/c {}", 'site:zoominfo.com/c "{}"')


def candidates(query: str, top: int = 3) -> list[str]:
    url = f"https://www.google.com/search?q={quote_plus(query)}&brd_json=1"
    for _ in range(3):
        try:
            resp = httpx.post(
                API,
                headers={"Authorization": f"Bearer {KEY}"},
                json={"zone": ZONE, "url": url, "format": "raw"},
                timeout=120,
            )
            results = resp.json().get("organic", [])  # parsed results
        except (httpx.HTTPError, ValueError):  # timeout or an empty body
            continue
        links = [row.get("link", "") for row in results]
        found = (PROFILE.match(link) for link in links)
        return list(dict.fromkeys(m.group(0) for m in found if m))[:top]
    return []


def find_profile(domain: str) -> dict | None:
    for query in QUERIES:
        for url in candidates(query.format(domain)):
            result = fetch(url)
            # Keep a profile only when its website is the CRM domain.
            if result["ok"] and result["record"]["domain"] == domain:
                return result["record"]
    return None


if __name__ == "__main__":
    domains = [d.strip().lower().removeprefix("www.") for d in sys.argv[1:]]
    with ThreadPoolExecutor(max_workers=5) as pool, \
            open("matches.jsonl", "a") as out:
        records = pool.map(find_profile, domains)
        for domain, record in zip(domains, records, strict=True):
            out.write(json.dumps({"domain": domain, "record": record}) + "\n")
            print(domain, record["id"] if record else "no match")

对于 4 个 CRM 域名python find_profiles.py aviagen.com pecofoods.com brightdata.com walmart.com 输出:

aviagen.com 9047889
pecofoods.com 82428957
brightdata.com 430599389
walmart.com no match

对于每个匹配项,脚本都返回完整的公司记录。不带引号的搜索找到了 Aviagen 和 Bright Data,带引号的搜索找到了 Peco Foods。Walmart 的资料未出现在这两次搜索结果中,因此脚本返回无匹配项,而不是错误的公司。将未匹配的域名发送到名称搜索或审核队列。

在每条记录进入 CRM 前进行验证

我们看到的 ZoomInfo 及 Bright Data API 的每种响应都需要采取相应措施:

响应 含义 操作
200,且 ng-state 中包含公司数据 真实公司页面 解析并存储
301 重定向 URL 已更改 存储新 URL,并使用 ID 作为键
403,且响应正文中包含 px-captcha HUMAN 封锁了请求 在新会话中重试
429,且包含 cf-mitigated: challenge 已达到 Cloudflare 的单 IP 请求限制 轮换 IP,并停止在该 IP 上重试
429,但没有 Cloudflare 标头 Bright Data 因成功率过低而触发自动限流 降低请求速率,并修复失败的请求
502,且没有反机器人特征 临时上游错误 使用退避策略重试
网络解锁器 x-brd-status-code 标头 目标网站自身的状态 存储 200 结果,并对其他结果进行有限次数的重试
/scrape 返回爬虫 API 202 任务已转为快照 轮询快照 ID
目录页面的 pageNum 小于请求值 已超过最后一个可访问页面 停止分页并缩小筛选范围

状态检查可发现失败页面,字段检查可发现错误记录:

  • 将 ID 保留为字符串。在我们检查的数据中,ZoomInfo 人员 ID 的范围为 -2,135,143,530 至 17,773,369,573,既无法放入有符号 32 位整数,也无法放入无符号 32 位整数。
  • 从范围或嵌入式 JSON 读取员工人数。使用 clean 中的 employees_min 和 employees_max或者使用 JSON 中的 numberOfEmployees。
  • 标记营收范围。“<$5 Million”之类的值属于范围,而 JSON 中包含估算值。
  • 拆分行业代码。“54,541”之类的值包含行业门类代码和行业大类代码。
  • 预期 CEO 字段为空。在我们检查的每个公开页面中,ZoomInfo 都将 CEO 字段留空。
  • 跟踪每个批次的字段填充率。某个字段的填充率突然下降,可能意味着 ZoomInfo 更改了模板。

我们的修复不准确网页爬虫数据指南更深入地介绍了数据清理。

大规模爬虫 ZoomInfo 的成本

不同方案采用不同的计费单位,每个定价页面都会列出当前费率:

方案 计费依据 其他成本
网络解锁器和你的解析器 采用默认设置时按成功请求计费 URL 发现、解析、去重和托管
ZoomInfo 爬虫 API 按交付的记录计费 去重和存储
ZoomInfo 数据集 按文件中的记录计费 去重和存储
住宅代理和你自己的技术栈 按带宽计费 会话、重试、解析、去重及其运行所需的工程工作

我们测量的公司页面压缩后大小为 42 至 55 KB,因此 10 万个页面会使用约 4 至 6 GB 带宽。在更高数据量下,请查看各定价页面中的批量方案。

住宅代理为你提供最大的控制权。对于这种大小的页面,与工程成本相比,其带宽费用很低。使用自建技术栈时,你还需要为被封锁的首次请求、重试,以及每个 IP 约 3 个页面所需的额外会话付出成本。你还需要一名工程师,在 ZoomInfo 更改反机器人规则或页面结构时维护该技术栈。

在 IsraelTech 播客采访中,Bright Data CEO Or Lenchner 表示:“当你需要始终反复进行大规模采集时,就需要我们。”对于少量数据,Lenchner 建议团队自行采集。我们对托管服务与自建爬虫的比较列出了每种方式的成本。

总结

ZoomInfo 页面将未取整的企业统计数据作为嵌入式 JSON 保存,但在我们的测试中,反机器人系统会检查每个首次请求,并在大约抓取 3 个页面后封锁 IP。JSON 让数据提取成为较小的工作,因此在大规模采集时,会话、URL 发现和去重最为耗费精力。我们的脚本仅采集公开的公司数据,因为联系人详情需要账户才能查看,而处理人员数据可能需要合法依据。

Bright Data 的网络解锁器和 ZoomInfo 爬虫 API 会处理会话,而 ZoomInfo 数据集则交付 Bright Data 已采集的记录。使用你自己的账户样本测试 ZoomInfo 爬虫 API,并将记录与你的 CRM 进行比较。

常见问题

可以爬虫 ZoomInfo 吗?

可以,但仅限无需登录即可加载的公司页面。联系人详情需要账户才能查看。公开公司页面通过嵌入式 JSON 携带数据。在我们的测试中,ZoomInfo 封锁了全部 32 个没有其他网站 referrer 的首次请求,因此每个新会话都应带上一个外部 referrer。

如何从 ZoomInfo 爬虫数据?

找到每家公司的资料 URL,获取公开页面,然后从其嵌入式 JSON 中读取数据。测试中的 IP 获取页面数量中位数为 3,因此需要规划大量会话。Bright Data 的 ZoomInfo 爬虫 API 会为你处理会话和解析。

ZoomInfo 有 API 吗?

有。ZoomInfo 通过 API 销售其自有数据库的访问权限。对于公开公司资料,Bright Data 的 ZoomInfo 爬虫 API 会采集这些页面并返回结构化记录。Bright Data 的 ZoomInfo 数据集提供包含相同字段、已完成采集并按计划刷新的记录。

ZoomInfo 的数据是否过时?

其中部分记录可能过时,因此在其进入 CRM 前应进行检查。我们搜索一个域名时返回了 3 个 ZoomInfo ID,其中一个是员工人数不同的重复资料。请进行去重、比较填充率,并根据字段变化频率执行刷新。

可以免费使用 ZoomInfo 吗?

部分可以。zoominfo.com 上的公司资料页面无需账户即可加载,但其 HTML 将联系人详情和部分内容标记为付费内容。ZoomInfo 还提供每月额度有限的免费 Lite 方案。新的 Bright Data 账户会获得免费额度,用于在这些页面上测试 ZoomInfo 爬虫 API。

有免费的 ZoomInfo 爬虫工具吗?

有。Bright Data 在 GitHub 上的开源 zoominfo-爬虫代码库提供了使用 Bright Data API 的 Python 脚本。在我们的测试中,即使带有 Google referrer,使用 curl 或 python-requests User-Agent 的请求也连续 29 次被封锁。Bright Data 的 API 会为这些脚本处理解锁。

支持支付宝等多种支付方式

Satyam Tripathi

技术写作者

5 years experience

Satyam Tripathi 帮助 SaaS 和数据初创公司将复杂技术转化为可执行的内容,提升开发者采用度并增强用户理解。

Expertise
Python 开发者教育 技术写作