Como extrair resultados de busca do Google em Python

Todo projeto de scraping que envolve o Google começa do mesmo jeito: requests.get("https://www.google.com/search?q=..."), um pouco de BeautifulSoup, e um parser que funciona lindamente por uns dois dias.

Depois para de funcionar. Aqui está o porquê, e como é de fato uma configuração que continua funcionando.

Por que a versão DIY quebra

Quatro coisas diferentes quebram isso, geralmente nessa ordem:

1. Bloqueio de IP. IPs de datacenter são limitados rapidamente, e depois recebem uma página de consentimento ou um CAPTCHA em vez de resultados. Rotacionar alguns proxies adia isso, não resolve.
2. Instabilidade do markup. O markup de resultados do Google usa nomes de classe ofuscados e rotativos. Um seletor CSS que funciona hoje não é um contrato estável.
3. Tipos de resultado. Links orgânicos são a parte fácil. Featured snippets, People Also Ask, knowledge panels, anúncios e local packs têm cada um sua própria estrutura, e a presença deles varia por busca e por país.
4. A matemática da paginação. O Google não respeita mais o parâmetro num=100, então “os 100 melhores resultados” agora significa dez requisições paginadas separadas, não uma. Qualquer estimativa de custo escrita antes dessa mudança está errada em 10x.

Nenhum desses é um problema difícil isoladamente. Juntos, viram uma esteira de manutenção: você acaba sendo dono de um pool de proxies e de um parser que precisa de atenção todo trimestre, para obter dados que não são o seu produto.

A versão que continua funcionando

Mande a busca para uma API que já tem os proxies e o parsing, e receba JSON de volta:

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

Sem configuração de proxy, sem seletores. A resposta também traz people_also_ask, knowledge_graph, ads e related_searches quando a busca os gera, então você verifica a presença de uma chave em vez de escrever um parser para cada tipo de resultado.

Segmentação por país e idioma

Dados de ranking não significam nada sem uma localização. Passe o país explicitamente em vez de torcer para a requisição cair em algum lugar útil:

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

A busca é feita a partir dessa localização, com o domínio do Google e os padrões de idioma correspondentes.

Coletando mais de uma página

Como os resultados vêm dez por página, trate a profundidade como um loop e lembre que cada página é uma requisição cobrada separadamente:

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

Tratando erros direito

Três códigos de status importam em produção, e tratar todos como “tenta de novo” é como você queima um saldo de créditos em uma 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()

        # Sem créditos: tentar de novo nunca vai resolver.
        if response.status_code == 402:
            raise RuntimeError("Sem créditos, recarregue antes de tentar de novo")

        # Limite de concorrência: espere exatamente o tempo que a API pedir.
        if response.status_code == 429:
            time.sleep(int(response.headers.get("Retry-After", 1)))
            continue

        # Falha upstream: não é cobrada, seguro tentar de novo com 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("Tentativas esgotadas")

A regra prática: 402 é definitivo, 429 significa desacelerar exatamente pelo tempo indicado em Retry-After, e 5xx pode ser tentado de novo porque requisições falhas não são cobradas.

Rodando buscas em paralelo

Pesquisa de palavras-chave significa centenas de buscas, e o limite que importa é a concorrência, não o volume total. Limite isso explicitamente:

from concurrent.futures import ThreadPoolExecutor

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

# Mantenha isso igual ou abaixo da concorrência permitida pelo seu plano.
with ThreadPoolExecutor(max_workers=5) as pool:
    for keyword, data in zip(keywords, pool.map(search_safe, keywords)):
        print(keyword, len(data["organic_results"]))

Ir além da concorrência do seu plano não deixa o trabalho mais rápido, só converte requisições em 429.

Quanto custa

Uma busca custa 5 créditos, e cada resposta informa quanto foi gasto no header X-FastAF-Credits-Used, para você medir por funcionalidade em vez de descobrir só depois. Veja preços para os números por pacote, ou a referência da Google SERP API para a lista completa de parâmetros e respostas.