Comment scraper les résultats de recherche Google en Python

Tout projet de scraping qui touche à Google commence de la même façon : requests.get("https://www.google.com/search?q=..."), un peu de BeautifulSoup, et un parser qui fonctionne à merveille pendant environ deux jours.

Puis ça s’arrête. Voici pourquoi ça s’arrête, et à quoi ressemble réellement une configuration qui continue de fonctionner.

Pourquoi la version fait maison casse

Quatre choses distinctes la cassent, généralement dans cet ordre :

1. Blocage IP. Les IP de datacenter sont rapidement limitées en débit, puis reçoivent une page de consentement ou un CAPTCHA au lieu de résultats. Faire tourner une poignée de proxies retarde cela, ça ne le résout pas.
2. Balisage instable. Le balisage des résultats de Google porte des noms de classes obfusqués et changeants. Un sélecteur CSS qui correspond aujourd’hui n’est pas un contrat stable.
3. Types de résultats. Les liens organiques sont la partie facile. Les extraits optimisés, les « Autres questions posées », les panneaux de connaissances, les publicités et les blocs locaux ont chacun leur propre structure, et leur présence varie selon la requête et le pays.
4. Le calcul de la pagination. Google n’honore plus le paramètre num=100, donc « les 100 premiers résultats » signifie dix requêtes paginées séparées, pas une seule. Toute estimation de coût écrite avant ce changement est fausse d’un facteur 10.

Aucun de ces problèmes n’est difficile pris individuellement. Ensemble, c’est un tapis roulant de maintenance : vous finissez par posséder un pool de proxies et un parser qui a besoin d’attention chaque trimestre, pour obtenir des données qui ne sont pas votre produit.

La version qui continue de fonctionner

Envoyez la requête à une API qui possède les proxies et le parsing, et récupérez du JSON en retour :

import requests

API_KEY = "faf_live_..."

def search(query, country="us", 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"])

Pas de configuration de proxy, pas de sélecteurs. La réponse porte aussi people_also_ask, knowledge_graph, ads et related_searches quand la requête les produit, donc vous vérifiez la présence d’une clé au lieu d’écrire un parser par type de résultat.

Ciblage par pays et par langue

Les données de classement n’ont aucun sens sans localisation. Passez le pays explicitement plutôt que d’espérer que la requête atterrisse quelque part d’utile :

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

La requête est émise depuis cette localisation, avec le domaine Google et les valeurs de langue par défaut correspondants.

Collecter plus d’une page

Comme les résultats arrivent par dix par page, traitez la profondeur comme une boucle et souvenez-vous que chaque page est une requête facturable séparée :

def search_deep(query, pages=3, country="us"):
    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), "results")

Bien gérer les erreurs

Trois codes de statut comptent en production, et les traiter tous comme un « retry » est la façon dont vous brûlez un solde de crédits en un après-midi :

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

        # Plus de crédits : réessayer n'aidera jamais.
        if response.status_code == 402:
            raise RuntimeError("Out of credits, top up before retrying")

        # Limite de concurrence : attendez exactement le temps que l'API demande.
        if response.status_code == 429:
            time.sleep(int(response.headers.get("Retry-After", 1)))
            continue

        # Échec en amont : non facturé, sûr de réessayer avec un backoff.
        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("Exhausted retries")

La règle empirique : 402 est terminal, 429 signifie ralentir exactement du temps indiqué dans Retry-After, et 5xx est réessayable car les requêtes échouées ne sont pas facturées.

Exécuter des requêtes en parallèle

La recherche de mots-clés implique des centaines de requêtes, et la limite qui compte est la concurrence, pas le volume total. Bornez-la explicitement :

from concurrent.futures import ThreadPoolExecutor

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

# Restez à ou en dessous de la concurrence autorisée par votre plan.
with ThreadPoolExecutor(max_workers=5) as pool:
    for keyword, data in zip(keywords, pool.map(search_safe, keywords)):
        print(keyword, len(data["organic_results"]))

Dépasser la concurrence de votre plan ne rend pas le travail plus rapide, cela ne fait que convertir des requêtes en 429.

Ce que ça coûte

Une recherche coûte 5 crédits, et chaque réponse indique ce qu’elle a dépensé dans l’en-tête X-FastAF-Credits-Used, pour que vous puissiez mesurer par fonctionnalité plutôt que deviner après coup. Voir les tarifs pour les chiffres par pack, ou la référence de l’API Google SERP pour la liste complète des paramètres et réponses.