在 Python 中，将公开的 linkedin.com/in/... 个人主页 URL 发送到 Datablist 的 Waterfall Personal Email Finder API，即可从 LinkedIn 查找私人邮箱。API 会按顺序调用当前可用的邮箱数据提供商，并在其中一家找到结果后返回私人邮箱地址。
下面的完整 Requests 示例会提交一个后台任务、等待任务完成，并获取所有结果分页。成功完成查询后，你可以复用返回的对象来整理招聘联系人记录，将每位候选人的姓名、LinkedIn URL、目标职位与找到的邮箱保存在一起。
此工作流查找的是独立于候选人当前雇主的私人邮箱。如果你需要与其所在公司关联的邮箱地址，请参阅另一篇指南：通过 LinkedIn 个人主页查找工作邮箱。
🔑 快速了解要点
提供候选人姓名、公开的 LinkedIn 个人主页 URL 和目标职位。这个 Python 工作流会查找可用的私人邮箱，并将邮箱与原始招聘信息对应保存。
- **输入：**候选人姓名、公开的
linkedin.com/in/...个人主页 URL 和目标职位。- **输出：**找到的私人邮箱，以及与之匹配的候选人信息。
💡 **更想使用 no-code 方案？**使用 Datablist 的 Waterfall Personal Email Finder，无需编写代码即可查找私人邮箱。
如果源数据只有姓名、没有个人主页 URL，请先使用 Python 查找公开的 LinkedIn 个人主页，再将找到的 URL 传给此 API。
配置 Python、Requests 和 Datablist API Key
本例使用 Python 和 requests 包。新建一个名为 find_personal_emails.py 的文件，然后安装 Requests：
python -m pip install requests在 Datablist 账户中创建一个 Personal API Key，并将其导出为环境变量：
export DATABLIST_API_KEY='your-api-key'Growth 套餐支持访问 REST API。每次成功找到私人邮箱也会消耗 credits，具体费用将在下文的成本部分计算。
用一个 Python 函数查找私人邮箱
脚本中可编辑的 candidates 列表包含一条招聘记录。target_role 是示例源字段，仅供招聘工作流在本地使用。API payload 中只会包含 candidate_id 和 linkedin_url。
import json
import os
import time
import uuid
import requests
def find_personal_emails(candidates, api_key):
"""提交一个任务，等待任务完成并返回全部结果。"""
tasks_url = "https://api.datablist.com/v1/enrichments/tasks"
headers = {"X-API-Key": api_key}
inputs = [
{
"inputId": candidate["candidate_id"],
"data": {"profileUrl": candidate["linkedin_url"]},
}
for candidate in candidates
]
unique_id = str(uuid.uuid4())
print(f"Recovery uniqueId: {unique_id}", flush=True)
payload = {
"enrichmentId": "waterfall-personal-email-finder",
"uniqueId": unique_id,
"settings": {},
"inputs": inputs,
}
response = requests.post(tasks_url, headers=headers, json=payload)
response.raise_for_status()
task = response.json()
task_id = task["@id"]
task_url = f"{tasks_url}/{task_id}"
while task["status"] not in ("completed", "failed", "stopped"):
time.sleep(3)
response = requests.get(task_url, headers=headers)
response.raise_for_status()
task = response.json()
if task["status"] != "completed":
raise RuntimeError(f"Task {task_id} {task['status']}: {task.get('error')}")
results = []
results_url = f"{task_url}/results"
params = {"limit": 100}
while True:
response = requests.get(results_url, headers=headers, params=params)
response.raise_for_status()
page = response.json()
results.extend(page["results"])
next_cursor = page["nextCursor"]
if next_cursor is None:
break
params["cursor"] = next_cursor
return results
if __name__ == "__main__":
candidates = [
{
"candidate_id": "candidate-florian-poullin",
"name": "Florian Poullin",
"linkedin_url": "https://www.linkedin.com/in/florianpoullin/",
"target_role": "Backend Engineer",
}
]
api_key = os.environ.get("DATABLIST_API_KEY")
if not api_key:
raise RuntimeError("Set DATABLIST_API_KEY before running this script.")
personal_email_results = find_personal_emails(candidates, api_key)
for result in personal_email_results:
print(json.dumps(result, indent=2))在同一个终端中运行该文件：
python find_personal_emails.py你也可以下载持续维护的同步 Requests 示例。上面的脚本采用相同的提交、轮询和分页流程，同时使用源记录中的候选人 ID。
提交查询任务
该函数会将每位候选人转换为一条 API 输入。inputId 接收稳定不变的候选人 ID，data.profileUrl 接收公开的 LinkedIn 个人主页 URL。此 enrichment 要求使用标准的 linkedin.com/in/... URL。Sales Navigator URL、公司主页、空值以及没有个人主页 URL 的姓名都不是有效输入。
请求使用 waterfall-personal-email-finder 作为 enrichmentId，并传入一个空的 settings 对象，具体格式可参考 Waterfall Personal Email Finder API 文档。脚本会在提交前输出生成的 uniqueId。请保存这个值：如果 API 接受请求后连接中断，你可以用它找到已经创建的任务。
以下是单个个人主页示例的完整请求体：
{
"enrichmentId": "waterfall-personal-email-finder",
"uniqueId": "55001d29-4fb9-4fe2-9497-80f4b346eed7",
"settings": {},
"inputs": [
{
"inputId": "candidate-florian-poullin",
"data": {
"profileUrl": "https://www.linkedin.com/in/florianpoullin/"
}
}
]
}提交成功后，API 会返回 HTTP 202 Accepted，其中 @id 字段包含任务 ID。收到响应后，Datablist 服务器仍会继续处理任务。
等待后台任务完成
服务器运行后台任务期间，Requests 脚本会在本地阻塞等待。它每三秒检查一次任务；当状态为 queued 或 running 时继续轮询。任务状态变为 completed 后，脚本会开始获取结果。如果状态为 failed 或 stopped，脚本会抛出错误，并附上任务 ID 和可用的错误信息。
获取所有结果分页
结果 endpoint 会返回每一行的源 ID、状态、outputs、credits 和错误信息。第一次请求最多获取 100 条结果。脚本会持续获取下一页，直到 nextCursor 为 null。
任务结果会保留输入顺序。即使结果分布在多个分页中，inputId 仍能将每条结果明确关联到对应的候选人记录。
将此示例扩展到更多记录时，请确保每个 candidate_id 与源系统中的值保持一致。有了这个稳定 ID，即使任务包含多个结果分页，也能可靠地完成数据关联。
私人邮箱瀑布查询如何找到结果
Datablist 会先清理提交的公开 LinkedIn 个人主页 URL，再将尚未找到结果的个人主页按顺序交给当前可用的私人邮箱数据提供商。如果某一家没有返回可用邮箱，该个人主页可以继续进入下一家提供商。当某家找到可用邮箱时，这个个人主页的查询序列就会停止，并通过 outputs.emailFound 返回结果。
可用的数据提供商由 Datablist 统一管理，因此你的集成无需针对不同提供商编写请求或 fallback 代码。公开结果只会报告邮箱查询结果，不会披露固定的提供商顺序或每次尝试的历史记录。
Waterfall 是查找邮箱的方法，而轮询承担的是另一项工作：让 Python 客户端等待后台查询完成。
为什么瀑布查询适合跨国招聘团队
不同的私人邮箱数据提供商使用不同的数据集，因此它们在各个国家和市场的覆盖能力也不一样。如果第一家提供商没有候选人的邮箱，瀑布序列中的另一家仍有可能找到。
对于跨多个国家开展招聘的团队，只需集成一个 API，就能获得更广泛的数据提供商覆盖。整个查询顺序由 Datablist 管理，因此团队不必接入和维护多个提供商 API，也能找回单一数据源可能遗漏的联系人信息。
查看完整的 API 返回结果
📘 **已隐藏邮箱：**下方示例使用
**替换了邮箱地址的本地部分。你的 API 结果会包含完整值。
以下是该候选人输入对应的完整结果分页：
{
"taskId": "b5de73ffaa3f4f349c125d7c85125bb4",
"results": [
{
"inputId": "candidate-florian-poullin",
"status": "success",
"outputs": {
"emailFound": "**@yahoo.fr"
},
"creditsUsed": "75",
"error": null
}
],
"nextCursor": null
}inputId 是与源候选人记录关联的 join key。outputs.emailFound 包含私人邮箱地址。creditsUsed 是这一行实际计费的权威数据，error: null 表示 API 没有返回行级错误。cursor 为 null 则说明此响应之后没有下一页结果。
请将该邮箱视为与所提交个人主页相关联的一种联系方式，并在用于招聘前进行人工审核。
整理候选人联系记录供招聘人员审核
查询返回后，无需再发送 API 请求，即可在本地整理一份列表供招聘人员审核。下面的代码会复用已有的 candidates 和 personal_email_results 变量：
results_by_id = {
result["inputId"]: result for result in personal_email_results
}
prepared_contacts = []
for candidate in candidates:
result = results_by_id.get(candidate["candidate_id"])
if not result or result.get("status") != "success":
continue
email_found = result.get("outputs", {}).get("emailFound")
if not email_found:
continue
prepared_contact = candidate.copy()
prepared_contact["result"] = result
prepared_contacts.append(prepared_contact)
print(json.dumps(prepared_contacts, indent=2))对于 candidate-florian-poullin，字典查询会找到具有相同 inputId 的结果。复制后的记录会保留 Florian Poullin 的姓名、LinkedIn URL，以及示例目标职位 Backend Engineer。随后，完整的 API 对象会被添加到 result 字段下，招聘人员可以从 result.outputs.emailFound 读取邮箱。
循环从源候选人列表开始处理。未来即使出现无结果、行级错误、结果缺失或 emailFound 为空的情况，源列表也不会被修改，相应记录也不会进入 prepared_contacts。
如果你需要更完整的审核流程或其他联系渠道，请阅读招聘人员如何查找候选人的邮箱地址指南。
估算 Waterfall Personal Email Finder API 成本
每找到一个私人邮箱会消耗 75 credits，没有找到结果则消耗 0 credits。计算实际支出时，应读取每条结果中的 creditsUsed，不要直接用提交的个人主页总数进行乘法计算。
预算公式为：
找到的私人邮箱数 × 75 credits
例如，提交 100 个个人主页，其中 20 个成功找到私人邮箱，任务将消耗 20 × 75 = 1,500 credits。这只是预算计算示例，并不代表预期的查找成功率。若处理更大规模的数据，找到 100 个邮箱会消耗 7,500 credits，找到 1,000 个邮箱则会消耗 75,000 credits。
按照当前基础美元充值价格，$20 / 20,000 credits = 每个 credit $0.001。因此，每找到一个邮箱会消耗价值 $0.075 的 credits，找到 100 个邮箱则消耗价值 $7.50 的 credits。此估算仅计算所用 credits 的价值。REST API 所需的 Growth 订阅费用需另行支付。请查看 Datablist 定价，了解当前套餐和 credits 充值包。
排查常见错误并恢复已有任务
大多数问题会出现在提交前、HTTP 请求阶段或单条结果中。根据错误出现的位置选择下一步操作：
- **缺少
DATABLIST_API_KEY：**脚本会在发出请求前在本地停止。导出该环境变量，然后重新运行文件。 - **HTTP 身份验证失败：**检查
DATABLIST_API_KEY中的 Personal API Key。函数会通过X-API-Keyheader 发送它。修正凭据后再重新提交。 - **不支持的个人主页 URL：**将 Sales Navigator URL、公司主页、空值或格式错误的 URL 替换为
linkedin.com/in/...格式的公开个人主页。 - **credits 不足：**提交更多任务前，请检查账户余额和可用 credits。
- **任务状态为
failed或stopped：**先检查任务错误、已有结果和 credits 使用情况，再判断哪些数据仍需重新处理。 - **某一行没有结果或出现错误：**保留源候选人记录，以便审核或改用其他联系渠道。查询失败并不能说明邮箱为何无法获取。
如果 POST 响应丢失，请先列出已有任务，不要直接新建任务。在响应中搜索脚本输出的 uniqueId，然后复制匹配任务的 @id：
curl "https://api.datablist.com/v1/enrichments/tasks?enrichmentId=waterfall-personal-email-finder&limit=100" \
--header "X-API-Key: $DATABLIST_API_KEY"
TASK_ID='paste-matching-task-id'
curl "https://api.datablist.com/v1/enrichments/tasks/$TASK_ID" \
--header "X-API-Key: $DATABLIST_API_KEY"
curl "https://api.datablist.com/v1/enrichments/tasks/$TASK_ID/results" \
--header "X-API-Key: $DATABLIST_API_KEY"如果第一份任务历史响应中没有匹配项，请使用返回的分页 cursor 查看下一页。继续检查已恢复的任务或获取其现有结果，不要自动创建另一个可能产生费用的任务。任务文档详细介绍了任务生命周期、分页和结果保留规则。
后续步骤
现在，你已经拥有一个可复用的 Python Requests 函数，可用于处理公开的 LinkedIn 个人主页 URL；同时还有一套基于稳定 ID 的合并逻辑，能将找到的私人邮箱结果与每位候选人的源数据关联起来。
接下来，用一小批真实候选人记录替换示例名单，为每条记录保留稳定 ID 和目标职位，并在将联系人用于招聘流程前逐一审核。当你的应用需要非阻塞并发时，可以使用持续维护的异步 HTTPX 示例作为下一步实现方案。