如何使用 Python 与 Waterfall Email Finder API 批量查找并验证联系人工作邮箱

1python -m pip install requests

1export DATABLIST_API_KEY="your-api-key"

1import json
2import os
3import time
4import uuid
5
6import requests
7
8
9def find_emails(people, api_key, settings):
10    """Submit one email-finder task, wait, and return every result."""
11    tasks_url = "https://api.datablist.com/v1/enrichments/tasks"
12    headers = {"X-API-Key": api_key}
13
14    unique_id = str(uuid.uuid4())
15    print(f"Task uniqueId: {unique_id}", flush=True)
16
17    payload = {
18        "enrichmentId": "waterfall-email-finder",
19        "uniqueId": unique_id,
20        "settings": settings,
21        "inputs": people,
22    }
23    response = requests.post(tasks_url, headers=headers, json=payload)
24    response.raise_for_status()
25    task = response.json()
26    task_id = task["@id"]
27    task_url = f"{tasks_url}/{task_id}"
28
29    while task["status"] not in ("completed", "failed", "stopped"):
30        time.sleep(3)
31        response = requests.get(task_url, headers=headers)
32        response.raise_for_status()
33        task = response.json()
34
35    if task["status"] != "completed":
36        raise RuntimeError(
37            f"Task {task_id} {task['status']}: {task.get('error')}"
38        )
39
40    results = []
41    results_url = f"{task_url}/results"
42    params = {"limit": 100}
43
44    while True:
45        response = requests.get(results_url, headers=headers, params=params)
46        response.raise_for_status()
47        page = response.json()
48        results.extend(page["results"])
49
50        next_cursor = page["nextCursor"]
51        if next_cursor is None:
52            break
53        params["cursor"] = next_cursor
54
55    return results
56
57
58if __name__ == "__main__":
59    api_key = os.environ.get("DATABLIST_API_KEY")
60    if not api_key:
61        raise RuntimeError("Set DATABLIST_API_KEY before running this script.")
62
63    settings = {"sourceKind": "name_company"}
64    people = [
65        {
66            "inputId": "contact-satya-nadella",
67            "data": {
68                "firstName": "Satya",
69                "lastName": "Nadella",
70                "companyNameOrDomain": "microsoft.com",
71            },
72        },
73        {
74            "inputId": "contact-reid-hoffman",
75            "data": {
76                "firstName": "Reid",
77                "lastName": "Hoffman",
78                "companyNameOrDomain": "greylock.com",
79            },
80        },
81    ]
82
83    lookup_results = find_emails(people, api_key, settings)
84    for result in lookup_results:
85        print(json.dumps(result, indent=2))

1python find_work_emails.py

1settings = {"sourceKind": "name_company"}
2people = [
3    {
4        "inputId": "contact-satya-nadella",
5        "data": {
6            "firstName": "Satya",
7            "lastName": "Nadella",
8            "companyNameOrDomain": "microsoft.com",
9        },
10    }
11]
12lookup_results = find_emails(people, api_key, settings)

1settings = {"sourceKind": "name_company", "useFullName": True}
2people = [
3    {
4        "inputId": "contact-satya-nadella",
5        "data": {
6            "fullName": "Satya Nadella",
7            "companyNameOrDomain": "Microsoft",
8        },
9    }
10]
11lookup_results = find_emails(people, api_key, settings)

1settings = {"sourceKind": "linkedin"}
2people = [
3    {
4        "inputId": "contact-satya-nadella",
5        "data": {
6            "linkedInProfile": "https://www.linkedin.com/in/satyanadella/"
7        },
8    }
9]
10lookup_results = find_emails(people, api_key, settings)

{
  "taskId": "2286103ce43748828e27ce71b9cc526b",
  "results": [
    {
      "inputId": "contact-satya-nadella",
      "status": "success",
      "outputs": {
        "gender": null,
        "mxProvider": "microsoft",
        "emailFound": "s*******@microsoft.com",
        "emailStatus": "verified"
      },
      "creditsUsed": "25",
      "error": null
    },
    {
      "inputId": "contact-reid-hoffman",
      "status": "success",
      "outputs": {
        "gender": null,
        "mxProvider": "mimecast",
        "emailFound": "r*******@greylock.com",
        "emailStatus": "verified"
      },
      "creditsUsed": "25",
      "error": null
    }
  ],
  "nextCursor": null
}

1results_by_id = {result["inputId"]: result for result in lookup_results}
2
3campaign_contacts = []
4needs_verification = []
5
6for person in people:
7    result = results_by_id.get(person["inputId"])
8    if result is None:
9        continue
10    if result["status"] != "success":
11        continue
12    if not result["outputs"].get("emailFound"):
13        continue
14
15    contact = person.copy()
16    contact["result"] = result
17
18    if result["outputs"].get("emailStatus") == "verified":
19        campaign_contacts.append(contact)
20    else:
21        needs_verification.append(contact)
22
23print(json.dumps({
24    "campaign_contacts": campaign_contacts,
25    "needs_verification": needs_verification,
26}, indent=2))

1curl "https://api.datablist.com/v1/enrichments/tasks" \
2  -H "X-API-Key: $DATABLIST_API_KEY"