Jak znaleźć stronę firmy po nazwie w Pythonie przez API

Aby znaleźć stronę internetową firmy na podstawie jej nazwy w Pythonie, wyślij nazwę oraz dostępne informacje o kraju do Company Name to Domain API Datablist. API przetworzy wyszukiwanie w tle, a następnie zwróci domenę i pełny adres URL strony.

Najpierw pokażę proste zapytanie API w Pythonie. Następnie użyję tej samej funkcji, aby uzupełnić wyłącznie brakujące strony internetowe na liście firm, zachowując adresy URL, które już znajdują się w rekordach.

Jeśli wolisz interfejs przypominający arkusz kalkulacyjny, skorzystaj z poradnika zbiorczego wyszukiwania stron firm na podstawie nazw.

💡 Podsumowanie dla zabieganych

Utwórz jedną funkcję w Pythonie, która wyszukuje strony na podstawie nazw firm, a następnie wykorzystaj ją do uzupełniania rekordów z pustym polem strony bez zastępowania znanych adresów URL.

• Dane wejściowe: nazwy firm z opcjonalną informacją o kraju.
• Dane wyjściowe: domena, pełny adres URL strony, poziom pewności, status wyniku i liczba wykorzystanych kredytów.

Skonfiguruj Pythona i klucz API Datablist

W tym przykładzie używam pakietu Requests do wykonywania synchronicznych wywołań HTTP. Zainstaluj go w środowisku wirtualnym:

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

Potrzebujesz również planu Datablist Growth, ponieważ dostęp do REST API jest funkcją tego planu. Utwórz Personal API Key na stronie API swojego konta Datablist, a następnie zapisz go w zmiennej środowiskowej:

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

🔑 Masz już klucz API? Starsze klucze mają format auth0|<userId>:<accountId>:<secret>:1 i nadal są obsługiwane. Korzystaj z dotychczasowego klucza dokładnie w takiej postaci, w jakiej został wygenerowany — nie musisz go wymieniać ani zmieniać jego formatu.

Zastąp powyższą przykładową wartość kluczem wygenerowanym na stronie konta. Wyszukiwanie stron firm zużywa kredyty. Przed przejściem do większej skali wyjaśnię warianty kosztujące 1 oraz 2,5 kredytu. Aktualne wartości dla poszczególnych planów znajdziesz na stronie cennika Datablist.

Wyszukaj stronę jednej firmy w Pythonie

Utwórz plik o nazwie find_company_websites.py. Skrypt przesyła nazwy, czeka na zakończenie zadania i pobiera wyniki za pomocą jednej funkcji. Każdy element wejściowy wykorzystuje stały identyfikator rekordu jako inputId, dzięki czemu można przypisać każdy wynik do właściwego rekordu, nawet jeśli zmieni się kolejność odpowiedzi.

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

Uruchom skrypt w terminalu, w którym wcześniej ustawiono zmienną z kluczem:

1python find_company_websites.py

Oto wynik dla firmy Microsoft:

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

Użyj inputId, aby przypisać wynik do pierwotnego rekordu. Pole status może przyjmować wartość success, no_result, error albo skipped. Pomyślny wynik może zawierać companyDomain, pełny adres companyUrl oraz domainConfidence. Pole creditsUsed pokazuje koszt danego elementu, a error opisuje błąd dotyczący konkretnego rekordu.

Funkcja wykonuje kolejno trzy rodzaje wywołań API.

Najpierw przesyła zapytanie. Payload zawiera ustawienia oraz rekordy firm. Żądanie POST zwraca 202 Accepted, a @id jest identyfikatorem zadania potrzebnym w kolejnych wywołaniach:

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

Jeśli zmienna DATABLIST_API_KEY nie jest ustawiona, Python zgłosi KeyError podczas odczytu środowiska, jeszcze przed wykonaniem żądania HTTP. Ustaw zmienną i ponownie uruchom skrypt. Jeśli zawiera ona nieprawidłowy klucz, żądanie POST dotrze do API, które zwróci HTTP 401. Skopiuj aktualny Personal API Key ze strony konta i zastąp nim błędną wartość.

Następnie sprawdza status zadania. Żądanie POST uruchamia pracę w tle, dlatego skrypt odczytuje stan zadania co trzy sekundy, aż pojawi się status completed, failed albo 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)

Jeśli zadanie nie powiedzie się z powodu braku kredytów, przyczyną błędu będzie credits_error. Dokup kredyty lub poczekaj na odnowienie miesięcznego limitu, sprawdź wszystkie zwrócone już wyniki, a następnie prześlij w nowym zadaniu wyłącznie nieukończone elementy.

Na końcu pobiera wszystkie strony wyników. Każda odpowiedź dodaje wyniki do listy i przekazuje kursor potrzebny do kolejnego zapytania:

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

Skrypt pobiera kolejne strony, dopóki nextCursor nie przyjmie wartości null.

Przed wykonaniem żądania POST skrypt wyświetla wygenerowany uniqueId. Jeśli połączenie zostanie przerwane przed otrzymaniem odpowiedzi o utworzeniu zadania, najpierw wyświetl listę zadań i sprawdź ich status, zanim utworzysz kolejne:

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"

Znajdź zadanie, którego uniqueId odpowiada wartości wygenerowanej przez skrypt, a następnie skopiuj jego @id. Endpoint listy pozwala filtrować po enrichmentId, ale nie po uniqueId. Użyj @id z tymi samymi adresami URL statusu i wyników, z których korzysta skrypt. Jeśli zadanie jest nadal aktywne, dalej sprawdzaj jego status. Jeśli zostało zatrzymane lub zakończyło się błędem, przed ponownym przesłaniem elementów sprawdź wyniki częściowe.

Jak działa dopasowanie i fallback uwzględniający kraj

Skrypt korzysta z dataset_google, czyli zalecanego ustawienia domyślnego. Datablist najpierw sprawdza własne zbiory danych firm. Jeśli brakuje dopasowania lub wyniki są sprzeczne, może użyć Google jako fallbacku. Przed zwróceniem poziomu pewności weryfikuje potencjalne dopasowania.

Informacja o kraju pomaga przy niejednoznacznych nazwach. W tym przykładzie ta sama nazwa firmy dała różne wyniki:

• Mercury z krajem US zwróciło shipmercury.com oraz https://www.shipmercury.com/.
• Mercury z krajem IE zwróciło mercuryeng.com oraz http://www.mercuryeng.com.

Oba wyniki miały poziom pewności High. Informacja o kraju pomaga wybrać właściwą firmę spośród podmiotów o tej samej nazwie, natomiast poziom pewności ułatwia określenie, które wyniki wymagają weryfikacji.

Jeśli dopasowanie zależy również od branży, wielkości firmy lub innego kontekstu biznesowego, skorzystaj z workflow researchu z AI Agent. Dodaj te informacje do promptu AI Agent, aby mógł wyszukać właściwą firmę na podstawie bogatszego kontekstu biznesowego.

Uzupełnij tylko brakujące strony w rekordach

Skoro funkcja find_domains() potrafi znaleźć stronę na podstawie nazwy firmy, wykorzystaj ją ponownie do uzupełnienia wyłącznie brakujących adresów w rekordach. Rekordy, które już zawierają stronę, omijają API, dzięki czemu ich wartości pozostają bez zmian i nie zużywają kredytów.

W pliku find_company_websites.py zastąp wszystko od companies = [ aż po końcową pętlę wyświetlającą wyniki poniższym blokiem. Pozostaw znajdujące się wyżej przypisanie api_key. Kontrola pustej wartości obsługuje None, pusty ciąg znaków oraz same białe znaki. Jeśli wszystkie strony są uzupełnione, skrypt nie wykonuje żadnego wywołania 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))

Kod indeksuje wyniki według inputId, a następnie przechodzi kolejno przez pierwotne rekordy. Obiekt website_lookup przechowuje rezultat i informacje potrzebne do weryfikacji obok każdego przetwarzanego rekordu, również w przypadku niepowodzenia lub braku wyniku.

Dla crm-001 funkcja website_is_missing() zwraca False, ponieważ adres URL Microsoftu jest już zapisany. Rekord nie trafia do records_to_lookup, więc API nigdy go nie otrzymuje. Podczas scalania strona pozostaje bez zmian, a zapisany status to not_requested.

Dla crm-002 pole strony jest puste, dlatego zapytanie wysyła nazwę piekarni z inputId: crm-002. API zwraca ten sam inputId. Obiekt results_by_id używa go do znalezienia właściwego wyniku, a podczas scalania do rekordu trafia https://boulangerie-du-palais-lyon.eatbu.com — niezależnie od kolejności wyników.

Piekarnia korzysta z subdomeny przypisanej do konkretnej firmy: boulangerie-du-palais-lyon.eatbu.com. Zachowaj tę wartość zamiast skracać ją do eatbu.com i zapisz dokładny adres z pola companyUrl. Ustawienie acceptNonRoot: true pozwala również zwrócić hostowaną stronę firmy, której adres zawiera ścieżkę po nazwie hosta, dlatego w obu przypadkach istotny jest pełny URL.

Ukończone zadanie nadal może zawierać elementy ze statusem no_result, error albo skipped. Przykład zmienia pole website tylko wtedy, gdy status to success i dostępne jest companyUrl. Przy statusie no_result strona pozostaje pusta, a status nadal jest widoczny w website_lookup. Najpierw sprawdź nazwę firmy i kraj. Jeśli któreś dane są błędne, popraw je i prześlij rekord w nowym zadaniu. W przeciwnym razie pozostaw puste pole strony do ręcznej weryfikacji.

Oszacuj zużycie kredytów przed skalowaniem

Wszystkie pięć przykładowych wyszukiwań zwróciło success z poziomem pewności High i zużyło łącznie 5 kredytów.

Wyszukiwanie rozstrzygnięte na podstawie zbioru danych kosztuje 1 kredyt. Wyszukiwanie przetworzone przez Google kosztuje 2,5 kredytu. Płacisz tylko za wykorzystaną ścieżkę, a nie za obie. Dla 1000 elementów oznacza to 1000 kredytów, jeśli wszystkie zostaną znalezione w zbiorze danych, albo 2500 kredytów, jeśli wszystkie zostaną przetworzone przez Google. Są to scenariusze przydatne do planowania, a nie gwarancja powodzenia każdego wyszukiwania.

Przesuń suwak, aby porównać zużycie kredytów dla swojej listy.

Przy cenie 20 USD za 20 000 kredytów te scenariusze kosztują odpowiednio 1 USD oraz 2,50 USD. Subskrypcja Growth jest rozliczana osobno i obecnie obejmuje 20 000 kredytów miesięcznie, które są wykorzystywane przed dodatkowymi pakietami. Sprawdź aktualny cennik, ponieważ wartości planów i pakietów mogą się zmieniać.

Jedno zadanie może zawierać maksymalnie 10 000 elementów, przy limicie 16 KiB na element i 10 MiB na całe żądanie. W tym przykładzie blokujące wywołania Requests służą do przesłania zadania, sprawdzania jego statusu i pobrania wyników z jednego niewielkiego, asynchronicznego zadania Datablist. Utrzymywane repozytorium zawiera również przykłady przetwarzania CSV w partiach oraz nieblokującego klienta.

Co dalej

Gdy potwierdzisz, że zwrócona strona należy do właściwej firmy, możesz pobrać dane z Impressum za pomocą Pythona, aby dodać dane prawne firmy oraz adres źródłowej strony prawnej do tego samego rekordu dostawcy. Weryfikację dopasowania strony przeprowadź niezależnie od późniejszej ekstrakcji.

Zastąp przykładowe rekordy niewielką próbką własnych danych, uruchom skrypt i sprawdź scalone rekordy. Zachowaj stałe identyfikatory oraz informacje o kraju. Zwiększaj skalę dopiero wtedy, gdy upewnisz się, że skrypt zachowuje znane strony i prawidłowo uzupełnia oczekiwane braki. Jeśli potrzebujesz gotowego skryptu do pobrania, przetwarzania CSV w partiach lub wersji asynchronicznej, zacznij od utrzymywanych przykładów Company Name to Domain API na GitHubie.