PythonとWaterfall APIでLinkedInの電話番号を検索する方法

PythonからDatablistのWaterfall Phone Finder APIへ、公開されている linkedin.com/in/... 形式のプロフィールURLを送信すると、LinkedInから電話番号を検索できます。 APIは利用可能な電話番号データプロバイダーを順番に照会し、見つかった番号を国番号付きの国際形式で返します。

本ガイドでは、Requestsライブラリを使って一連のPython連携を実装します。さらに、同じ検索結果を再利用し、各見込み客の氏名、LinkedIn URL、担当者を保持した営業電話リストをローカルで作成します。

📌 お急ぎの方向けまとめ

1つのAPIで公開LinkedInプロフィールURLのリストから電話番号を検索し、見込み客の氏名と担当者を含む営業電話リストをローカルで作成できます。 Waterfall Phone Finderは利用可能なプロバイダーを順番に試し、見つかった番号をチームが確認できる状態で保持します。

入力： 公開LinkedInプロフィールURLを含む見込み客レコード。

出力： 見つかった国際形式の電話番号、国コード、氏名、担当者を含むローカルの架電リスト。

💡 ノーコードで進めたい場合

DatablistのWaterfall Phone Finderを使えば、コードを書かずに LinkedInプロフィールURLから電話番号を検索できます。

プロフィールURLではなく氏名しかない場合は、先に氏名から電話番号を検索するノーコード手順、またはPythonで氏名からLinkedIn プロフィールを検索する方法をご覧ください。

Python、Requests、Datablist APIキーの準備

このPythonサンプルではrequestsパッケージを使用します。プロジェクトフォルダーを作成し、仮想環境を有効化してRequestsをインストールした後、スクリプトファイルを作成します。

shell

1mkdir waterfall-phone-lookup
2cd waterfall-phone-lookup
3python -m venv .venv
4source .venv/bin/activate
5python -m pip install requests
6touch find_phone_numbers.py
7export DATABLIST_API_KEY="your-api-key"

DatablistアカウントでPersonal API Keyを作成し、 DATABLIST_API_KEYに設定してください。REST APIへのアクセスはGrowthプランの機能です。検索成功時に消費されるクレジットについては、後ほど説明します。

1つのPython関数で電話番号を検索

この例では、編集可能な3件の見込み客レコードを使用します。各レコードには、固定の prospect_id、氏名、公開LinkedIn URL、検索結果を確認する担当者が含まれます。

Florian Poullin：https://www.linkedin.com/in/florianpoullin/
Satya Nadella：https://www.linkedin.com/in/satyanadella/
Reid Hoffman：https://www.linkedin.com/in/reidhoffman/

APIへ送る必要があるのは、固定IDとLinkedIn URLだけです。氏名と担当者はローカルのソースレコードに残します。このように分けることで、検索リクエストを小さく保ちながら、結果取得後にチームが必要とするビジネス情報を維持できます。

以下のスクリプト全体をfind_phone_numbers.pyへ貼り付けてください。

python

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

次のコマンドでファイルを実行します。

shell

1python find_phone_numbers.py

Datablistが管理する同期型Requestsのサンプルもダウンロードできます。上記のスクリプトも同じシンプルな流れで、タスクを1件送信し、完了を待ってから、すべての結果ページを取得します。

電話番号検索を送信する

リスト内包表記により、ソースの各見込み客をAPIの入力形式へ変換します。 prospect_idはinputIdに、linkedin_urlはdata.profileUrlになります。 linkedin.com/in/...形式の公開個人プロフィールURLを使用してください。このenrichmentはSales NavigatorのURLに対応していません。

この関数はuniqueIdを生成し、POST前に表示します。リクエストでは、その値に加えて waterfall-phone-finder enrichment ID、キャッシュ設定、入力データを送信します。正常に受け付けられるとHTTP 202 Acceptedとタスクの@idが返り、処理はバックグラウンドで続行されます。

Waterfall Phone Finder API リファレンスでは、現在の設定、入力フィールド、出力、クレジット、制限を確認できます。

見込み客が1件の場合、リクエストボディは次の形式です。

json

{
  "enrichmentId": "waterfall-phone-finder",
  "uniqueId": "generated-uuid",
  "settings": {
    "cacheAge": "30"
  },
  "inputs": [
    {
      "inputId": "prospect-florian-poullin",
      "data": {
        "profileUrl": "https://www.linkedin.com/in/florianpoullin/"
      }
    }
  ]
}

公開プロフィールURLが検索入力です。inputIdは、結果をソースレコード内の氏名や担当者と結び付けるための結合キーです。自社システム内で変わらず、同じタスク内で一意になるIDを選んでください。

バックグラウンドタスクの完了を待つ

Requestsクライアントは3秒ごとにタスクを確認する間、ローカルで処理をブロックします。 Datablistのタスク自体はバックグラウンドで実行されます。ステータスがcompletedになると、関数は結果の取得へ進みます。failedまたはstoppedの場合は、確認に使えるタスクIDを含むエラーが発生します。

すべての結果ページを取得する

結果リクエストでは、1回につき最大100行を取得します。nextCursorがnullになるまで、ページの取得を続けます。各結果にはソースのinputIdが含まれるため、レスポンスの順序に依存せず、正しい見込み客へマッピングできます。

Waterfallで電話番号が見つかる仕組み

この例ではcacheAgeを"30"に設定しています。Datablistは最初に、30日以内の利用可能な検索結果があるか確認し、存在すればAPIで再利用できます。この設定はキャッシュの利用を許可するものであり、キャッシュヒットを保証するものではありません。

プロフィールが未解決のままの場合、managed waterfallが利用可能な電話番号データプロバイダーを順番に試します。利用できる番号が返らなければ次のプロバイダーへ進み、利用可能な番号が見つかると、そのレコードの検索を終了します。

結果がない場合は検索を続け、電話番号が見つかると停止するWaterfall電話番号検索

国や市場をまたぐ検索にWaterfallが有効な理由

Datablistは、単一の電話番号データプロバイダーだけに依存していません。各プロバイダーが使うデータセットは異なり、カバー範囲も国や市場によって変わります。あるプロバイダーに電話番号がなくても、別のプロバイダーが見つけられる場合があります。Datablistなら、これらを同じAPI連携で利用できます。

複数市場向けの架電リストを作る営業チームにとっては、最初のプロバイダーで結果がなくても、検索対象から漏れる見込み客を減らせます。Pythonコードは1つのタスクを送信し、統一された結果形式を読むだけなので、プロバイダーごとに個別の連携を構築・保守する必要はありません。

ワークフローで明示的に最新の検索結果が必要な場合は、cacheAgeを"0"に設定してください。この例では、利用可能なキャッシュ結果を再利用できるよう"30"のままにします。

検索結果全体を確認する

以下はFlorian Poullinの検索結果例です。電話番号はマスキングしています。

json

{
  "inputId": "prospect-florian-poullin",
  "status": "success",
  "outputs": {
    "finderProvider": "wizleads",
    "phoneNumber": "+33 • •• •• •• ••",
    "phoneNumberCountry": "FR"
  },
  "creditsUsed": "500",
  "error": null
}

電話番号が見つかった場合、phoneNumberには国際形式の番号が入ります。 +33のプレフィックスとphoneNumberCountry: "FR"は、フランスの電話番号であることを示します。

finderProviderは、この結果を返したfinderです。この行では500クレジットが使用され、 error: nullはAPIが行単位のエラーを返していないことを意味します。

確認用の営業電話リストを作成する

phone_resultsを取得したら、営業チームが確認するためのローカルリストを作成できます。同じファイル内で、結果を表示する処理の後に次のブロックを追加してください。既存の結果を再利用するため、2回目のAPI呼び出しは発生しません。

python

1results_by_id = {
2    result["inputId"]: result
3    for result in phone_results
4}
5
6call_list = []
7
8for prospect in prospects:
9    result = results_by_id.get(prospect["prospect_id"])
10    if not result or result["status"] != "success":
11        continue
12
13    phone_number = result.get("outputs", {}).get("phoneNumber")
14    if not phone_number:
15        continue
16
17    call_list_record = prospect.copy()
18    call_list_record["result"] = result
19    call_list.append(call_list_record)
20
21print(json.dumps(call_list, indent=2, ensure_ascii=False))

結果のインデックスではinputIdを結合キーとして使うため、リスト上の位置に依存せず、各結果とソースレコードの対応を維持できます。API payloadには各見込み客のIDとプロフィールURLを含め、元のprospectsリストには架電リストで必要な氏名と担当者を残します。

Florianの行では、prospect-florian-poullinにより、返された結果が元のname、 linkedin_url、account_ownerと結び付きます。チームはresult.outputsから電話番号と国コードFRを確認できます。ステータス、クレジット、エラーを含むAPIオブジェクト全体も、 resultの下に保持されます。

架電リストのループはprospectsの順序に従います。対象の各見込み客をコピーしてから resultを付加するため、元のソースリストは修正や別のフォローアップチャネルに利用できます。

結果なし、エラー、電話番号が空の行は、変更されていないprospectsリストには残りますが、 call_listには追加されません。このコードにより、確認用の営業電話リストがローカルに作成されます。

Waterfall Phone Finder APIの料金を試算する

デフォルトのmanaged waterfallで新しい未キャッシュの電話番号が見つかると、1件につき 500クレジットを使用します。利用可能なキャッシュ結果は0クレジット、結果が見つからない検索も 0クレジットです。実際の消費量は、送信した全プロフィール数に500を掛けるのではなく、返された各行のcreditsUsedで確認してください。

予算計画には、次の式を使用します。

新たに見つかった未キャッシュの電話番号数 × 500クレジット

新しい番号40件なら20,000クレジット、100件なら50,000クレジットです。現在のUSD建て基本追加購入レートでは、 $20で20,000クレジットを購入できるため、500クレジットは$0.50相当です。REST APIへのアクセスを提供する Growthサブスクリプションは別料金です。現在のプランとクレジットパッケージは Datablistの料金ページでご確認ください。

試算例として、100件のプロフィールから新しい番号20件、利用可能なキャッシュ番号5件が見つかり、 75件は結果なしだったとします。課金対象は新しい20件なので、検索には10,000クレジットを使用します。架電リストには、見つかった25件すべてを追加できます。計算ツールには、入力総数の100や、見つかった総数の25ではなく、20を入力してください。この例は料金計算を示すものであり、想定成功率を示すものではありません。

Loading pricing…

よくあるエラーの解決と既存タスクの復元

DATABLIST_API_KEYが未設定： スクリプトはリクエスト前に停止します。環境変数をexportしてから、ファイルを再実行してください。
APIキーが拒否される： raise_for_status()がHTTPエラーを返します。 Personal API Keyをコピーし直し、リクエストのX-API-Keyヘッダーで送信されていることを確認してください。
プロフィールURLが未対応： 空欄、不正な形式、企業プロフィール、Sales NavigatorのURLは、 linkedin.com/in/...形式の公開個人プロフィールURLへ置き換えてください。
クレジットが不足している： 追加タスクを送信する前に、アカウント残高と利用可能なクレジットを確認してください。
タスクがfailedまたはstopped： 再実行が必要な処理を判断する前に、エラー、取得済みの結果、クレジット使用量を確認してください。
行に結果がない、またはエラーがある： 修正や別のフォローアップチャネルに備えて、ソースの見込み客情報を残してください。その行だけでは、どのプロバイダー、地域、公開範囲の条件が結果に影響したかは判断できません。

POSTレスポンスを取得できなかった場合は、送信前に表示されたuniqueIdを使用します。 再送信する前にタスク一覧を確認してください。 次のリクエストのyour-api-keyを置き換え、表示された値とuniqueIdが一致するタスクを探します。

shell

1curl --get "https://api.datablist.com/v1/enrichments/tasks" \
2  --header "X-API-Key: your-api-key" \
3  --data-urlencode "enrichmentId=waterfall-phone-finder" \
4  --data-urlencode "limit=100"

タスクが最初のページにない場合は、タスク一覧レスポンスのnextCursorをたどってください。一致するタスクの@idをコピーし、以下のTASK_IDを置き換えて、ステータスの確認または既存結果の取得を行います。

shell

1curl "https://api.datablist.com/v1/enrichments/tasks/TASK_ID" \
2  --header "X-API-Key: your-api-key"
3
4curl --get "https://api.datablist.com/v1/enrichments/tasks/TASK_ID/results" \
5  --header "X-API-Key: your-api-key" \
6  --data-urlencode "limit=100"

タスクAPIドキュメントでは、タスクの状態、結果のページネーション、中断時の動作について説明しています。

次のステップ

これで、公開LinkedInプロフィールURLに再利用できるRequests関数と、見つかった電話番号を営業担当者の確認用にまとめる固定IDベースの結合処理が用意できました。

編集可能なサンプルを、少数の有望な見込み客に置き換えてください。固定IDと担当者を維持し、見つかった各番号を確認してから、チームの架電プロセスへリストを渡します。まずは少量のバッチで、 IDのマッピングと返されるフィールドを確認することをおすすめします。ノンブロッキングな並行処理が必要な場合は、 Datablistが管理するasync HTTPXのサンプルをご利用ください。