Come estrarre i risultati di Google in Python

Ogni progetto che tocca Google comincia allo stesso modo: requests.get("https://www.google.com/search?q=..."), un po’ di BeautifulSoup e un parser che funziona benissimo per circa due giorni.

Poi smette. Ecco perché smette, e come si imposta invece qualcosa che continua a funzionare.

Perché la versione fai-da-te si rompe

Si rompe per quattro motivi distinti, di solito in quest’ordine:

1. Blocco degli IP. Gli IP da datacenter vengono limitati in fretta, e al posto dei risultati arriva una pagina di consenso o un CAPTCHA. Ruotare qualche proxy rimanda il problema, non lo risolve.
2. Markup che cambia. Google usa nomi di classi offuscati e rotanti. Un selettore CSS che funziona oggi non è un contratto stabile.
3. Tipi di risultato. I link organici sono la parte facile. Featured snippet, People Also Ask, knowledge panel, annunci e local pack hanno ognuno una struttura propria, e la loro presenza cambia per query e per paese.
4. Conto della paginazione. Google non onora più il parametro num=100: “i primi 100 risultati” significa dieci richieste paginate, non una. Qualsiasi stima di costo scritta prima di quel cambiamento sbaglia di dieci volte.

Nessuno di questi problemi è difficile preso da solo. Insieme diventano manutenzione perpetua: ti ritrovi a gestire un pool di proxy e un parser da sistemare ogni trimestre, per ottenere dati che non sono il tuo prodotto.

La versione che regge

Mandi la query a un’API che possiede i proxy e il parsing, e ricevi JSON:

import requests

API_KEY = "faf_live_..."

def search(query, country="it", page=0):
    response = requests.get(
        "https://api.fastaf.co/v1/serp",
        params={"query": query, "country": country, "page": page},
        headers={"Authorization": f"Bearer {API_KEY}"},
        timeout=60,
    )
    response.raise_for_status()
    return response.json()

data = search("web scraping api")
for result in data["organic_results"]:
    print(result["rank"], result["title"], result["link"])

Niente configurazione dei proxy, niente selettori. La risposta contiene anche people_also_ask, knowledge_graph, ads e related_searches quando la query li produce, quindi controlli la presenza di una chiave invece di scrivere un parser per ogni tipo di risultato.

Targeting per paese e lingua

I dati di posizionamento non significano niente senza una località. Passa il paese in modo esplicito:

for country in ["it", "us", "gb", "de"]:
    data = search("software gestione progetti", country=country)
    top = data["organic_results"][0]
    print(country, "->", top["link"])

La query viene emessa da quella località, con il dominio Google e i default di lingua corrispondenti.

Raccogliere più di una pagina

Dato che i risultati arrivano dieci per pagina, tratta la profondità come un ciclo e ricorda che ogni pagina è una richiesta a pagamento:

def search_deep(query, pages=3, country="it"):
    results = []
    for page in range(pages):
        data = search(query, country=country, page=page)
        results.extend(data.get("organic_results", []))
    return results

top_30 = search_deep("headless browser", pages=3)
print(len(top_30), "risultati")

Gestire gli errori come si deve

In produzione contano tre status code, e trattarli tutti come “riprova” è il modo più rapido di bruciare il saldo crediti in un pomeriggio:

import time

def search_safe(query, **kwargs):
    for attempt in range(3):
        response = requests.get(
            "https://api.fastaf.co/v1/serp",
            params={"query": query, **kwargs},
            headers={"Authorization": f"Bearer {API_KEY}"},
            timeout=60,
        )

        if response.status_code == 200:
            return response.json()

        # Crediti finiti: riprovare non servirà mai a niente.
        if response.status_code == 402:
            raise RuntimeError("Crediti esauriti")

        # Limite di concorrenza: aspetta esattamente quanto chiede l'API.
        if response.status_code == 429:
            time.sleep(int(response.headers.get("Retry-After", 1)))
            continue

        # Errore a monte: non viene addebitato, si può riprovare.
        if response.status_code in (502, 503, 504):
            time.sleep(2 ** attempt)
            continue

        error = response.json().get("error", {})
        raise RuntimeError(f"{error.get('code')}: {error.get('message')}")

    raise RuntimeError("Tentativi esauriti")

La regola: 402 è definitivo, 429 significa rallentare esattamente di quanto indica Retry-After, e 5xx si può riprovare perché le richieste fallite non vengono addebitate.

Query in parallelo

La keyword research significa centinaia di query, e il limite che conta è la concorrenza, non il volume totale:

from concurrent.futures import ThreadPoolExecutor

keywords = ["serp api", "web scraping api", "rotazione proxy", "headless browser"]

# Tienilo pari o inferiore alla concorrenza consentita dal tuo piano.
with ThreadPoolExecutor(max_workers=5) as pool:
    for keyword, data in zip(keywords, pool.map(search_safe, keywords)):
        print(keyword, len(data["organic_results"]))

Superare la concorrenza del piano non rende il lavoro più veloce: trasforma le richieste in 429.

Quanto costa

Una ricerca costa 5 crediti, e ogni risposta dichiara quanto ha speso nell’header X-FastAF-Credits-Used, così puoi misurare per funzionalità invece di scoprirlo a cose fatte. I numeri per pacchetto sono nella pagina prezzi.