Google-Suchergebnisse mit Python scrapen

Jedes Scraping-Projekt, das mit Google zu tun hat, beginnt gleich: requests.get("https://www.google.com/search?q=..."), ein bisschen BeautifulSoup, und ein Parser, der etwa zwei Tage lang wunderbar funktioniert.

Dann hört er auf. Hier ist, warum er aufhört, und wie ein Setup aussieht, das wirklich weiterläuft.

Warum die DIY-Version bricht

Vier separate Dinge bringen sie zum Scheitern, meist in dieser Reihenfolge:

1. IP-Blocking. Datacenter-IPs werden schnell gedrosselt und bekommen dann statt Ergebnissen eine Consent-Seite oder ein CAPTCHA. Ein paar Proxys zu rotieren verzögert das, es löst es nicht.
2. Markup-Fluktuation. Googles Ergebnis-Markup trägt verschleierte, rotierende Klassennamen. Ein CSS-Selektor, der heute matcht, ist kein stabiler Vertrag.
3. Ergebnistypen. Organische Links sind der einfache Teil. Featured Snippets, People Also Ask, Knowledge Panels, Ads und Local Packs haben jeweils ihre eigene Struktur, und ihr Vorkommen variiert nach Query und Land.
4. Pagination-Mathematik. Google respektiert den Parameter num=100 nicht mehr, der früher 100 Ergebnisse in einer Anfrage lieferte. “Die Top 100 Ergebnisse” bedeutet jetzt zehn separate paginierte Anfragen, nicht eine. Jede Kostenschätzung, die vor dieser Änderung geschrieben wurde, liegt um das 10-Fache daneben.

Keins davon ist einzeln ein schweres Problem. Zusammen sind sie ein Wartungstretmühle: Du besitzt am Ende einen Proxy-Pool und einen Parser, der jedes Quartal Aufmerksamkeit braucht, für Daten, die nicht dein Produkt sind.

Die Version, die weiterläuft

Schicke die Query an eine API, der die Proxys und das Parsing gehören, und bekomme JSON zurück:

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

Keine Proxy-Konfiguration, keine Selektoren. Die Antwort trägt außerdem people_also_ask, knowledge_graph, ads und related_searches, wenn die Query sie hergibt, sodass du auf einen Key prüfst statt einen Parser pro Ergebnistyp zu schreiben.

Land- und Sprach-Targeting

Ranking-Daten sind ohne Standort bedeutungslos. Übergib das Land explizit, statt zu hoffen, dass die Anfrage irgendwo Sinnvollem landet:

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

Die Query wird von diesem Standort aus gestellt, mit der passenden Google-Domain und den Sprach-Defaults.

Mehr als eine Seite sammeln

Da Ergebnisse zu je zehn pro Seite kommen, behandle Tiefe als Schleife und denke daran, dass jede Seite eine separate abrechenbare Anfrage ist:

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

Fehler richtig behandeln

Drei Statuscodes zählen in Produktion, und alle als “retry” zu behandeln, ist der Weg, wie du an einem Nachmittag ein Credit-Guthaben verbrennst:

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

        # Keine Credits mehr: erneute Versuche helfen nie.
        if response.status_code == 402:
            raise RuntimeError("Out of credits, top up before retrying")

        # Concurrency-Limit: genau so lange warten, wie die API verlangt.
        if response.status_code == 429:
            time.sleep(int(response.headers.get("Retry-After", 1)))
            continue

        # Upstream-Fehler: nicht berechnet, sicher erneut zu versuchen mit 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")

Die Faustregel: 402 ist endgültig, 429 heißt, genau so viel langsamer machen, wie in Retry-After steht, und 5xx ist wiederholbar, weil fehlgeschlagene Anfragen nicht berechnet werden.

Queries parallel ausführen

Keyword-Recherche bedeutet Hunderte Queries, und das Limit, das zählt, ist Concurrency, nicht das Gesamtvolumen. Grenze sie explizit ein:

from concurrent.futures import ThreadPoolExecutor

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

# Bleib bei oder unter der Concurrency, die dein Plan erlaubt.
with ThreadPoolExecutor(max_workers=5) as pool:
    for keyword, data in zip(keywords, pool.map(search_safe, keywords)):
        print(keyword, len(data["organic_results"]))

Über die Concurrency deines Plans zu gehen, macht den Job nicht schneller, es verwandelt nur Anfragen in 429s.

Was es kostet

Eine Suche kostet 5 Credits, und jede Antwort verrät im Header X-FastAF-Credits-Used, was sie ausgegeben hat, sodass du pro Feature messen kannst, statt es im Nachhinein zu erraten. Die Zahlen pro Paket findest du unter Preise, oder in der Google-SERP-API-Referenz die vollständige Liste der Parameter und Antwortfelder.