Como encontrar sites de empresas com Python e API

Para encontrar o site de uma empresa a partir do nome em Python, envie o nome e, se disponível, o país para a API Company Name to Domain da Datablist. A API processa a pesquisa em segundo plano e devolve o domínio e a URL completa do site.

Primeiro, vou mostrar uma requisição simples à API em Python. Depois, reutilizarei a mesma função para preencher apenas os sites em falta numa lista de empresas, preservando as URLs já existentes.

Se preferir uma interface semelhante a uma folha de cálculo, consulte o guia para encontrar sites de empresas em massa.

💡 Resumo para quem tem pressa

Crie uma função em Python para encontrar sites a partir de nomes de empresas e use-a para completar registos sem site, sem substituir URLs conhecidas.

• Input: nomes de empresas, com país opcional.
• Output: domínio, URL completa do site, nível de confiança, estado do resultado e créditos utilizados.

Configure o Python e a chave da API Datablist

Neste exemplo, utilizo o pacote Requests para fazer chamadas HTTP síncronas. Instale-o num ambiente virtual:

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

Também precisa de um plano Datablist Growth, pois o acesso à REST API é uma funcionalidade desse plano. Crie uma Personal API Key na página da API da sua conta Datablist e guarde-a numa variável de ambiente:

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

🔑 Já tem uma chave de API? As chaves antigas utilizam auth0|<userId>:<accountId>:<secret>:1 e continuam a ser aceites. Continue a usar a sua chave exatamente como foi gerada; não precisa de a substituir nem de alterar o formato.

Substitua o valor de exemplo acima pela chave gerada na página da sua conta. As pesquisas de sites de empresas consomem créditos. Antes de dimensionarmos o script, explicarei os percursos de 1 e 2,5 créditos. Os valores atuais dos planos estão na página de preços da Datablist.

Pesquise o site de uma empresa em Python

Crie um ficheiro chamado find_company_websites.py. O script envia os nomes, aguarda a conclusão da tarefa e obtém os resultados através de uma única função. Cada input utiliza o ID estável do seu registo como inputId, permitindo associar cada resultado ao registo correto mesmo que a ordem da resposta mude.

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

Execute-o no terminal onde exportou a chave:

1python find_company_websites.py

Este é o resultado da Microsoft:

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

Utilize inputId para associar o resultado ao registo original. O status pode ser success, no_result, error ou skipped. Um resultado bem-sucedido pode incluir companyDomain, a URL completa em companyUrl e domainConfidence. O campo creditsUsed indica o custo desse input, enquanto error explica uma falha ao nível do item.

A função faz três tipos de chamadas à API, por esta ordem.

Primeiro, envia a pesquisa. O payload contém as definições e os registos das empresas. O POST devolve 202 Accepted, e @id é o ID da tarefa necessário para as chamadas seguintes:

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

Se DATABLIST_API_KEY não estiver definida, o Python gera um KeyError ao procurar a variável de ambiente, antes de fazer qualquer requisição HTTP. Exporte a variável e volte a executar o script. Se a variável contiver uma chave inválida, o POST chega à API e devolve o código HTTP 401. Copie uma Personal API Key atual da página da sua conta e substitua o valor inválido.

Em segundo lugar, consulta o estado da tarefa. Como o POST inicia um processo em segundo plano, o script verifica a tarefa a cada três segundos até esta chegar a completed, failed ou 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)

Se a tarefa falhar por falta de créditos, o motivo indicado no erro será credits_error. Adicione créditos ou aguarde a renovação dos créditos mensais, verifique os resultados já devolvidos e envie apenas os inputs por concluir numa nova tarefa.

Em terceiro lugar, obtém todas as páginas de resultados. Cada resposta acrescenta resultados à lista e fornece o cursor para a requisição seguinte:

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

O script continua a obter páginas até nextCursor ser null.

Antes do POST, o script apresenta o uniqueId gerado. Se a ligação cair antes de receber a resposta de criação, liste primeiro as suas tarefas e verifique o respetivo estado antes de criar outra:

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"

Encontre a tarefa cujo uniqueId corresponde ao valor gerado pelo script e copie o respetivo @id. O endpoint de listagem permite filtrar por enrichmentId, mas não por uniqueId. Utilize o @id com as mesmas URLs de estado e resultados utilizadas pelo script. Se a tarefa ainda estiver ativa, continue a consultar o estado. Se tiver parado ou falhado, verifique se existem resultados parciais antes de voltar a enviar os inputs.

Entenda a correspondência e o fallback por país

O script utiliza dataset_google, a opção predefinida recomendada. Primeiro, a Datablist consulta os seus datasets de empresas. Se não encontrar uma correspondência ou detetar informações contraditórias, pode recorrer ao Google como fallback. Antes de devolver um nível de confiança, verifica as correspondências possíveis.

O país ajuda a distinguir nomes ambíguos. Neste exemplo, o mesmo nome de empresa gerou resultados diferentes:

• Mercury com US devolveu shipmercury.com e https://www.shipmercury.com/.
• Mercury com IE devolveu mercuryeng.com e http://www.mercuryeng.com.

Ambos apresentaram um nível de confiança High. O país ajuda a pesquisa a escolher entre empresas com o mesmo nome, enquanto o nível de confiança permite decidir que resultados precisam de revisão.

Quando a correspondência também depender do setor, da dimensão da empresa ou de outro contexto de negócio, utilize o workflow de pesquisa do AI Agent. Inclua esses dados no prompt do AI Agent para que este pesquise a empresa pretendida com mais contexto.

Preencha apenas os sites em falta nos seus registos

Agora que find_domains() consegue encontrar um site a partir do nome da empresa, reutilize a função para preencher apenas os sites em falta nos seus registos. Os registos que já têm um site não passam pela API. Assim, os valores permanecem inalterados e não consomem créditos de pesquisa.

No ficheiro find_company_websites.py, substitua todo o conteúdo desde companies = [ até ao ciclo de impressão final pelo bloco abaixo. Mantenha a atribuição de api_key acima. A verificação de campos vazios abrange None, uma string vazia e espaços em branco. Se todos os sites estiverem preenchidos, não é feita qualquer chamada à 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))

O código indexa os resultados por inputId e, em seguida, percorre os registos originais pela ordem inicial. O objeto website_lookup mantém o resultado e os dados necessários para revisão junto de cada registo processado, incluindo resultados sem sucesso ou em falta.

Para crm-001, website_is_missing() devolve False, porque a URL da Microsoft já existe. O registo não é adicionado a records_to_lookup, pelo que nunca é enviado à API. A combinação preserva o site e regista o estado como not_requested.

Para crm-002, o site está vazio, por isso a requisição envia a padaria com inputId: crm-002. A API devolve o mesmo inputId. O dicionário results_by_id utiliza-o para encontrar o resultado correto, e a combinação grava https://boulangerie-du-palais-lyon.eatbu.com nesse registo sem depender da ordem dos resultados.

A padaria utiliza um subdomínio específico, boulangerie-du-palais-lyon.eatbu.com. Preserve esse valor em vez de o reduzir a eatbu.com e guarde o endereço exato devolvido em companyUrl. A definição acceptNonRoot: true também permite páginas empresariais alojadas cujo endereço inclui um caminho após o hostname. Por isso, a URL completa é importante nos dois casos.

Uma tarefa concluída pode ainda conter itens com o estado no_result, error ou skipped. O exemplo só altera website quando o estado é success e existe um companyUrl. Em caso de no_result, o site continua vazio e o estado permanece visível em website_lookup. Verifique primeiro o nome e o país da empresa. Se algum dos inputs estiver incorreto, corrija-o e volte a enviar esse registo numa nova tarefa. Caso contrário, mantenha o site vazio e assinale-o para revisão manual.

Calcule os créditos antes de escalar o script

As cinco pesquisas do exemplo devolveram success com nível de confiança High e consumiram um total de 5 créditos.

Uma pesquisa resolvida pelo dataset custa 1 crédito. Uma pesquisa processada com o Google custa 2,5 créditos. Paga apenas pelo percurso utilizado, não pelos dois. Para 1.000 inputs, isso representa 1.000 créditos se todos forem resolvidos pelo dataset ou 2.500 créditos se todos utilizarem o Google. Estes valores servem para planeamento e não garantem que todas as pesquisas terão sucesso.

Mova o seletor para comparar o consumo de créditos da sua lista.

Com um preço de 20 dólares por 20.000 créditos, esses cenários custam 1 dólar e 2,50 dólares. A subscrição Growth é cobrada à parte e inclui atualmente 20.000 créditos mensais, utilizados antes dos carregamentos adicionais. Consulte os preços atuais, pois os valores dos planos e pacotes podem mudar.

Uma tarefa aceita até 10.000 inputs, com um limite de 16 KiB por input e 10 MiB por requisição. Este exemplo utiliza chamadas bloqueantes do Requests para enviar, acompanhar e obter os resultados de uma pequena tarefa assíncrona da Datablist. O repositório mantido pela equipa também inclui exemplos de processamento de ficheiros CSV em lotes e clientes não bloqueantes.

Próximos passos

Depois de confirmar que o site devolvido pertence à empresa pretendida, pode extrair o Impressum com Python para adicionar os dados legais da empresa e a URL da página legal de origem ao mesmo registo de fornecedor. Mantenha a revisão da correspondência do site separada da extração posterior.

Substitua os registos de exemplo por uma pequena amostra dos seus próprios dados. Depois, execute o script e analise os registos combinados. Preserve IDs estáveis e o país, e só aumente a escala depois de confirmar que o script mantém os sites conhecidos e preenche os campos vazios esperados. Pode começar pelos exemplos atualizados da API Company Name to Domain no GitHub se quiser descarregar o script ou consultar versões com processamento de CSV em lotes e execução assíncrona.