Как найти сайты компаний по названию через API Python

Чтобы найти сайт компании по её названию с помощью Python, передайте название и, если известно, страну в Company Name to Domain API от Datablist. API обработает запрос в фоновом режиме, а затем вернёт домен и полный URL сайта.

Сначала разберём простой API-запрос на Python. Затем используем ту же функцию, чтобы заполнить только отсутствующие сайты в списке компаний, сохранив уже имеющиеся URL.

Если вам удобнее интерфейс в виде таблицы, воспользуйтесь инструкцией по массовому поиску сайтов по названиям компаний.

💡 Кратко для тех, кто спешит

Создайте одну функцию на Python для поиска сайтов по названиям компаний, а затем используйте её, чтобы заполнить записи с пустыми полями, не заменяя уже известные URL.

• Входные данные: названия компаний и, при наличии, страны.
• Результат: домен, полный URL сайта, уровень уверенности, статус результата и количество использованных кредитов.

Настройка Python и API-ключа Datablist

В этом примере для синхронных HTTP-запросов используется пакет Requests. Установите его в виртуальном окружении:

1python3 -m venv .venv
2source .venv/bin/activate
3python -m pip install requests

Также понадобится тариф Datablist Growth, поскольку доступ к REST API входит в возможности этого тарифа. Создайте Personal API Key на странице API вашего аккаунта Datablist, а затем сохраните его в переменной окружения:

1export DATABLIST_API_KEY="exampleuser-exampleaccount-EXAMPLE_TOKEN_NOT_VALID-2"

🔑 Уже есть API-ключ? Старые ключи имеют формат auth0|<userId>:<accountId>:<secret>:1. Они по-прежнему поддерживаются. Продолжайте использовать существующий ключ именно в том виде, в котором он был создан: заменять или менять его формат не нужно.

Замените значение из примера на ключ, созданный на странице вашего аккаунта. Поиск сайтов компаний расходует кредиты. Перед масштабированием скрипта мы разберём сценарии стоимостью 1 и 2,5 кредита. Актуальные условия тарифов приведены на странице цен Datablist.

Поиск сайта одной компании через API в Python

Создайте файл find_company_websites.py. Скрипт отправляет названия, дожидается завершения задачи и получает результаты в рамках одной функции. Для каждого входного объекта ваш постоянный ID записи передаётся как inputId. Благодаря этому результат можно сопоставить с нужной записью, даже если порядок ответа изменится.

1import json
2import os
3import time
4import uuid
5
6import requests
7
8
9API_BASE_URL = "https://api.datablist.com/v1"
10
11
12def find_domains(companies, api_key):
13    headers = {
14        "Accept": "application/json",
15        "Content-Type": "application/json",
16        "X-API-Key": api_key,
17    }
18    inputs = [
19        {
20            "inputId": company["id"],
21            "data": {
22                "companyName": company["company_name"],
23                "targetCountry": company.get("country", ""),
24            },
25        }
26        for company in companies
27    ]
28    unique_id = f"python-domain-lookup-{uuid.uuid4().hex}"
29    payload = {
30        "enrichmentId": "company-name-to-domain",
31        "uniqueId": unique_id,
32        "settings": {
33            "workflow": "dataset_google",
34            "countryOverwrite": True,
35            "acceptNonRoot": True,
36        },
37        "inputs": inputs,
38    }
39
40    print(f"Task uniqueId: {unique_id}")
41    response = requests.post(
42        f"{API_BASE_URL}/enrichments/tasks",
43        headers=headers,
44        json=payload,
45    )
46    response.raise_for_status()
47    task_id = response.json()["@id"]
48
49    while True:
50        response = requests.get(
51            f"{API_BASE_URL}/enrichments/tasks/{task_id}",
52            headers=headers,
53        )
54        response.raise_for_status()
55        task = response.json()
56        if task["status"] in {"completed", "failed", "stopped"}:
57            break
58        time.sleep(3)
59
60    if task["status"] != "completed":
61        raise RuntimeError(
62            f"Task {task_id} ended with status {task['status']}: {task.get('error')}"
63        )
64
65    results = []
66    cursor = None
67    while True:
68        params = {"limit": 100}
69        if cursor is not None:
70            params["cursor"] = cursor
71
72        response = requests.get(
73            f"{API_BASE_URL}/enrichments/tasks/{task_id}/results",
74            headers=headers,
75            params=params,
76        )
77        response.raise_for_status()
78        page = response.json()
79        results.extend(page["results"])
80        cursor = page.get("nextCursor")
81        if cursor is None:
82            break
83
84    return results
85
86
87api_key = os.environ["DATABLIST_API_KEY"]
88companies = [
89    {
90        "id": "direct-microsoft-us",
91        "company_name": "Microsoft",
92        "country": "US",
93    }
94]
95
96lookup_results = find_domains(companies, api_key)
97for result in lookup_results:
98    print(json.dumps(result, indent=2))

Запустите его из терминала, в котором вы экспортировали ключ:

1python find_company_websites.py

Вот результат для Microsoft:

{
  "inputId": "direct-microsoft-us",
  "status": "success",
  "outputs": {
    "companyUrl": "https://microsoft.com",
    "companyDomain": "microsoft.com",
    "domainConfidence": "High"
  },
  "creditsUsed": "1",
  "error": null
}

Используйте inputId, чтобы сопоставить результат с исходной записью. Поле status принимает значения success, no_result, error или skipped. Успешный результат может содержать companyDomain, полный адрес companyUrl и уровень уверенности domainConfidence. Поле creditsUsed показывает стоимость обработки этого объекта, а error объясняет причину ошибки на уровне отдельной записи.

Функция последовательно выполняет три типа API-запросов.

Сначала она отправляет запрос на поиск. Payload содержит настройки и записи компаний. POST-запрос возвращает 202 Accepted, а поле @id содержит ID задачи, необходимый для следующих запросов:

1response = requests.post(
2    f"{API_BASE_URL}/enrichments/tasks",
3    headers=headers,
4    json=payload,
5)
6response.raise_for_status()
7task_id = response.json()["@id"]

Если переменная DATABLIST_API_KEY не задана, Python выдаст KeyError при обращении к переменной окружения ещё до отправки HTTP-запроса. Экспортируйте переменную и снова запустите скрипт. Если в переменной указан недействительный ключ, POST-запрос дойдёт до API, но вернёт HTTP 401. Скопируйте актуальный Personal API Key со страницы аккаунта и замените неверное значение.

Затем функция опрашивает статус задачи. POST-запрос запускает фоновую обработку, поэтому скрипт проверяет задачу каждые три секунды, пока она не перейдёт в статус completed, failed или stopped:

1while True:
2    response = requests.get(
3        f"{API_BASE_URL}/enrichments/tasks/{task_id}",
4        headers=headers,
5    )
6    response.raise_for_status()
7    task = response.json()
8    if task["status"] in {"completed", "failed", "stopped"}:
9        break
10    time.sleep(3)

Если задача завершается с ошибкой из-за нехватки кредитов, в поле причины будет указано credits_error. Пополните баланс или дождитесь ежемесячного обновления кредитов, проверьте уже полученные результаты, а затем отправьте в новой задаче только необработанные записи.

Наконец, функция получает все страницы с результатами. Каждый ответ добавляет результаты в список и передаёт cursor для следующего запроса:

1results.extend(page["results"])
2cursor = page.get("nextCursor")
3if cursor is None:
4    break

Скрипт загружает следующие страницы, пока nextCursor не станет равен null.

Перед POST-запросом скрипт выводит созданный uniqueId. Если соединение прервётся до получения ответа о создании задачи, сначала запросите список задач и проверьте их статус, прежде чем создавать новую:

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

Найдите задачу, у которой uniqueId совпадает со значением, созданным скриптом, и скопируйте её @id. Endpoint списка поддерживает фильтрацию по enrichmentId, но не по uniqueId. Используйте @id в тех же URL статуса и результатов, что и скрипт. Если задача всё ещё выполняется, продолжайте проверять её статус. Если она остановлена или завершилась с ошибкой, проверьте частичные результаты перед повторной отправкой записей.

Сопоставление данных и поиск с учётом страны

Скрипт использует dataset_google — рекомендуемый вариант по умолчанию. Сначала Datablist проверяет собственные датасеты компаний. Если совпадение не найдено или данные противоречат друг другу, сервис может использовать Google как резервный источник. Перед возвратом уровня уверенности он проверяет возможные совпадения.

Страна помогает уточнить поиск для неоднозначных названий. В этом примере одно и то же название компании дало разные результаты:

• Для Mercury со страной US были возвращены shipmercury.com и https://www.shipmercury.com/.
• Для Mercury со страной IE были возвращены mercuryeng.com и http://www.mercuryeng.com.

В обоих случаях уровень уверенности был High. Страна помогает выбрать нужную компанию среди организаций с одинаковыми названиями, а уровень уверенности — определить, какие результаты требуют проверки.

Если совпадение также зависит от отрасли, размера компании или другого бизнес-контекста, используйте процесс исследования с AI Agent. Добавьте эти сведения в prompt для AI Agent, чтобы он мог найти нужную компанию с учётом более подробного бизнес-контекста.

Заполнение только отсутствующих сайтов в записях

Теперь, когда find_domains() умеет находить сайт по названию компании, используйте эту функцию для заполнения только отсутствующих сайтов в записях. Записи, в которых сайт уже указан, не отправляются в API: их значения остаются неизменными, а кредиты не расходуются.

В файле find_company_websites.py замените весь код от companies = [ до последнего цикла вывода на приведённый ниже блок. Оставьте расположенное выше присваивание api_key. Проверка пустого значения учитывает None, пустую строку и строку из пробелов. Если сайты заполнены во всех записях, API-запрос не выполняется.

1records = [
2    {
3        "id": "crm-001",
4        "company_name": "Microsoft",
5        "country": "US",
6        "website": "https://www.microsoft.com",
7    },
8    {
9        "id": "crm-002",
10        "company_name": "boulangerie du palais lyon",
11        "country": "FR",
12        "website": "",
13    },
14    {
15        "id": "crm-003",
16        "company_name": "Mercury Engineering",
17        "country": "IE",
18        "website": None,
19    },
20    {
21        "id": "crm-004",
22        "company_name": "Mercury",
23        "country": "US",
24        "website": "   ",
25    },
26    {
27        "id": "crm-005",
28        "company_name": "Datablist",
29        "country": "FR",
30        "website": "https://www.datablist.com",
31    },
32    {
33        "id": "crm-006",
34        "company_name": "Mercury",
35        "country": "IE",
36        "website": None,
37    },
38]
39
40
41def website_is_missing(value):
42    return not isinstance(value, str) or not value.strip()
43
44
45records_to_lookup = [
46    record for record in records if website_is_missing(record.get("website"))
47]
48lookup_results = (
49    find_domains(records_to_lookup, api_key) if records_to_lookup else []
50)
51results_by_id = {result["inputId"]: result for result in lookup_results}
52
53for record in records:
54    if not website_is_missing(record.get("website")):
55        record["website_lookup"] = {"status": "not_requested"}
56        continue
57
58    result = results_by_id.get(record["id"])
59    if result is None:
60        record["website_lookup"] = {
61            "status": "missing_result",
62            "domain": None,
63            "confidence": None,
64            "credits": None,
65            "error": {
66                "code": "missing_result",
67                "message": "No result was returned for this inputId.",
68            },
69        }
70        continue
71
72    outputs = result.get("outputs") or {}
73    if result["status"] == "success" and outputs.get("companyUrl"):
74        record["website"] = outputs["companyUrl"]
75
76    record["website_lookup"] = {
77        "status": result["status"],
78        "domain": outputs.get("companyDomain"),
79        "confidence": outputs.get("domainConfidence"),
80        "credits": result.get("creditsUsed"),
81        "error": result.get("error"),
82    }
83
84print(json.dumps(records, indent=2))

Код индексирует результаты по inputId, а затем последовательно обходит исходные записи. Объект website_lookup хранит результат и сведения для проверки рядом с каждой обработанной записью, включая неудачные и отсутствующие результаты.

Для crm-001 функция website_is_missing() возвращает False, поскольку URL Microsoft уже указан. Запись не добавляется в records_to_lookup, поэтому API её не получает. При объединении сайт сохраняется, а статус записывается как not_requested.

У crm-002 поле сайта пустое, поэтому в запрос отправляется пекарня с inputId: crm-002. API возвращает тот же inputId. По нему results_by_id находит нужный результат, после чего при объединении в запись добавляется https://boulangerie-du-palais-lyon.eatbu.com — независимо от порядка результатов.

Для пекарни используется бизнес-поддомен boulangerie-du-palais-lyon.eatbu.com. Сохраните это значение, не сокращая его до eatbu.com, и записывайте точный адрес из companyUrl. Настройка acceptNonRoot: true также разрешает возвращать страницу компании на сторонней платформе, адрес которой содержит путь после имени хоста. Поэтому в обоих случаях важен полный URL.

Даже завершённая задача может содержать объекты со статусами no_result, error или skipped. В примере поле website изменяется только при статусе success и наличии companyUrl. При статусе no_result сайт остаётся пустым, а статус сохраняется в website_lookup. Сначала проверьте название компании и страну. Если какое-либо из этих значений неверно, исправьте его и отправьте запись в новой задаче. В противном случае оставьте поле сайта пустым для ручной проверки.

Расчёт кредитов перед масштабированием скрипта

Все пять запросов из примера вернули статус success с уровнем уверенности High и суммарно использовали 5 кредитов.

Поиск по датасету стоит 1 кредит, а поиск через Google — 2,5 кредита. Вы платите только за использованный способ, а не за оба. Поэтому для 1 000 записей потребуется 1 000 кредитов, если все совпадения будут найдены в датасете, или 2 500 кредитов, если все запросы будут обработаны через Google. Это сценарии для планирования, а не гарантия успешного результата для каждого запроса.

Перемещайте ползунок, чтобы сравнить расход кредитов для своего списка.

При цене $20 за 20 000 кредитов эти сценарии обойдутся соответственно в $1 и $2,50. Подписка Growth оплачивается отдельно и сейчас включает 20 000 ежемесячных кредитов, которые расходуются до купленных дополнительно. Проверяйте актуальные цены, поскольку условия тарифов и пакетов могут меняться.

Одна задача принимает до 10 000 входных объектов. Лимит на один объект составляет 16 КиБ, а на запрос — 10 МиБ. В этом примере блокирующие вызовы Requests используются для отправки, проверки статуса и получения результатов одной небольшой асинхронной задачи Datablist. В поддерживаемом репозитории также есть примеры пакетной обработки CSV и неблокирующих клиентов.

Следующие шаги

Убедившись, что найденный сайт принадлежит нужной компании, вы можете собрать данные из Impressum с помощью Python, чтобы добавить юридические сведения о компании и URL страницы-источника в ту же запись поставщика. Проверяйте совпадение сайта отдельно от последующего извлечения данных.

Замените тестовые записи небольшим фрагментом собственных данных, запустите скрипт и проверьте объединённые записи. Используйте постоянные ID и указывайте страну. Масштабируйте обработку только после того, как убедитесь, что скрипт сохраняет известные сайты и заполняет нужные пустые поля. Если вам нужен готовый к скачиванию скрипт, пакетная обработка CSV или асинхронная версия, начните с поддерживаемых примеров Company Name to Domain API на GitHub.