如何用 Python API 根据企业网站批量查找 LinkedIn 公司主页

1mkdir company-linkedin-matcher
2cd company-linkedin-matcher
3python -m venv .venv
4source .venv/bin/activate
5python -m pip install requests
6touch find_linkedin_company_pages.py

1export DATABLIST_API_KEY="your-api-key"

1import json
2import os
3import time
4import uuid
5
6import requests
7
8
9def match_companies(accounts, api_key, matching_direction):
10    """Submit one matching task, wait for it, 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": account["account_id"],
16            "data": {"sourceUrl": account["source_url"]},
17        }
18        for account in accounts
19    ]
20
21    unique_id = str(uuid.uuid4())
22    print(f"Task uniqueId: {unique_id}", flush=True)
23    payload = {
24        "enrichmentId": "company-to-linkedin",
25        "uniqueId": unique_id,
26        "settings": {"matchingDirection": matching_direction},
27        "inputs": inputs,
28    }
29    response = requests.post(tasks_url, headers=headers, json=payload)
30    response.raise_for_status()
31    task = response.json()
32    task_id = task["@id"]
33    task_url = f"{tasks_url}/{task_id}"
34
35    while task["status"] not in ("completed", "failed", "stopped"):
36        time.sleep(3)
37        response = requests.get(task_url, headers=headers)
38        response.raise_for_status()
39        task = response.json()
40
41    if task["status"] != "completed":
42        raise RuntimeError(
43            f"Task {task_id} {task['status']}: {task.get('error')}"
44        )
45
46    results = []
47    results_url = f"{task_url}/results"
48    params = {"limit": 100}
49    while True:
50        response = requests.get(results_url, headers=headers, params=params)
51        response.raise_for_status()
52        page = response.json()
53        results.extend(page["results"])
54        next_cursor = page["nextCursor"]
55        if next_cursor is None:
56            break
57        params["cursor"] = next_cursor
58
59    return results
60
61
62accounts = [
63    {
64        "account_id": "account-datablist",
65        "company_name": "Datablist",
66        "source_url": "datablist.com",
67    },
68    {
69        "account_id": "account-posthog",
70        "company_name": "PostHog",
71        "source_url": "posthog.com",
72    },
73    {
74        "account_id": "account-n8n",
75        "company_name": "n8n",
76        "source_url": "https://n8n.io",
77    },
78    {
79        "account_id": "account-vaude",
80        "company_name": "VAUDE",
81        "source_url": "vaude.com",
82    },
83    {
84        "account_id": "account-drooms",
85        "company_name": "Drooms",
86        "source_url": "https://www.drooms.com",
87    },
88]
89
90api_key = os.environ.get("DATABLIST_API_KEY")
91if not api_key:
92    raise RuntimeError("Set DATABLIST_API_KEY before running this script.")
93
94match_results = match_companies(
95    accounts, api_key, matching_direction="domain_to_linkedin"
96)
97print(json.dumps(match_results, indent=2))

1python find_linkedin_company_pages.py

1response = requests.post(tasks_url, headers=headers, json=payload)
2response.raise_for_status()
3task = response.json()
4task_id = task["@id"]

[
  {
    "inputId": "account-datablist",
    "status": "success",
    "outputs": {
      "companyWebsite": "datablist.com",
      "confidence": "high",
      "companyLinkedInUrl": "https://www.linkedin.com/company/datablist",
      "companyName": "datablist"
    },
    "creditsUsed": "1.5",
    "error": null
  },
  {
    "inputId": "account-posthog",
    "status": "success",
    "outputs": {
      "companyWebsite": "posthog.com",
      "confidence": "high",
      "companyLinkedInUrl": "https://www.linkedin.com/company/posthog",
      "companyName": "PostHog"
    },
    "creditsUsed": "1.5",
    "error": null
  },
  {
    "inputId": "account-n8n",
    "status": "success",
    "outputs": {
      "companyWebsite": "n8n.io",
      "confidence": "high",
      "companyLinkedInUrl": "https://www.linkedin.com/company/n8n",
      "companyName": "n8n"
    },
    "creditsUsed": "1.5",
    "error": null
  },
  {
    "inputId": "account-vaude",
    "status": "success",
    "outputs": {
      "companyWebsite": "vaude.com",
      "confidence": "high",
      "companyLinkedInUrl": "https://www.linkedin.com/company/vaude",
      "companyName": "vaude"
    },
    "creditsUsed": "1.5",
    "error": null
  },
  {
    "inputId": "account-drooms",
    "status": "success",
    "outputs": {
      "companyWebsite": "drooms.com",
      "confidence": "low",
      "companyLinkedInUrl": "https://www.linkedin.com/company/drooms-software",
      "companyName": "Drooms"
    },
    "creditsUsed": "1.5",
    "error": null
  }
]

1results_by_id = {result["inputId"]: result for result in match_results}
2ready_for_research = []
3needs_review = []
4
5for account in accounts:
6    result = results_by_id.get(account["account_id"])
7    if result is not None:
8        outputs = result.get("outputs", {})
9    else:
10        outputs = {}
11
12    research_record = {
13        **account,
14        "companyLinkedInUrl": outputs.get("companyLinkedInUrl"),
15        "confidence": outputs.get("confidence"),
16        "api_result": result,
17    }
18
19    is_ready = (
20        result is not None
21        and result["status"] == "success"
22        and outputs.get("companyLinkedInUrl")
23        and outputs.get("confidence") == "high"
24    )
25    if is_ready:
26        ready_for_research.append(research_record)
27    else:
28        needs_review.append(research_record)
29
30print("Ready for research:")
31print(json.dumps(ready_for_research, indent=2))
32print("Needs review:")
33print(json.dumps(needs_review, indent=2))

1linkedin_accounts = [
2    {
3        "account_id": "account-datablist",
4        "company_name": "Datablist",
5        "source_url": "https://www.linkedin.com/company/datablist",
6    }
7]
8
9website_results = match_companies(
10    linkedin_accounts,
11    api_key,
12    matching_direction="linkedin_to_domain",
13)

1curl --get "https://api.datablist.com/v1/enrichments/tasks" \
2  --header "X-API-Key: $DATABLIST_API_KEY" \
3  --data-urlencode "enrichmentId=company-to-linkedin" \
4  --data-urlencode "limit=100"