Скрейпинг Google People Also Ask и Autocomplete для исследования ключевых слов: руководство по Python

Скрейпинг Google People Also Ask и Autocomplete для исследования ключевых слов — это один из самых дешёвых способов собрать тысячи реальных поисковых запросов пользователей без платных SEO-инструментов. Google бесплатно отдаёт три источника данных: автодополнения (Autocomplete), блок «People Also Ask» (PAA) и связанные запросы (Related Searches). В этом руководстве мы разберём, как написать на Python надёжный скрейпер, который комбинирует все три источника, обходит ограничения Google с помощью резидентных прокси и выдаёт готовый CSV для контент-планирования.

Три бесплатных источника ключевых слов: Autocomplete, PAA и Related Searches

Google предоставляет три публичных механизма, каждый из которых отражает разный уровень поискового намерения (search intent). Понимание того, как они устроены, критично для скрейпинга Google People Also Ask и Autocomplete.

1. Autocomplete (Suggest API)

Эндпоинт suggestqueries.google.com/complete/search возвращает JSON-массив подсказок, которые Google показывает в строке поиска при вводе текста. Это самый быстрый способ собрать длиннохвостые (long-tail) запросы — один запрос возвращает 8–10 вариантов. Подсказки ранжируются по популярности и релевантности, поэтому они отлично подходят для верхнего уровня воронки контента.

Базовый URL выглядит так:

https://suggestqueries.google.com/complete/search?client=chrome&q=seo

Параметр client=chrome возвращает компактный JSON вместо XML. Ответ — это массив, где второй элемент содержит список подсказок.

2. People Also Ask (PAA)

Блок PAA — это аккордеон вопросов, который Google вставляет в органическую выдачу. Каждый вопрос можно развернуть, и при разворачивании Google подгружает ответ и новые вложенные вопросы. Это делает PAA рекурсивным золотым дном: один вопрос порождает 4–8 новых, и так до глубины 5–7 уровней. PAA-запросы обычно соответствуют информационному интенту (informational intent) — пользователи ищут ответы, а не товары.

3. Related Searches (Связанные запросы)

Внизу страницы выдачи Google показывает 8 связанных запросов. Они шире, чем Autocomplete, и часто содержат переформулировки, синонимы и смежные темы. Это хороший источник для кластеризации и расширения семантического ядра.

Источник	Тип интента	Глубина	Формат	Скорость сбора
Autocomplete	Коммерческий + информационный	Long-tail	JSON API	~100 запросов/сек
People Also Ask	Информационный	Рекурсивная, 5–7 уровней	HTML + JS	~2–5 запросов/сек
Related Searches	Смешанный	Широкий	HTML	~10 запросов/сек

Почему скрейпинг Google People Also Ask и Autocomplete требует прокси

Google активно защищает свои страницы от автоматизированного сбора. Проблемы начинаются, когда вы делаете десятки запросов в минуту с одного IP:

• Rate limits по IP — после ~50–100 запросов за короткое время Google начинает возвращать CAPTCHA или HTTP 429. По данным документации Google для разработчиков, официальные API имеют жёсткие квоты, но неофициальные эндпоинты вообще не задокументированы и лимитируются эвристически.
• Локальная предвзятость (locale bias) — Autocomplete и PAA зависят от языка и региона запроса. Если вы запрашиваете из дата-центра в США, вы получите американские подсказки, даже если исследуете немецкий рынок.
• Отпечатки TLS и заголовки — Google частично определяет ботов по TLS-отпечатку и неестественным заголовкам. Дата-центровые IP усиливают подозрения.

Решение — резидентные прокси с геотаргетингом. ProxyHat предоставляет пул резидентных IP с настройкой страны и города прямо в имени пользователя, что позволяет получать локализованные подсказки без искажений. Подробнее о доступных локациях — на странице локаций ProxyHat.

Архитектура скрейпера: от seed-слова до CSV

Полный пайплайн выглядит так:

1. Принять seed-слова (например, «seo», «proxy», «web scraping»).
2. Расширить каждое seed-слово модификаторами (a–z, «how», «what», «best», «vs»).
3. Для каждой комбинации вызвать Autocomplete API через прокси.
4. Для топ-N результатов открыть SERP через Playwright и собрать PAA-вопросы.
5. Рекурсивно разворачивать PAA-аккордеоны.
6. Собрать Related Searches со страницы.
7. Дедуплицировать, кластеризовать по интенту, экспортировать в CSV.

Настройка прокси ProxyHat

ProxyHat использует единый шлюз gate.proxyhat.com с геотаргетингом и сессиями в имени пользователя. Для локализованного сбора в Германии используйте формат user-country-DE-city-berlin. Для HTTP — порт 8080, для SOCKS5 — 1080.

# .env
PROXYHAT_USERNAME=user-country-DE-city-berlin
PROXYHAT_PASSWORD=your_password_here
PROXYHAT_GATEWAY=gate.proxyhat.com
PROXYHAT_HTTP_PORT=8080
PROXYHAT_SOCKS5_PORT=1080

Пример 1: Сбор Autocomplete через httpx с прокси

import os
import httpx
import asyncio
import json
from typing import List

PROXY_USER = os.getenv("PROXYHAT_USERNAME", "user-country-DE-city-berlin")
PROXY_PASS = os.getenv("PROXYHAT_PASSWORD", "your_password")
PROXY_URL = f"http://{PROXY_USER}:{PROXY_PASS}@gate.proxyhat.com:8080"

AUTOCOMPLETE_URL = "https://suggestqueries.google.com/complete/search"

HEADERS = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
                  "AppleWebKit/537.36 (KHTML, like Gecko) "
                  "Chrome/120.0.0.0 Safari/537.36",
    "Accept-Language": "de-DE,de;q=0.9,en;q=0.8",
}

async def fetch_autocomplete(
    client: httpx.AsyncClient,
    query: str,
    retries: int = 3,
) -> List[str]:
    """Получить подсказки Autocomplete для одного запроса."""
    params = {"client": "chrome", "q": query}
    for attempt in range(retries):
        try:
            resp = await client.get(
                AUTOCOMPLETE_URL,
                params=params,
                headers=HEADERS,
                timeout=10.0,
            )
            resp.raise_for_status()
            data = resp.json()
            # data[1] — список подсказок
            if len(data) > 1 and isinstance(data[1], list):
                return data[1]
            return []
        except (httpx.HTTPStatusError, httpx.RequestError, json.JSONDecodeError) as e:
            print(f"[Попытка {attempt+1}] Ошибка для '{query}': {e}")
            await asyncio.sleep(2 ** attempt)
    return []

async def expand_keyword(keyword: str, modifiers: List[str]) -> List[str]:
    """Расширить seed-слово модификаторами и собрать все подсказки."""
    expanded = []
    # Сначала сам запрос
    queries = [keyword]
    # Добавляем модификаторы: a-z, how, what, best, vs
    for mod in modifiers:
        queries.append(f"{keyword} {mod}")
    # Префиксы a-z
    for letter in "abcdefghijklmnopqrstuvwxyz":
        queries.append(f"{keyword} {letter}")

    async with httpx.AsyncClient(proxy=PROXY_URL) as client:
        tasks = [fetch_autocomplete(client, q) for q in queries]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        for r in results:
            if isinstance(r, list):
                expanded.extend(r)
    return list(set(expanded))

# Запуск
if __name__ == "__main__":
    seed = "seo"
    mods = ["how", "what", "best", "vs", "tools", "tips", "guide", "tutorial"]
    suggestions = asyncio.run(expand_keyword(seed, mods))
    print(f"Собрано {len(suggestions)} подсказок для '{seed}'")
    for s in suggestions[:20]:
        print(f"  - {s}")

Этот код превращает одно seed-слово в 200–400 подсказок за один проход. Модификаторы a–z — классический приём, описанный в документации Google по поисковым API как способ расширения запросов.

Пример 2: Сбор People Also Ask через Playwright

PAA требует рендеринга JavaScript, поэтому httpx недостаточно. Playwright может открыть страницу, кликать по аккордеонам и собирать вопросы вместе с источниками ответов.

import asyncio
from playwright.async_api import async_playwright
from typing import List, Dict
import os
import random

PROXY_USER = os.getenv("PROXYHAT_USERNAME", "user-country-DE-city-berlin")
PROXY_PASS = os.getenv("PROXYHAT_PASSWORD", "your_password")

async def scrape_paa(
    keyword: str,
    max_depth: int = 3,
    max_questions: int = 50,
) -> List[Dict[str, str]]:
    """Собрать PAA-вопросы и источники ответов для ключевого слова."""
    questions = []
    seen = set()

    async with async_playwright() as p:
        browser = await p.chromium.launch(
            headless=True,
            proxy={
                "server": "http://gate.proxyhat.com:8080",
                "username": PROXY_USER,
                "password": PROXY_PASS,
            },
        )
        context = await browser.new_context(
            locale="de-DE",
            user_agent=(
                "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
                "AppleWebKit/537.36 (KHTML, like Gecko) "
                "Chrome/120.0.0.0 Safari/537.36"
            ),
        )
        page = await context.new_page()

        url = f"https://www.google.com/search?q={keyword}&hl=de&gl=de"
        await page.goto(url, wait_until="domcontentloaded", timeout=30000)

        depth = 0
        while depth < max_depth and len(questions) < max_questions:
            # Найти все PAA-вопросы на странице
            paa_elements = await page.query_selector_all(
                "div[jsname='Cpkphb'] .wDYxhc, "
                "div.related-question-pair"
            )

            if not paa_elements:
                break

            # Кликнуть по неразвёрнутым вопросам для подгрузки новых
            for el in paa_elements[:8]:
                text = await el.inner_text()
                question_text = text.strip().split("\n")[0]
                if question_text and question_text not in seen:
                    seen.add(question_text)
                    questions.append({
                        "question": question_text,
                        "source_url": "",
                        "depth": depth,
                    })
                    # Кликнуть для разворачивания и подгрузки вложенных
                    try:
                        await el.click(timeout=3000)
                        await page.wait_for_timeout(
                            random.randint(1500, 3000)
                        )
                        # Попробовать извлечь источник ответа
                        link = await page.query_selector(
                            "div[jsname='Cpkphb'] a[data-header-feature]"
                        )
                        if link:
                            href = await link.get_attribute("href")
                            if href:
                                questions[-1]["source_url"] = href
                    except Exception:
                        pass

            depth += 1
            await page.wait_for_timeout(random.randint(2000, 4000))

        await browser.close()
    return questions

if __name__ == "__main__":
    results = asyncio.run(scrape_paa("was ist seo", max_depth=4))
    print(f"Собрано {len(results)} PAA-вопросов")
    for q in results[:15]:
        print(f"  [{q['depth']}] {q['question']}")

Пример 3: Сбор Related Searches

import httpx
import asyncio
from selectolax.parser import HTMLParser
import os

PROXY_URL = (
    f"http://{os.getenv('PROXYHAT_USERNAME', 'user-country-DE')}"
    f":{os.getenv('PROXYHAT_PASSWORD', 'pass')}"
    "@gate.proxyhat.com:8080"
)

HEADERS = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
                  "AppleWebKit/537.36 (KHTML, like Gecko) "
                  "Chrome/120.0.0.0 Safari/537.36",
    "Accept-Language": "de-DE,de;q=0.9",
}

async def fetch_related_searches(keyword: str) -> list[str]:
    """Собрать связанные запросы со страницы выдачи Google."""
    url = f"https://www.google.com/search?q={keyword}&hl=de&gl=de"
    async with httpx.AsyncClient(proxy=PROXY_URL) as client:
        resp = await client.get(url, headers=HEADERS, timeout=15.0)
        resp.raise_for_status()

    tree = HTMLParser(resp.text)
    related = []
    for node in tree.css("a"):  # Related searches — ссылки внизу страницы
        href = node.attributes.get("href", "")
        if "/search?q=" in href and "related" in href:
            text = node.text(strip=True)
            if text and len(text) > 5:
                related.append(text)
    return list(set(related))

if __name__ == "__main__":
    results = asyncio.run(fetch_related_searches("seo tools"))
    print(f"Связанных запросов: {len(results)}")
    for r in results:
        print(f"  - {r}")

Пример 4: Рекурсивное расширение PAA

Ключевая особенность PAA — рекурсия. Каждый развёрнутый вопрос порождает новые. Вот функция, которая обходит PAA вглубь:

import asyncio
from playwright.async_api import async_playwright
import os, random

PROXY_USER = os.getenv("PROXYHAT_USERNAME", "user-country-DE-city-berlin")
PROXY_PASS = os.getenv("PROXYHAT_PASSWORD", "pass")

async def recursive_paa_expansion(
    seed_keyword: str,
    max_total: int = 200,
    max_depth: int = 5,
) -> list[dict]:
    """Рекурсивно разворачивать PAA-аккордеоны."""
    all_questions: list[dict] = []
    seen: set[str] = set()

    async with async_playwright() as p:
        browser = await p.chromium.launch(
            headless=True,
            proxy={
                "server": "http://gate.proxyhat.com:8080",
                "username": PROXY_USER,
                "password": PROXY_PASS,
            },
        )
        context = await browser.new_context(locale="de-DE")
        page = await context.new_page()

        url = f"https://www.google.com/search?q={seed_keyword}&hl=de&gl=de"
        await page.goto(url, wait_until="domcontentloaded", timeout=30000)

        for depth in range(max_depth):
            if len(all_questions) >= max_total:
                break

            # Найти видимые PAA-вопросы
            paa_items = await page.query_selector_all(
                "div[jsname='Cpkphb'] span[jsname='r5nVd']"
            )

            if not paa_items:
                break

            # Кликнуть по первым 4 неразвёрнутым
            clicked = 0
            for item in paa_items:
                if len(all_questions) >= max_total:
                    break
                text = (await item.inner_text()).strip()
                if text not in seen and clicked < 4:
                    seen.add(text)
                    all_questions.append({
                        "question": text,
                        "depth": depth,
                        "keyword": seed_keyword,
                    })
                    try:
                        await item.click(timeout=3000)
                        await page.wait_for_timeout(
                            random.randint(2000, 4000)
                        )
                        clicked += 1
                    except Exception:
                        pass

            await page.wait_for_timeout(random.randint(3000, 5000))

        await browser.close()
    return all_questions

if __name__ == "__main__":
    data = asyncio.run(recursive_paa_expansion("proxy server", max_total=100))
    print(f"Всего PAA-вопросов: {len(data)}")
    for d in data[:20]:
        print(f"  [d={d['depth']}] {d['question']}")

Пример 5: Дедупликация, кластеризация и экспорт в CSV

import csv
import re
from collections import defaultdict
from typing import List, Dict

def deduplicate(keywords: List[str]) -> List[str]:
    """Удалить дубликаты с нормализацией."""
    seen = set()
    result = []
    for kw in keywords:
        normalized = re.sub(r"\s+", " ", kw.strip().lower())
        if normalized not in seen and len(normalized) > 3:
            seen.add(normalized)
            result.append(kw.strip())
    return result

def cluster_by_intent(keywords: List[str]) -> Dict[str, List[str]]:
    """Грубо кластеризовать по интенту по модификаторам."""
    clusters = defaultdict(list)
    intent_markers = {
        "informational": ["was ist", "how", "what", "why", "guide", "tutorial"],
        "commercial": ["best", "top", "vs", "review", "vergleich"],
        "transactional": ["buy", "kaufen", "price", "cost", "preis"],
        "navigational": ["login", "download", "official"],
    }
    for kw in keywords:
        kw_lower = kw.lower()
        matched = False
        for intent, markers in intent_markers.items():
            if any(m in kw_lower for m in markers):
                clusters[intent].append(kw)
                matched = True
                break
        if not matched:
            clusters["general"].append(kw)
    return dict(clusters)

def export_to_csv(
    all_keywords: List[str],
    paa_questions: List[Dict],
    related: List[str],
    output_path: str = "keyword_research.csv",
):
    """Экспортировать все данные в CSV."""
    with open(output_path, "w", newline="", encoding="utf-8") as f:
        writer = csv.writer(f)
        writer.writerow(["type", "keyword", "intent", "source", "depth"])

        # Autocomplete
        for kw in deduplicate(all_keywords):
            writer.writerow(["autocomplete", kw, "", "suggestqueries.google.com", 0])

        # PAA
        for item in paa_questions:
            writer.writerow([
                "paa",
                item.get("question", ""),
                "informational",
                "google.com/search",
                item.get("depth", 0),
            ])

        # Related
        for kw in deduplicate(related):
            writer.writerow(["related", kw, "", "google.com/search", 0])

    print(f"Экспортировано в {output_path}")

# Сборка
if __name__ == "__main__":
    # Имитация собранных данных
    autocomplete_kw = ["seo tools", "seo tips", "best seo software", "was ist seo"]
    paa_data = [
        {"question": "Was ist SEO?", "depth": 0},
        {"question": "Wie funktioniert SEO?", "depth": 1},
        {"question": "Was kostet SEO?", "depth": 2},
    ]
    related_kw = ["seo optimierung", "seo agentur", "seo ranking"]

    export_to_csv(autocomplete_kw, paa_data, related_kw)

    # Кластеризация
    clusters = cluster_by_intent(autocomplete_kw + related_kw)
    for intent, kws in clusters.items():
        print(f"\n{intent} ({len(kws)}):")
        for k in kws:
            print(f"  - {k}")

Пример 6: Полный пайплайн с ротацией прокси и rate limiting

import asyncio
import httpx
import os
import random
from itertools import cycle

# Несколько сессий для ротации
SESSIONS = [
    "user-session-sess01-country-DE-city-berlin",
    "user-session-sess02-country-DE-city-munich",
    "user-session-sess03-country-DE-city-hamburg",
    "user-session-sess04-country-AT-city-vienna",
]
PROXY_PASS = os.getenv("PROXYHAT_PASSWORD", "pass")

def get_proxy_url(session: str) -> str:
    return f"http://{session}:{PROXY_PASS}@gate.proxyhat.com:8080"

async def rate_limited_fetch(
    queries: list[str],
    max_concurrent: int = 5,
    delay_range: tuple[float, float] = (1.0, 2.5),
) -> dict[str, list[str]]:
    """Собрать Autocomplete с ротацией прокси и rate limiting."""
    results: dict[str, list[str]] = {}
    semaphore = asyncio.Semaphore(max_concurrent)
    session_pool = cycle(SESSIONS)

    async def fetch_one(client: httpx.AsyncClient, q: str) -> list[str]:
        async with semaphore:
            await asyncio.sleep(random.uniform(*delay_range))
            params = {"client": "chrome", "q": q}
            try:
                resp = await client.get(
                    "https://suggestqueries.google.com/complete/search",
                    params=params,
                    timeout=10.0,
                )
                data = resp.json()
                return data[1] if len(data) > 1 else []
            except Exception as e:
                print(f"Ошибка для '{q}': {e}")
                return []

    # Ротируем прокси между пакетами запросов
    batch_size = max(1, len(queries) // len(SESSIONS) + 1)
    for i in range(0, len(queries), batch_size):
        batch = queries[i:i + batch_size]
        proxy_url = get_proxy_url(next(session_pool))
        async with httpx.AsyncClient(proxy=proxy_url) as client:
            tasks = [fetch_one(client, q) for q in batch]
            batch_results = await asyncio.gather(*tasks)
            for q, r in zip(batch, batch_results):
                results[q] = r

    return results

if __name__ == "__main__":
    seeds = ["seo", "proxy", "web scraping", "keyword research"]
    mods = ["how", "what", "best", "tools", "guide"]
    all_queries = []
    for s in seeds:
        for m in mods:
            all_queries.append(f"{s} {m}")

    data = asyncio.run(rate_limited_fetch(all_queries, max_concurrent=3))
    total = sum(len(v) for v in data.values())
    print(f"Запросов: {len(all_queries)}, подсказок: {total}")

Частые ошибки и edge cases

• Пустые ответы Autocomplete — Google иногда возвращает пустой массив для слишком специфичных запросов. Добавляйте fallback на модификаторы.
• Изменение структуры PAA-селекторов — Google регулярно меняет CSS-классы и jsname атрибуты. Используйте несколько fallback-селекторов и логируйте, когда ничего не найдено.
• CAPTCHA при агрессивном скрейпинге — если вы получаете CAPTCHA, снизьте частоту до 1 запроса в 3–5 секунд и используйте разные сессии ProxyHat для ротации IP.
• Локальная предвзятость — если вы не указываете country в имени пользователя ProxyHat, вы можете получить подсказки для случайной страны. Всегда явно задавайте гео.
• Кодировка — Autocomplete возвращает UTF-8, но PAA может содержать эмодзи и нестандартные символы. Используйте encoding="utf-8" при записи CSV.

Этика и легальность

Autocomplete и PAA — это публичные данные, которые Google показывает любому пользователю. Однако массовый автоматизированный сбор может нарушать Условия использования Google. Несколько принципов:

• Соблюдайте robots.txt — проверяйте, разрешён ли путь перед скрейпингом.
• Ограничивайте частоту: 1 запрос в 2–3 секунды на IP — разумный минимум.
• Для production-масштаба рассмотрите официальные API: Google Custom Search API и Google Trends API. Они легальнее, но ограничены по объёму.
• Не используйте собранные данные для спама или манипуляций.

Практическое правило: если ваш объём сбора превышает 10 000 запросов в день, переходите на официальные API или комбинируйте с прокси-ротацией очень аккуратно.

ProxyHat-специфичная настройка

Для скрейпинга Google мы рекомендуем резидентные прокси ProxyHat с геотаргетингом. Конфигурация:

• HTTP: http://user-country-DE-city-berlin:pass@gate.proxyhat.com:8080
• SOCKS5: socks5://user-country-DE-city-berlin:pass@gate.proxyhat.com:1080
• Sticky-сессии: user-session-abc123-country-DE:pass@gate.proxyhat.com:8080

Sticky-сессии полезны, когда вам нужно сохранить один IP для серии запросов (например, для разворачивания PAA-аккордеонов в Playwright). Ротация сессий — для параллельного сбора Autocomplete. Подробнее о тарифах — на странице цен ProxyHat. Документация по подключению — на docs.proxyhat.com.

Key Takeaways

• Autocomplete API — самый быстрый источник: ~100 запросов/сек, 8–10 подсказок на запрос.
• PAA — рекурсивный источник: один seed может дать 50–200 вопросов за 5 уровней разворачивания.
• Related Searches — широкий источник для кластеризации тем.
• Резидентные прокси с геотаргетингом обязательны для локализованных подсказок без CAPTCHA.
• Ротация сессий ProxyHat (user-session-xxx-country-DE) балансирует между стабильностью и обходом лимитов.
• Всегда нормализуйте и дедуплицируйте перед экспортом — до 30% собранных ключевых слов могут быть дубликатами.

Готовы начать? Изучите use-case скрейпинга и SERP-трекинга для дополнительных сценариев.