Cómo encontrar webs de empresas con Python y una API

Para encontrar el sitio web de una empresa a partir de su nombre con Python, envíe el nombre y, si dispone de él, el país a la API Company Name to Domain de Datablist. La API procesa la búsqueda en segundo plano y devuelve el dominio y la URL completa del sitio web.

Primero le mostraré una solicitud sencilla a la API con Python. Después, reutilizaré la misma función para completar únicamente los sitios web que falten en una lista de empresas, sin modificar las URL existentes.

Si prefiere una interfaz similar a una hoja de cálculo, consulte la guía para buscar sitios web de empresas en bloque.

💡 Resumen para quienes tienen prisa

Cree una función de Python que encuentre sitios web a partir del nombre de la empresa y úsela para completar registros sin sitio web, sin sustituir las URL conocidas.

• Entrada: nombres de empresas y, opcionalmente, el país.
• Salida: dominio, URL completa, nivel de confianza, estado del resultado y créditos utilizados.

Configure Python y su clave API de Datablist

Para este ejemplo, utilizo el paquete Requests para realizar llamadas HTTP síncronas. Instálelo en un entorno virtual:

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

También necesita un plan Growth de Datablist, ya que el acceso a la REST API es una función de Growth. Cree una Personal API Key desde la página de API de su cuenta de Datablist y guárdela en una variable de entorno:

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

🔑 ¿Ya tiene una clave API? Las claves antiguas utilizan auth0|<userId>:<accountId>:<secret>:1 y siguen siendo compatibles. Continúe usando su clave actual exactamente como se generó; no necesita sustituirla ni cambiar su formato.

Sustituya el valor de ejemplo por la clave generada en la página de su cuenta. Las búsquedas de sitios web consumen créditos. Antes de escalar el script, explicaré cuándo se cobra 1 crédito y cuándo 2,5 créditos. Puede consultar los valores actuales de los planes en la página de precios de Datablist.

Busque el sitio web de una empresa con Python

Cree un archivo llamado find_company_websites.py. El script envía los nombres, espera a que termine la tarea y obtiene los resultados mediante una sola función. Cada entrada utiliza el ID estable de su registro como inputId, de modo que pueda asociar cada resultado con el registro correcto aunque cambie el orden de la respuesta.

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))

Ejecútelo desde el terminal en el que exportó la clave:

1python find_company_websites.py

Este es el resultado de Microsoft:

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

Utilice inputId para asociar el resultado con el registro original. status puede ser success, no_result, error o skipped. Un resultado correcto puede incluir companyDomain, la URL completa en companyUrl y domainConfidence. creditsUsed indica el coste de esa entrada, mientras que error explica cualquier fallo individual.

La función realiza tres tipos de llamadas a la API, en este orden.

Primero, envía la búsqueda. El payload contiene la configuración y los registros de las empresas. La solicitud POST devuelve 202 Accepted, y @id es el ID de tarea necesario para las siguientes llamadas:

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"]

Si DATABLIST_API_KEY no está definida, Python genera un KeyError al consultar el entorno, antes de realizar la solicitud HTTP. Exporte la variable y vuelva a ejecutar el script. Si contiene una clave no válida, la solicitud POST llegará a la API y devolverá un error HTTP 401. Copie una Personal API Key vigente desde la página de su cuenta y sustituya el valor incorrecto.

Segundo, consulta periódicamente la tarea. La solicitud POST inicia un proceso en segundo plano, por lo que el script consulta la tarea cada tres segundos hasta que alcanza el estado completed, failed o 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)

Si la tarea falla por falta de créditos, el motivo del error será credits_error. Añada créditos o espere a que se renueven sus créditos mensuales, revise los resultados ya devueltos y envíe únicamente las entradas pendientes en una nueva tarea.

Tercero, recupera todas las páginas de resultados. Cada respuesta añade sus resultados a la lista y proporciona el cursor para la siguiente solicitud:

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

El script continúa obteniendo páginas hasta que nextCursor es null.

Antes de la solicitud POST, el script muestra el uniqueId generado. Si la conexión se interrumpe antes de recibir la respuesta de creación, consulte primero la lista de tareas y compruebe su estado antes de crear otra:

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"

Busque la tarea cuyo uniqueId coincida con el valor generado por el script y copie su @id. El endpoint de listado permite filtrar por enrichmentId, pero no por uniqueId. Utilice el @id con las mismas URL de estado y resultados empleadas por el script. Si la tarea sigue activa, continúe consultándola. Si se ha detenido o ha fallado, revise si existen resultados parciales antes de volver a enviar las entradas.

Cómo funciona la búsqueda por país y dataset

El script utiliza dataset_google, la opción predeterminada recomendada. Datablist consulta primero sus datasets de empresas. Si no encuentra una coincidencia o detecta información contradictoria, puede recurrir a Google. Antes de devolver un nivel de confianza, comprueba las posibles coincidencias.

El país ayuda a resolver nombres ambiguos. En este ejemplo, el mismo nombre de empresa produjo resultados diferentes:

• Mercury con US devolvió shipmercury.com y https://www.shipmercury.com/.
• Mercury con IE devolvió mercuryeng.com y http://www.mercuryeng.com.

Ambos obtuvieron un nivel de confianza High. El país permite elegir entre empresas con el mismo nombre, mientras que el nivel de confianza le ayuda a decidir qué resultados debe revisar.

Cuando la coincidencia también dependa del sector, el tamaño de la empresa u otros datos comerciales, utilice el workflow de investigación con AI Agent. Añada esos datos al prompt de AI Agent para que pueda investigar la empresa correcta con más contexto de negocio.

Complete solo los sitios web que faltan

Ahora que find_domains() puede encontrar un sitio web a partir del nombre de una empresa, reutilice la función para completar únicamente los sitios web que falten en sus registros. Los registros que ya tengan un sitio web no se envían a la API, por lo que sus valores permanecen intactos y no consumen créditos de búsqueda.

En find_company_websites.py, sustituya todo el contenido desde companies = [ hasta el bucle final de impresión por el siguiente bloque. Conserve la asignación de api_key que aparece antes. La comprobación de campos vacíos contempla None, una cadena vacía y espacios en blanco. Si todos los registros tienen sitio web, no se realiza ninguna llamada a la 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))

El código indexa los resultados por inputId y después recorre los registros originales en el mismo orden. El objeto website_lookup conserva el resultado y los datos necesarios para revisar cada registro procesado, incluidos los resultados fallidos o ausentes.

En crm-001, website_is_missing() devuelve False porque la URL de Microsoft ya está presente. El registro no se añade a records_to_lookup, por lo que la API nunca lo recibe. Al combinar los datos, se conserva el sitio web y se registra el estado not_requested.

En crm-002, el sitio web está vacío, por lo que la solicitud envía la panadería con inputId: crm-002. La API devuelve el mismo inputId. results_by_id lo utiliza para localizar el resultado correcto y la combinación guarda https://boulangerie-du-palais-lyon.eatbu.com en ese registro, sin depender del orden de los resultados.

La panadería utiliza un subdominio propio del negocio: boulangerie-du-palais-lyon.eatbu.com. Conserve ese valor en lugar de reducirlo a eatbu.com y guarde la dirección exacta de companyUrl. La opción acceptNonRoot: true también admite una página alojada cuya dirección incluya una ruta después del hostname, por lo que la URL completa es importante en ambos casos.

Una tarea completada puede contener elementos con estado no_result, error o skipped. El ejemplo solo modifica website cuando el estado es success y existe companyUrl. Si el estado es no_result, el sitio web permanece vacío y el estado sigue visible en website_lookup. Compruebe primero el nombre y el país de la empresa. Si alguno es incorrecto, corríjalo y envíe ese registro en una nueva tarea. De lo contrario, mantenga el sitio web vacío para revisarlo manualmente.

Calcule los créditos antes de escalar el script

Las cinco búsquedas del ejemplo devolvieron success con un nivel de confianza High y consumieron 5 créditos en total.

Una búsqueda resuelta mediante el dataset cuesta 1 crédito. Una búsqueda procesada con Google cuesta 2,5 créditos. Se cobra la vía utilizada, no ambas. Para 1000 entradas, esto supone 1000 créditos si todas se resuelven mediante el dataset o 2500 créditos si todas utilizan Google. Son escenarios orientativos, no una garantía de que todas las búsquedas vayan a ofrecer resultados.

Mueva el control deslizante para comparar el consumo de créditos de su lista.

Con un precio de 20 $ por 20 000 créditos, esos escenarios cuestan 1 $ y 2,50 $. La suscripción Growth se cobra por separado y actualmente incluye 20 000 créditos mensuales, que se utilizan antes que las recargas. Consulte los precios actuales, ya que los planes y paquetes pueden cambiar.

Cada tarea admite hasta 10 000 entradas, con un límite de 16 KiB por entrada y 10 MiB por solicitud. Este ejemplo utiliza llamadas bloqueantes de Requests para enviar, consultar y recuperar los resultados de una pequeña tarea asíncrona de Datablist. El repositorio mantenido también incluye ejemplos de procesamiento de CSV por lotes y clientes no bloqueantes.

Próximos pasos

Cuando confirme que el sitio web devuelto pertenece a la empresa correcta, puede extraer su Impressum con Python para añadir al mismo registro del proveedor los datos legales de la empresa y la URL de la página legal utilizada como fuente. Mantenga la revisión de la coincidencia del sitio web separada de esta extracción posterior.

Sustituya los registros de ejemplo por una pequeña muestra de sus propios datos, ejecute el script y revise los registros combinados. Conserve IDs estables y el país, y amplíe el volumen solo después de comprobar que el script respeta los sitios web conocidos y completa los campos vacíos previstos. Si necesita descargar el script, procesar archivos CSV por lotes o utilizar una versión asíncrona, puede partir de los ejemplos mantenidos de la API Company Name to Domain en GitHub.