Trovare siti web aziendali con Python e API

Per trovare il sito web di un’azienda partendo dal suo nome in Python, invia il nome e, se disponibile, il Paese di riferimento alla Company Name to Domain API di Datablist. L’API elabora la ricerca in background, quindi restituisce il dominio e l’URL completo del sito.

Per prima cosa ti mostrerò una semplice richiesta API in Python. Poi riutilizzerò la stessa funzione per compilare solo i siti web mancanti in un elenco di aziende, senza sovrascrivere gli URL già presenti.

Se preferisci un’interfaccia simile a un foglio di calcolo, consulta la guida per trovare in blocco i siti web dai nomi delle aziende.

💡 Riepilogo per chi va di fretta

Crea una funzione Python che trovi i siti web dai nomi delle aziende, quindi usala per completare i record con il sito vuoto senza sostituire gli URL già noti.

• Input: nomi delle aziende con Paese facoltativo.
• Output: dominio, URL completo del sito, livello di confidenza, stato del risultato e crediti utilizzati.

Configurare Python e la chiave API Datablist

Per questo esempio uso il package Requests per effettuare chiamate HTTP sincrone. Installalo in un ambiente virtuale:

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

Ti serve anche un piano Datablist Growth, perché l’accesso alla REST API è una funzionalità Growth. Crea una Personal API Key dalla pagina API del tuo account Datablist, quindi salvala in una variabile d’ambiente:

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

🔑 Hai già una chiave API? Le chiavi meno recenti usano il formato auth0|<userId>:<accountId>:<secret>:1 e sono ancora supportate. Continua a usare la chiave esistente esattamente come è stata generata: non devi sostituirla né modificarne il formato.

Sostituisci il valore di esempio con la chiave generata nella pagina del tuo account. Le ricerche dei siti aziendali consumano crediti. Prima di estendere lo script spiegherò i percorsi da 1 e 2,5 crediti; i valori aggiornati dei piani sono disponibili nella pagina dei prezzi di Datablist.

Trovare il sito di un’azienda con Python

Crea un file chiamato find_company_websites.py. Lo script invia i nomi, attende il completamento del task e recupera i risultati tramite un’unica funzione. Ogni input usa l’ID stabile del tuo record come inputId, così puoi associare ogni risultato al record corretto anche se l’ordine della risposta cambia.

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

Eseguilo dal terminale in cui hai esportato la chiave:

1python find_company_websites.py

Ecco il risultato per Microsoft:

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

Usa inputId per associare il risultato al record originale. status può essere success, no_result, error o skipped. Un risultato positivo può includere companyDomain, l’URL completo companyUrl e domainConfidence. creditsUsed indica il costo di questo input, mentre error descrive un errore relativo al singolo elemento.

La funzione effettua tre tipi di chiamate API, in quest’ordine.

Per prima cosa invia la ricerca. Il payload contiene le impostazioni e i record aziendali. La richiesta POST restituisce 202 Accepted, mentre @id è l’ID del task necessario per le chiamate successive:

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 non è impostata, Python genera un KeyError durante la lettura della variabile d’ambiente, prima ancora di effettuare una richiesta HTTP. Esporta la variabile e avvia nuovamente lo script. Se la variabile contiene una chiave non valida, la richiesta POST raggiunge l’API e restituisce l’errore HTTP 401. Copia una Personal API Key valida dalla pagina del tuo account e sostituisci il valore errato.

Poi verifica periodicamente il task. La richiesta POST avvia l’elaborazione in background, quindi lo script legge lo stato del task ogni tre secondi finché non diventa 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)

Se il task non riesce per mancanza di crediti, l’errore riporta il motivo credits_error. Aggiungi crediti oppure attendi il rinnovo di quelli mensili, controlla gli eventuali risultati già restituiti e invia in un nuovo task solo gli input non completati.

Infine recupera tutte le pagine dei risultati. Ogni risposta aggiunge i risultati all’elenco e fornisce il cursore per la richiesta successiva:

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

Lo script recupera la pagina successiva finché nextCursor non è null.

Prima della richiesta POST, lo script stampa il valore uniqueId generato. Se la connessione si interrompe prima che tu riceva la risposta di creazione, controlla innanzitutto l’elenco dei task e il loro stato, evitando di crearne subito un altro:

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"

Trova il task il cui uniqueId corrisponde al valore generato dallo script, quindi copia il relativo @id. L’endpoint dell’elenco consente di filtrare per enrichmentId, ma non per uniqueId. Usa l’@id con gli stessi URL di stato e risultati utilizzati dallo script. Se il task è ancora attivo, continua a verificarne lo stato. Se si è interrotto o non è riuscito, controlla i risultati parziali prima di inviare nuovamente gli input.

Matching dei dati e fallback basato sul Paese

Lo script usa dataset_google, l’opzione predefinita consigliata. Datablist controlla innanzitutto i propri dataset aziendali. Se il match non è disponibile o presenta incongruenze, può usare Google come fallback. Prima di restituire un livello di confidenza, verifica i possibili risultati.

Il Paese aiuta a distinguere i nomi ambigui. In questo esempio, lo stesso nome aziendale ha prodotto risultati diversi:

• Mercury con US ha restituito shipmercury.com e https://www.shipmercury.com/.
• Mercury con IE ha restituito mercuryeng.com e http://www.mercuryeng.com.

Entrambi avevano un livello di confidenza High. Il Paese aiuta la ricerca a scegliere tra aziende con lo stesso nome, mentre il livello di confidenza ti permette di stabilire quali risultati richiedono una verifica.

Quando il match dipende anche dal settore, dalle dimensioni dell’azienda o da altri dati di business, usa il workflow di ricerca AI Agent. Inserisci queste informazioni nel prompt di AI Agent, così potrà individuare l’azienda corretta usando un contesto di business più completo.

Compilare solo i siti web mancanti nei record

Ora che find_domains() può trovare un sito web dal nome dell’azienda, riutilizzala per compilare solo i siti mancanti nei tuoi record aziendali. I record che contengono già un sito web non vengono inviati all’API: i valori restano invariati e non consumano crediti.

Nel file find_company_websites.py, sostituisci tutto il codice da companies = [ fino al ciclo di stampa finale con il blocco seguente. Mantieni l’assegnazione di api_key che si trova più in alto. Il controllo dei campi vuoti gestisce None, una stringa vuota e una stringa composta da soli spazi. Se tutti i siti sono già presenti, non viene effettuata alcuna chiamata 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))

Il codice indicizza i risultati per inputId, quindi scorre i record originali nello stesso ordine. L’oggetto website_lookup conserva l’esito e i dati necessari alla verifica accanto a ogni record elaborato, inclusi i risultati mancanti o non riusciti.

Per crm-001, website_is_missing() restituisce False perché l’URL di Microsoft è già presente. Il record non viene aggiunto a records_to_lookup, quindi l’API non lo riceve. Il merge mantiene il sito web esistente e registra lo stato not_requested.

Per crm-002, il sito web è vuoto, quindi la richiesta invia la panetteria con inputId: crm-002. L’API restituisce lo stesso inputId. results_by_id lo usa per individuare il risultato corretto e il merge scrive https://boulangerie-du-palais-lyon.eatbu.com nel record, senza dipendere dall’ordine dei risultati.

La panetteria usa un sottodominio specifico per l’attività, boulangerie-du-palais-lyon.eatbu.com. Conserva questo valore senza ridurlo a eatbu.com e salva l’indirizzo esatto restituito in companyUrl. L’impostazione acceptNonRoot: true consente anche di accettare una pagina aziendale ospitata il cui indirizzo contiene un percorso dopo l’hostname: in entrambi i casi, quindi, è importante conservare l’URL completo.

Un task completato può comunque contenere elementi no_result, error o skipped. L’esempio modifica website solo quando lo stato è success e companyUrl è presente. In caso di no_result, il sito rimane vuoto e lo stato resta visibile in website_lookup. Controlla innanzitutto il nome dell’azienda e il Paese. Se uno dei due input è errato, correggilo e invia nuovamente quel record in un altro task. In caso contrario, lascia visibile il sito vuoto per una verifica manuale.

Stimare i crediti prima di estendere lo script

Le cinque ricerche dell’esempio hanno restituito success con confidenza High e utilizzato in totale 5 crediti.

Una ricerca risolta tramite il dataset costa 1 credito. Una ricerca elaborata con Google costa 2,5 crediti. Paghi solo il percorso utilizzato, non entrambi. Per 1.000 input, il costo è quindi di 1.000 crediti se tutti vengono risolti tramite il dataset oppure di 2.500 crediti se tutti usano Google. Si tratta di scenari utili alla pianificazione, non della garanzia che ogni ricerca produca un risultato.

Sposta il cursore per confrontare il consumo di crediti del tuo elenco.

Con un prezzo di 20 $ per 20.000 crediti, questi scenari costano rispettivamente 1 $ e 2,50 $. L’abbonamento Growth è separato e attualmente include 20.000 crediti mensili, che vengono utilizzati prima dei crediti aggiuntivi. Consulta i prezzi aggiornati, perché i valori dei piani e dei pacchetti possono cambiare.

Un task accetta fino a 10.000 input, con un limite di 16 KiB per input e di 10 MiB per richiesta. Questo esempio usa chiamate Requests bloccanti per inviare, monitorare e recuperare i risultati di un piccolo task asincrono di Datablist. Il repository ufficiale include anche esempi di elaborazione in batch dei file CSV e client non bloccanti.

Prossimi passaggi

Dopo aver verificato che il sito restituito appartenga all’azienda corretta, puoi estrarre l’Impressum con Python per aggiungere allo stesso record del fornitore i dati legali dell’azienda e l’URL della pagina legale di origine. Mantieni separata la verifica del match del sito dalla successiva estrazione dei dati.

Sostituisci i record di esempio con un piccolo campione dei tuoi dati, quindi esegui lo script e controlla i record ottenuti dopo il merge. Mantieni ID stabili e il Paese di riferimento; aumenta i volumi solo dopo aver verificato che lo script conservi i siti già noti e compili correttamente quelli mancanti. Se vuoi scaricare lo script oppure usare l’elaborazione in batch dei CSV o una versione asincrona, puoi partire dagli esempi ufficiali della Company Name to Domain API su GitHub.