Google People Also AskとAutocompleteのスクレイピングでキーワード調査(Pythonガイド)

GoogleのAutocompleteエンドポイント、People Also Ask、Related Searchesの3つの無料キーワードソースをPythonでスクレイピングし、residential proxyでローカライズされた提案データを取得する実践ガイド。

Scrape Google People Also Ask and Autocomplete for Keyword Research (Python Guide)

Google People Also AskとAutocompleteのスクレイピングでキーワード調査を行うことは、SEO専門家やデベロッパーにとって、有料ツールに頼らずに大量のロングテールキーワードと検索インテントを発見する強力な手法です。本ガイドでは、Pythonを使ってAutocompleteの提案、PAAの質問、Related Searchesを収集し、residential proxyでローカライズされたクリーンなデータを取得する方法を、実行可能なコードとともに解説します。

3つの無料キーワード金脈:Autocomplete・PAA・Related Searches

Googleの検索結果ページには、キーワード調査に直結する3つの無料データソースが存在します。それぞれが異なる検索インテントを反映しているため、組み合わせて使うことで包括的なキーワードマップを構築できます。

データソースエンドポイント / 取得方法反映するインテントデータ形式
Autocomplete(サジェスト)suggestqueries.google.com/complete/search?client=chrome&q=入力中のリアルタイム検索意図JSON(Chrome client)
People Also Ask(PAA)検索結果ページ内のアコーディオン質問形式の情報検索インテントHTML(Playwrightで展開)
Related Searches検索結果ページ下部関連・派生検索インテントHTML

Autocompleteは Google Suggest として知られる機能で、ユーザーが検索バーに入力している最中にリアルタイムの提案を返します。このエンドポイントは client=chrome パラメータを指定するとJSON形式で応答するため、スクレイピングに最も適しています。

PAAは検索結果ページに表示される「他の人はこちらも質問」というアコーディオンで、クリックすると回答と出典リンクが展開されます。各質問をクリックすると新しい質問がネストされて追加されるため、再帰的に展開することで数百の質問キーワードを取得できます。

シードからロングテールへの展開戦略

1つのシードキーワードから数百のロングテールバリエーションを生成するには、Autocompleteエンドポイントに対して体系的にクエリを送信します。基本戦略は2つです。

1. a-z展開と修飾子プレフィックス

シードキーワードの後に a から z を順に付与してAutocompleteに送信すると、各文字で異なる提案が返ります。さらに how towhat isbestvs などの修飾子プレフィックスを組み合わせることで、1キーワードから200以上のバリエーションを生成できます。

2. 再帰的PAA展開

PAAアコーディオンをクリックして展開すると、新しい質問が追加されます。この新しい質問を次のシードとして再帰的に展開することで、階層的に深い質問ツリーを構築できます。

なぜレート制限とロケールバイアスが発生するのか

Autocompleteエンドポイントに短時間で大量のリクエストを送ると、Googleのper-IPレート制限により429 Too Many Requestsが返されます。また、同じIPから異なる国のローカライズ提案を取得しようとすると、IPの地理的位置に基づくロケールバイアスが発生し、期待した国の提案が得られません。

これを解決するには、residential proxyでリクエスト元IPを分散させ、かつ国・都市レベルのジオターゲティングを指定する必要があります。ProxyHatのresidential proxyでは、ユーザー名に -country-DE-city-berlin のようなフラグを付与することで、ドイツ・ベルリンの住宅IPからリクエストを送信できます。

# ProxyHatのプロキシURL形式(HTTP)
# 基本形式
http://USERNAME:PASSWORD@gate.proxyhat.com:8080

# 国指定(ドイツ)
http://user-country-DE:pass@gate.proxyhat.com:8080

# 国・都市指定(ドイツ・ベルリン)
http://user-country-DE-city-berlin:pass@gate.proxyhat.com:8080

# セッション固定(sticky session)
http://user-session-abc123:pass@gate.proxyhat.com:8080

# SOCKS5形式
socks5://user-country-US:pass@gate.proxyhat.com:1080

セッション固定を使うと、同じ仮想セッションIDの間は同じIPが維持されるため、PAA展開のように複数ページにわたるステップフルなスクレイピングでIPの一貫性を保てます。一方、Autocompleteのようなステートレスなリクエストでは、リクエストごとにIPをローテーションすることでレート制限を回避できます。

詳しいロケーション設定は ProxyHatのロケーション一覧 を参照してください。プロキシタイプの選択については Webスクレイピングのユースケース も参考になります。

実装例1:httpxでAutocomplete JSONを取得

まずは最もシンプルなAutocompleteスクレイピングから始めます。Playwright は後述のPAA展開で使いますが、AutocompleteはHTTPリクエストだけでJSONを取得できるため、httpxで十分です。

import httpx
import json
import time
import random

# ProxyHatプロキシ設定
PROXY_BASE = "http://user-country-{country}:pass@gate.proxyhat.com:8080"

def fetch_autocomplete(keyword: str, country: str = "US", city: str = None) -> list[str]:
    """Google Autocompleteからサジェストを取得する"""
    username = f"user-country-{country}"
    if city:
        username += f"-city-{city.lower().replace(' ', '')}"
    proxy_url = f"http://{username}:pass@gate.proxyhat.com:8080"

    suggestions = []
    for attempt in range(3):
        try:
            with httpx.Client(
                proxy=proxy_url,
                timeout=10.0,
                headers={
                    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
                    "Accept-Language": "en-US,en;q=0.9",
                },
            ) as client:
                resp = client.get(
                    "https://suggestqueries.google.com/complete/search",
                    params={"client": "chrome", "q": keyword},
                )
                resp.raise_for_status()
                data = resp.json()
                # data[1] はサジェストのリスト、各要素の[0]がテキスト
                suggestions = [s[0] for s in data[1]]
                break
        except (httpx.HTTPStatusError, httpx.ProxyError, httpx.ConnectError) as e:
            print(f"Attempt {attempt+1} failed for '{keyword}': {e}")
            time.sleep(2 ** attempt + random.uniform(0, 1))
    return suggestions

# 使用例
results = fetch_autocomplete("python tutorial", country="US")
print(json.dumps(results, indent=2, ensure_ascii=False))

curlを使った同等のリクエストは以下の通りです:

curl -s --proxy "http://user-country-US:pass@gate.proxyhat.com:8080" \
  "https://suggestqueries.google.com/complete/search?client=chrome&q=python%20tutorial"

実装例2:シードからロングテールへの展開

1つのシードキーワードを200以上のバリエーションに展開するエクスパンダーを実装します。

import string
import itertools
from typing import Generator

# 修飾子プレフィックス(検索インテント別)
PREFIX_MODIFIERS = [
    "how to", "what is", "why", "when", "where", "who",
    "best", "top", "cheap", "free", "review", "vs",
    "buy", "price", "cost", "alternative", "near me",
]

# サフィックス修飾子
SUFFIX_MODIFIERS = [
    "tutorial", "examples", "guide", "2025", "2026",
    "alternatives", "pricing", "reddit", "meaning",
    "download", "online", "course", "book",
]

def expand_seed(keyword: str) -> list[str]:
    """シードキーワードからロングテールバリエーションを生成"""
    queries = [keyword]

    # a-z サフィックス展開
    for c in string.ascii_lowercase:
        queries.append(f"{keyword} {c}")

    # プレフィックス修飾子
    for m in PREFIX_MODIFIERS:
        queries.append(f"{m} {keyword}")

    # サフィックス修飾子
    for m in SUFFIX_MODIFIERS:
        queries.append(f"{keyword} {m}")

    return queries

def collect_longtail(keyword: str, country: str = "US") -> set[str]:
    """展開したクエリすべてのAutocompleteを収集"""
    all_suggestions = set()
    queries = expand_seed(keyword)
    print(f"Generated {len(queries)} seed queries")

    for i, q in enumerate(queries):
        suggestions = fetch_autocomplete(q, country=country)
        all_suggestions.update(suggestions)
        print(f"[{i+1}/{len(queries)}] '{q}' -> {len(suggestions)} suggestions")
        # 礼儀正しいスロットリング
        time.sleep(random.uniform(0.5, 1.5))

    return all_suggestions

# 実行例
all_kw = collect_longtail("python", country="US")
print(f"\nTotal unique keywords: {len(all_kw)}")

1シードから約200クエリを生成し、各クエリで平均10件のサジェストが返るため、理論上は最大2000件のユニークキーワードが得られます。実際には重複が発生するため、ユニーク数は500〜1500件程度になります。

実装例3:PlaywrightでPAAアコーディオンを展開して質問を抽出

PAAはJavaScriptで動的にレンダリングされるため、httpxでは取得できません。Playwrightを使ってブラウザを起動し、アコーディオンをクリックして展開します。

from playwright.sync_api import sync_playwright
import time

def extract_paa_questions(
    keyword: str,
    country: str = "US",
    max_clicks: int = 25,
) -> list[dict]:
    """
    Google検索結果からPAA質問と回答出典を抽出する
    戻り値: [{"question": str, "answer_source": str}, ...]
    """
    proxy_url = f"http://user-country-{country}:pass@gate.proxyhat.com:8080"
    paa_data = []

    with sync_playwright() as p:
        browser = p.chromium.launch(
            proxy={"server": proxy_url},
            headless=True,
        )
        context = browser.new_context(
            locale="en-US",
            user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
        )
        page = context.new_page()

        try:
            page.goto(f"https://www.google.com/search?q={keyword}", wait_until="domcontentloaded")
            page.wait_for_timeout(2000)

            for click_round in range(max_clicks):
                # PAA質問要素を取得
                paa_elements = page.query_selector_all(
                    'div[jsname="FW5kfd"] div[role="button"] span'
                )
                if not paa_elements:
                    # フォールバック: 別のセレクタ
                    paa_elements = page.query_selector_all(
                        '[jsname="CQsKde"] div[role="button"]'
                    )

                if click_round < len(paa_elements):
                    # クリックして展開
                    paa_elements[click_round].click()
                    page.wait_for_timeout(1000)

                    # 展開後に全PAA質問を再取得
                    all_questions = page.query_selector_all(
                        'div[jsname="FW5kfd"] div[role="button"] > span'
                    )

                    # 回答の出典リンクを取得
                    source_links = page.query_selector_all(
                        'div[jsname="FW5kfd"] a[href]'
                    )

                    for q_el in all_questions:
                        q_text = q_el.inner_text().strip()
                        if q_text and q_text not in [d["question"] for d in paa_data]:
                            paa_data.append({
                                "question": q_text,
                                "answer_source": "",
                            })

                    # 出典リンクをマッピング
                    for link_el in source_links:
                        href = link_el.get_attribute("href")
                        text = link_el.inner_text().strip()
                        if href and paa_data:
                            paa_data[-1]["answer_source"] = href

        except Exception as e:
            print(f"Error during PAA extraction: {e}")
        finally:
            browser.close()

    return paa_data

# 実行例
paa_results = extract_paa_questions("python web scraping", country="US", max_clicks=20)
for item in paa_results[:10]:
    print(f"Q: {item['question']}")
    print(f"  Source: {item['answer_source']}")

実装例4:再帰的PAA展開でネストされた質問を取得

PAAの質問を次のシードとして再帰的に検索することで、質問の階層ツリーを構築します。

def recursive_paa_expansion(
    seed_keyword: str,
    country: str = "US",
    max_depth: int = 3,
    max_questions: int = 300,
) -> list[dict]:
    """
    PAA質問を再帰的に展開し、深い質問ツリーを構築する
    """
    all_questions = []
    seen = set()
    queue = [(seed_keyword, 0)]  # (keyword, depth)

    while queue and len(all_questions) < max_questions:
        current_kw, depth = queue.pop(0)
        if depth > max_depth or current_kw in seen:
            continue
        seen.add(current_kw)

        print(f"[depth={depth}] Expanding: {current_kw} ({len(all_questions)}/{max_questions})")

        paa_items = extract_paa_questions(current_kw, country=country, max_clicks=10)

        for item in paa_items:
            q = item["question"]
            if q and q not in seen:
                all_questions.append({
                    "question": q,
                    "answer_source": item.get("answer_source", ""),
                    "seed": current_kw,
                    "depth": depth,
                })
                # 各質問を次の展開キューに追加
                queue.append((q, depth + 1))

        # 礼儀正しいスロットリング
        time.sleep(random.uniform(1.0, 2.5))

    return all_questions

# 実行例
expanded = recursive_paa_expansion(
    "python web scraping",
    country="US",
    max_depth=3,
    max_questions=300,
)
print(f"\nTotal PAA questions collected: {len(expanded)}")

実装例5:重複排除・インテントクラスタリング・CSV出力

収集したAutocompleteとPAAのキーワードを統合し、検索インテントでクラスタリングしてCSVに出力します。

import csv
from collections import defaultdict

def cluster_by_intent(keywords: list[str]) -> dict[str, list[str]]:
    """キーワードを検索インテントでクラスタリング"""
    intent_rules = {
        "informational": ["what", "how", "why", "when", "who", "where", "which", "guide", "tutorial", "meaning"],
        "commercial": ["best", "top", "review", "vs", "compare", "alternative", "vs"],
        "transactional": ["buy", "price", "cost", "cheap", "discount", "deal", "pricing", "free"],
        "navigational": ["login", "sign up", "official", "download", "app"],
    }

    clusters = defaultdict(list)
    for kw in keywords:
        kw_lower = kw.lower().strip()
        matched = "other"
        for intent, prefixes in intent_rules.items():
            if any(kw_lower.startswith(p) or f" {p} " in kw_lower for p in prefixes):
                matched = intent
                break
        clusters[matched].append(kw)

    return dict(clusters)

def export_to_csv(
    autocomplete_kw: set[str],
    paa_items: list[dict],
    filename: str = "keyword_research.csv",
):
    """キーワード調査結果をCSVに出力"""
    # Autocompleteキーワード
    all_kw = list(autocomplete_kw)
    # PAA質問も追加
    paa_questions = [item["question"] for item in paa_items if item["question"]]
    all_kw.extend(paa_questions)

    # 重複排除
    unique_kw = list(set(all_kw))

    # インテントクラスタリング
    clusters = cluster_by_intent(unique_kw)

    with open(filename, "w", newline="", encoding="utf-8-sig") as f:
        writer = csv.writer(f)
        writer.writerow(["intent", "keyword", "source", "answer_source"])

        # Autocomplete由来
        for kw in autocomplete_kw:
            intent = "other"
            for i, kws in clusters.items():
                if kw in kws:
                    intent = i
                    break
            writer.writerow([intent, kw, "autocomplete", ""])

        # PAA由来
        for item in paa_items:
            q = item["question"]
            intent = "other"
            for i, kws in clusters.items():
                if q in kws:
                    intent = i
                    break
            writer.writerow([intent, q, "paa", item.get("answer_source", "")])

    print(f"Exported {len(unique_kw)} unique keywords to {filename}")
    print(f"Cluster breakdown: {dict((k, len(v)) for k, v in clusters.items())}")

# 実行例
export_to_csv(all_kw, expanded, filename="python_keyword_research.csv")

並行処理でスループットを向上させる

Autocompleteリクエストは独立しているため、async/awaitで並行処理することでスループットを大幅に向上できます。ただし、Googleのレート制限を考慮し、同時接続数は50〜100程度に抑えるべきです。

import asyncio
import httpx

async def fetch_autocomplete_async(
    client: httpx.AsyncClient,
    keyword: str,
    semaphore: asyncio.Semaphore,
) -> list[str]:
    """非同期Autocomplete取得"""
    async with semaphore:
        try:
            resp = await client.get(
                "https://suggestqueries.google.com/complete/search",
                params={"client": "chrome", "q": keyword},
            )
            resp.raise_for_status()
            data = resp.json()
            return [s[0] for s in data[1]]
        except Exception as e:
            print(f"Failed for '{keyword}': {e}")
            return []

async def batch_autocomplete(
    keywords: list[str],
    country: str = "US",
    max_concurrent: int = 50,
) -> dict[str, list[str]]:
    """バッチ非同期でAutocompleteを取得"""
    proxy_url = f"http://user-country-{country}:pass@gate.proxyhat.com:8080"
    semaphore = asyncio.Semaphore(max_concurrent)
    results = {}

    async with httpx.AsyncClient(
        proxy=proxy_url,
        timeout=15.0,
        headers={"User-Agent": "Mozilla/5.0"},
    ) as client:
        tasks = [fetch_autocomplete_async(client, kw, semaphore) for kw in keywords]
        responses = await asyncio.gather(*tasks)
        for kw, suggestions in zip(keywords, responses):
            results[kw] = suggestions

    return results

# 実行例
async def main():
    seeds = expand_seed("python")[:50]  # 最初の50クエリでテスト
    results = await batch_autocomplete(seeds, country="US", max_concurrent=50)
    total = sum(len(v) for v in results.values())
    print(f"Collected {total} suggestions from {len(seeds)} queries")

asyncio.run(main())

倫理的配慮とベストプラクティス

AutocompleteとPAAのデータは公開されている提案データですが、Googleの利用規約と robots.txt を尊重すべきです。大規模にスクレイピングする場合は、以下のベストプラクティスに従ってください。

  • スロットリング: リクエスト間に0.5〜2秒のランダムな待機を入れる。50並列でも1秒あたり50リクエスト以下に抑える。
  • 公式APIの優先: 商用利用や大規模データ収集では、Google Custom Search APIや公式キーワード調査ツールの使用を検討する。
  • データの最小化: 必要なキーワードデータのみを収集し、個人情報や不要なメタデータを保存しない。
  • キャッシュの活用: 同じキーワードの再スクレイピングを避けるため、結果をキャッシュする。
  • residential proxyの適切な使用: レート制限回避が目的であり、コンテンツの不正アクセスやToS違反を目的としない。

ProxyHatのプロキシプランと料金については 料金ページ を参照してください。SERP追跡の用途については SERP追跡のユースケース も参考になります。技術的な詳細は ProxyHatドキュメント をご覧ください。

Key Takeaways

重要なポイント:

  • Autocomplete(client=chromeでJSON取得)、PAA(Playwrightでアコーディオン展開)、Related Searchesの3ソースを組み合わせることで、無料で包括的なキーワードマップを構築できる。
  • 1シードからa-z展開+修飾子プレフィックスで200以上のバリエーションを生成し、各クエリで平均10件のサジェストが得られるため、1シードあたり最大2000件のキーワード候補が得られる。
  • 再帰的PAA展開により、質問の階層ツリーを深掘りでき、FAQページやコンテンツ企画に直結する質問キーワードを大量に発掘できる。
  • residential proxyのジオターゲティング(-country-DE-city-berlin)により、ロケールバイアスを回避し、目的の国・都市のローカライズされた提案データを取得できる。
  • 並行処理でスループットを向上させる際は、同時接続数を50〜100に抑え、0.5〜2秒のランダムスロットリングを組み合わせる。
  • 収集したキーワードはインテント別にクラスタリングし、CSV出力することで、コンテンツ戦略とFAQ企画に直接活用できる。

このガイドの内容は、公開されているGoogleの提案データを活用するものです。本番環境での大規模スクレイピングを行う前に、必ずGoogleの利用規約と該当管轄区域の法律を確認してください。

始める準備はできましたか?

AIフィルタリングで148か国以上、5,000万以上のレジデンシャルIPにアクセス。

料金を見るレジデンシャルプロキシ
← ブログに戻る