Cómo extraer resultados de búsqueda de Google con Python

Todo proyecto de scraping que toca Google empieza igual: requests.get("https://www.google.com/search?q=..."), un poco de BeautifulSoup, y un parser que funciona de maravilla durante unos dos días.

Y entonces deja de funcionar. Esto es por qué deja de funcionar, y cómo es en realidad una configuración que sigue funcionando.

Por qué se rompe la versión DIY

Cuatro cosas distintas la rompen, normalmente en este orden:

1. Bloqueo por IP. Las IPs de datacenter reciben rate limit rápido, y luego se les sirve una página de consentimiento o un CAPTCHA en lugar de resultados. Rotar un puñado de proxies retrasa esto, no lo soluciona.
2. Cambios de markup. El markup de resultados de Google lleva nombres de clase ofuscados y rotativos. Un selector CSS que coincide hoy no es un contrato estable.
3. Tipos de resultado. Los enlaces orgánicos son la parte fácil. Los featured snippets, People Also Ask, knowledge panels, anuncios y local packs tienen cada uno su propia estructura, y su presencia varía según la consulta y el país.
4. Matemática de paginación. Google ya no respeta el parámetro num=100, así que “los 100 primeros resultados” significa diez solicitudes paginadas separadas, no una. Cualquier estimación de coste escrita antes de ese cambio está equivocada por un factor de 10x.

Ninguno de estos es un problema difícil por separado. Juntos son una cinta de correr de mantenimiento: acabas siendo dueño de un pool de proxies y de un parser que necesita atención cada trimestre, para conseguir datos que no son tu producto.

La versión que sigue funcionando

Envía la consulta a una API que ya se encarga de los proxies y del parsing, y recibe JSON de vuelta:

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

Sin configuración de proxies, sin selectores. La respuesta también trae people_also_ask, knowledge_graph, ads y related_searches cuando la consulta los genera, así que compruebas si existe una clave en lugar de escribir un parser por tipo de resultado.

Segmentación por país e idioma

Los datos de ranking no significan nada sin una ubicación. Pasa el país explícitamente en lugar de confiar en que la solicitud caiga en algún sitio útil:

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

La consulta se emite desde esa ubicación, con el dominio de Google correspondiente y los valores de idioma por defecto.

Recopilar más de una página

Como los resultados llegan de diez en diez por página, trata la profundidad como un bucle y recuerda que cada página es una solicitud facturable independiente:

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

Manejar los errores correctamente

Tres códigos de estado importan en producción, y tratarlos todos como “reintentar” es la forma de quemar un saldo de créditos en una tarde:

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

        # Sin créditos: reintentar nunca ayudará.
        if response.status_code == 402:
            raise RuntimeError("Out of credits, top up before retrying")

        # Límite de concurrency: espera exactamente lo que pide la API.
        if response.status_code == 429:
            time.sleep(int(response.headers.get("Retry-After", 1)))
            continue

        # Fallo upstream: no se factura, es seguro reintentar con 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 regla general: 402 es terminal, 429 significa reduce la velocidad exactamente en la cantidad de Retry-After, y 5xx se puede reintentar porque las solicitudes fallidas no se cobran.

Ejecutar consultas en paralelo

La investigación de keywords implica cientos de consultas, y el límite que importa es la concurrency, no el volumen total. Acótala explícitamente:

from concurrent.futures import ThreadPoolExecutor

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

# Mantén esto igual o por debajo de la concurrency permitida por tu 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"]))

Superar la concurrency de tu plan no hace el trabajo más rápido, solo convierte solicitudes en 429.

Lo que cuesta

Una búsqueda cuesta 5 créditos, y cada respuesta te dice cuánto gastó en el header X-FastAF-Credits-Used, así que puedes medir por función en lugar de adivinar a posteriori. Consulta precios para las cifras por pack, o la referencia de la Google SERP API para la lista completa de parámetros y respuesta.