抓取Google People Also Ask与自动补全做关键词研究：Python实战指南

关键词研究是SEO的基石，但大多数团队只依赖付费工具，忽略了Google搜索结果页本身就是一个巨大的关键词金矿。本指南将带你用 抓取Google People Also Ask与自动补全做关键词研究 的方法，从三个免费数据源中提取高意图关键词，并用ProxyHat住宅代理解决地理定位和封锁问题。

为什么选择抓取Google People Also Ask与自动补全做关键词研究

Google每天处理超过 85亿次 搜索查询（根据 Google官方How Search Works页面），其中自动补全建议、People Also Ask（PAA）和相关搜索三个模块直接反映了用户的真实搜索意图。与第三方关键词工具相比，这些数据具有以下优势：

• 实时性：自动补全反映当前热门搜索趋势，延迟通常在数小时内
• 意图明确：PAA问题天然是问句形式，适合FAQ页面和内容规划
• 零成本：不需要购买Ahrefs、Semrush等工具的API订阅（通常 $99-$449/月）
• 地理精确：通过代理可以获取特定城市级别的搜索建议

三大关键词金矿详解

数据源	端点/位置	数据格式	搜索意图映射
自动补全	suggestqueries.google.com/complete/search?client=chrome&q=	JSON（Chrome客户端）	导航型 + 信息型
People Also Ask	SERP中的可展开手风琴组件	HTML（需渲染）	信息型 + 商业型
相关搜索	SERP底部	HTML文本链接	交易型 + 商业型

自动补全端点返回JSON格式数据，是最容易抓取的。PAA需要浏览器渲染来展开手风琴组件，因为每个问题的展开会触发新的AJAX请求并揭示嵌套问题。相关搜索则直接嵌入在SERP HTML中，用简单的解析即可提取。

技术背景：为什么快速抓取会触发封锁

Google对自动补全和SERP请求实施了严格的 每IP速率限制。根据 MDN关于HTTP 429的文档，当服务器检测到异常请求频率时会返回429状态码。Google的具体表现包括：

• 同一IP在短时间内发送超过 约100次/分钟 的自动补全请求会触发临时封锁
• SERP抓取的阈值更低，大约 20-30次/分钟 就可能触发验证码挑战
• 封锁持续时间从几分钟到24小时不等，重复违规会导致更长时间

此外，自动补全结果具有地理偏差。Google会根据请求IP的地理位置返回不同的建议词。如果你用美国数据中心IP抓取，你只能获得美国市场的关键词，无法获取德国、日本等市场的本地化搜索建议。这就是为什么需要住宅代理配合地理定位的原因。

种子词到长尾词的扩展策略

自动补全的a-z前缀扩展法

一个种子关键词可以通过添加字母前缀和修饰词扩展出数百个长尾词。核心思路是对种子词加上 a 到 z 的每个字母作为前缀或后缀，再加上常见修饰词（如"how"、"what"、"best"、"vs"等），批量请求自动补全端点。

PAA递归扩展

PAA的独特之处在于每个手风琴展开后会生成新的相关问题。一个初始PAA区块通常包含4个问题，点击展开一个后会出现一个新的替代问题。通过递归展开，4个初始问题可以扩展到 20-50个 甚至更多相关问题，形成深度问题树。

实战代码：用httpx和ProxyHat抓取自动补全

以下示例展示如何用Python的httpx库配合ProxyHat住宅代理抓取Google自动补全建议。我们同时展示原始代理配置和ProxyHat SDK的使用方式。

方式一：直接使用httpx + 原始代理URL

import httpx
import json
import time
import logging
from urllib.parse import quote

logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')
logger = logging.getLogger(__name__)

def fetch_autocomplete(keyword: str, country: str = "US", retries: int = 3) -> list[str]:
    """抓取Google自动补全建议词列表"""
    # ProxyHat住宅代理 - 按国家定位
    proxy_url = f"http://user-country-{country}:YOUR_PASSWORD@gate.proxyhat.com:8080"
    
    endpoint = "https://suggestqueries.google.com/complete/search"
    params = {
        "client": "chrome",
        "q": keyword,
        "gl": country.lower(),
        "hl": "en"
    }
    
    for attempt in range(retries):
        try:
            with httpx.Client(proxy=proxy_url, timeout=10.0) as client:
                resp = client.get(endpoint, params=params)
                resp.raise_for_status()
                data = resp.json()
                # Chrome客户端返回格式: [query, [suggestions...], ...]
                if isinstance(data, list) and len(data) > 1:
                    suggestions = data[1]
                    logger.info(f"[{country}] '{keyword}' -> {len(suggestions)} 条建议")
                    return suggestions
                return []
        except httpx.HTTPStatusError as e:
            if e.response.status_code == 429:
                wait = 2 ** attempt
                logger.warning(f"429限流，等待{wait}秒后重试...")
                time.sleep(wait)
            else:
                logger.error(f"HTTP错误: {e}")
                raise
        except Exception as e:
            logger.error(f"请求失败: {e}")
            time.sleep(1)
    return []

# 示例：抓取"python"的自动补全
results = fetch_autocomplete("python", country="US")
print(json.dumps(results, indent=2, ensure_ascii=False))

方式二：使用ProxyHat SDK + 异步并发

import httpx
import asyncio
import logging
from itertools import product

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class ProxyHatAutocompleteScraper:
    def __init__(self, password: str, max_concurrent: int = 10):
        self.password = password
        self.semaphore = asyncio.Semaphore(max_concurrent)
        
    def _proxy_for_country(self, country: str, city: str = None) -> str:
        """生成ProxyHat代理URL，支持国家和城市级定位"""
        username = f"user-country-{country}"
        if city:
            username += f"-city-{city.lower().replace(' ', '')}"
        return f"http://{username}:{self.password}@gate.proxyhat.com:8080"
    
    async def fetch_one(self, client: httpx.AsyncClient, keyword: str, 
                        country: str, city: str = None) -> list[str]:
        async with self.semaphore:
            proxy = self._proxy_for_country(country, city)
            params = {"client": "chrome", "q": keyword, 
                      "gl": country.lower(), "hl": "en"}
            try:
                async with httpx.AsyncClient(proxy=proxy, timeout=10.0) as c:
                    resp = await c.get(
                        "https://suggestqueries.google.com/complete/search",
                        params=params
                    )
                    resp.raise_for_status()
                    data = resp.json()
                    return data[1] if isinstance(data, list) and len(data) > 1 else []
            except Exception as e:
                logger.error(f"抓取失败 [{keyword}|{country}]: {e}")
                return []
    
    async def expand_seed(self, seed: str, country: str = "US") -> list[str]:
        """用a-z前缀扩展种子词"""
        prefixes = list("abcdefghijklmnopqrstuvwxyz")
        modifiers = ["", "how ", "what ", "best ", "vs ", "why ", " "]
        
        keywords = []
        for mod in modifiers:
            for prefix in prefixes:
                keywords.append(f"{seed} {prefix}")
                if mod:
                    keywords.append(f"{mod}{seed} {prefix}")
        
        # 去重
        keywords = list(set(keywords))
        logger.info(f"种子词 '{seed}' 扩展为 {len(keywords)} 个查询")
        
        tasks = [self.fetch_one(None, kw, country) for kw in keywords]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        all_suggestions = set()
        for r in results:
            if isinstance(r, list):
                all_suggestions.update(r)
        
        return list(all_suggestions)

# 运行示例
async def main():
    scraper = ProxyHatAutocompleteScraper(password="YOUR_PASSWORD")
    # 抓取德国柏林的自动补全
    suggestions = await scraper.expand_seed("python", country="DE")
    print(f"共获取 {len(suggestions)} 个建议词")
    for s in suggestions[:20]:
        print(f"  - {s}")

asyncio.run(main())

用Playwright抓取People Also Ask问题

PAA组件是交互式的手风琴，需要浏览器渲染才能展开并提取嵌套问题。以下示例使用Playwright配合ProxyHat代理来抓取PAA问题及其引用来源。

from playwright.async_api import async_playwright
import asyncio
import json
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class PAAScraper:
    def __init__(self, password: str):
        self.password = password
    
    def _proxy_config(self, country: str, city: str = None) -> dict:
        """生成ProxyHat代理配置"""
        username = f"user-country-{country}"
        if city:
            username += f"-city-{city.lower().replace(' ', '')}"
        return {
            "server": "http://gate.proxyhat.com:8080",
            "username": username,
            "password": self.password
        }
    
    async def scrape_paa(self, query: str, country: str = "US", 
                         max_expansions: int = 15) -> list[dict]:
        """抓取PAA问题及其引用来源"""
        proxy_config = self._proxy_config(country)
        paa_results = []
        
        async with async_playwright() as p:
            browser = await p.chromium.launch(
                headless=True,
                proxy=proxy_config
            )
            context = await browser.new_context(
                locale="en-US",
                viewport={"width": 1280, "height": 800}
            )
            page = await context.new_page()
            
            url = f"https://www.google.com/search?q={query}&gl={country.lower()}"
            await page.goto(url, wait_until="networkidle", timeout=30000)
            
            # 等待PAA区块加载
            try:
                await page.wait_for_selector(
                    "div[jsname=Sk3O9d]", timeout=10000
                )
            except Exception:
                logger.warning("未找到PAA区块")
                await browser.close()
                return []
            
            for i in range(max_expansions):
                # 查找未展开的PAA问题
                questions = await page.query_selector_all(
                    "div[jsname=Sk3O9d] div[jsname=Cpkphce] "
                    "span[jsname=DCQmZe]"
                )
                
                if not questions:
                    logger.info(f"第{i}轮：无更多PAA问题")
                    break
                
                # 点击第一个未展开的问题
                try:
                    await questions[0].click()
                    await page.wait_for_timeout(1500)
                except Exception as e:
                    logger.warning(f"点击失败: {e}")
                    continue
                
                # 提取已展开的问题和答案来源
                expanded = await page.query_selector_all(
                    "div[jsname=Sk3O9d] div[role='heading']"
                )
                for item in expanded:
                    text = await item.inner_text()
                    if text.endswith("?") and text not in [r["question"] for r in paa_results]:
                        # 尝试获取答案来源链接
                        parent = await item.evaluate_handle(
                            "el => el.closest('div[jsname=Sk3O9d]')"
                        )
                        link_el = await parent.query_selector("a[href]")
                        source_url = await link_el.get_attribute("href") if link_el else None
                        
                        paa_results.append({
                            "question": text,
                            "source": source_url,
                            "query": query,
                            "country": country
                        })
                        logger.info(f"提取PAA: {text[:60]}...")
            
            await browser.close()
        
        return paa_results

# 运行示例
async def main():
    scraper = PAAScraper(password="YOUR_PASSWORD")
    results = await scraper.scrape_paa(
        "what is python programming", country="US", max_expansions=20
    )
    print(json.dumps(results, indent=2, ensure_ascii=False))
    print(f"\n共获取 {len(results)} 个PAA问题")

asyncio.run(main())

抓取相关搜索

相关搜索是SERP底部的文本链接，提取相对简单，但仍需代理来避免封锁。

import httpx
from selectolax.parser import HTMLParser
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def fetch_related_searches(query: str, country: str = "US") -> list[str]:
    """抓取Google相关搜索"""
    proxy_url = f"http://user-country-{country}:YOUR_PASSWORD@gate.proxyhat.com:8080"
    
    url = "https://www.google.com/search"
    params = {"q": query, "gl": country.lower(), "hl": "en"}
    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": "en-US,en;q=0.9"
    }
    
    try:
        with httpx.Client(proxy=proxy_url, timeout=15.0, headers=headers) as client:
            resp = client.get(url, params=params)
            resp.raise_for_status()
            
            tree = HTMLParser(resp.text)
            related = []
            # 相关搜索通常在底部区块
            for node in tree.css("a[href*='/search?q=']"):
                text = node.text(strip=True)
                href = node.attributes.get("href", "")
                # 过滤掉导航类链接，只保留相关搜索
                if text and len(text) > 3 and "q=" in href and text != query:
                    related.append(text)
            
            # 去重
            related = list(dict.fromkeys(related))
            logger.info(f"[{country}] '{query}' -> {len(related)} 个相关搜索")
            return related
    except httpx.HTTPStatusError as e:
        logger.error(f"HTTP {e.response.status_code}: {e}")
        return []
    except Exception as e:
        logger.error(f"请求失败: {e}")
        return []

# 示例
related = fetch_related_searches("python tutorial", country="US")
for r in related[:15]:
    print(f"  - {r}")

去重、聚类与CSV导出

将三个数据源的结果合并后，需要进行去重和意图聚类，才能用于内容规划。

import csv
import re
from collections import defaultdict

class KeywordProcessor:
    INTENT_PATTERNS = {
        "informational": [r"^how", r"^what", r"^why", r"^when", r"^who", 
                         r"^where", r"^which", r"\?$"],
        "commercial": [r"best", r"top", r"review", r"vs", r"compare", 
                       r"alternative", r"vs"],
        "transactional": [r"buy", r"price", r"cost", r"cheap", 
                          r"discount", r"deal", r"order", r"shop"],
        "navigational": [r"login", r"sign up", r"download", r"official"]
    }
    
    @classmethod
    def classify_intent(cls, keyword: str) -> str:
        kw_lower = keyword.lower()
        for intent, patterns in cls.INTENT_PATTERNS.items():
            if any(re.search(p, kw_lower) for p in patterns):
                return intent
        return "informational"  # 默认归为信息型
    
    @staticmethod
    def dedupe(keywords: list[str]) -> list[str]:
        """去重并保留原始大小写"""
        seen = set()
        result = []
        for kw in keywords:
            key = kw.lower().strip()
            if key not in seen:
                seen.add(key)
                result.append(kw.strip())
        return result
    
    @classmethod
    def cluster_by_intent(cls, keywords: list[str]) -> dict[str, list[str]]:
        """按搜索意图聚类"""
        clusters = defaultdict(list)
        for kw in cls.dedupe(keywords):
            intent = cls.classify_intent(kw)
            clusters[intent].append(kw)
        return dict(clusters)
    
    @staticmethod
    def export_to_csv(keywords: list[dict], filepath: str) -> None:
        """导出关键词到CSV文件"""
        with open(filepath, "w", newline="", encoding="utf-8") as f:
            writer = csv.DictWriter(
                f, fieldnames=["keyword", "source", "intent", "country"]
            )
            writer.writeheader()
            for row in keywords:
                writer.writerow(row)
        print(f"已导出 {len(keywords)} 条关键词到 {filepath}")

# 使用示例
all_keywords = [
    {"keyword": kw, "source": "autocomplete", "intent": "", "country": "US"}
    for kw in autocomplete_results
] + [
    {"keyword": q["question"], "source": "paa", "intent": "", "country": q["country"]}
    for q in paa_results
] + [
    {"keyword": kw, "source": "related", "intent": "", "country": "US"}
    for kw in related_results
]

# 分类意图
for row in all_keywords:
    row["intent"] = KeywordProcessor.classify_intent(row["keyword"])

# 导出
KeywordProcessor.export_to_csv(all_keywords, "keywords.csv")

# 查看聚类结果
clusters = KeywordProcessor.cluster_by_intent(
    [r["keyword"] for r in all_keywords]
)
for intent, kws in clusters.items():
    print(f"\n{intent} ({len(kws)}个):")
    for kw in kws[:5]:
        print(f"  - {kw}")

ProxyHat代理配置详解

以上所有示例都使用了ProxyHat住宅代理。配置要点如下：

• HTTP网关：gate.proxyhat.com:8080
• SOCKS5网关：gate.proxyhat.com:1080（适用于需要更底层代理的场景）
• 国家定位：用户名中添加 -country-DE
• 城市定位：用户名中添加 -country-DE-city-berlin
• 粘性会话：用户名中添加 -session-abc123 保持同一IP

对于PAA抓取，建议使用粘性会话，因为Playwright需要在一个IP上完成整个浏览器会话。对于自动补全，每次请求轮换IP即可。更多配置细节请参考 ProxyHat官方文档。

想了解不同代理类型的对比和定价，请访问 ProxyHat定价页面。如果你需要更多Web抓取场景的参考，可以阅读我们的 Web抓取用例 和 SERP追踪用例。完整的代理位置列表见 ProxyHat位置页面。

常见错误与边缘情况

• 429限流：降低并发数至5-10，添加指数退避重试逻辑
• 空结果：检查代理是否过期或被封锁，更换会话ID
• 地理偏差：确保 gl 参数与代理国家一致，否则结果会混乱
• PAA不出现：某些查询不触发PAA区块，尝试添加问句形式的关键词
• 编码问题：自动补全端点返回UTF-8编码，确保正确处理非ASCII字符
• 验证码挑战：如果频繁触发，降低请求频率并考虑使用更长的粘性会话

伦理与合规考量

自动补全建议和PAA数据本质上是公开的搜索建议数据，但大规模抓取仍需注意以下几点：

• 尊重速率限制：保持合理请求频率，建议每IP不超过 60次/分钟
• 优先官方API：大规模关键词研究应考虑Google官方API或第三方API服务
• 遵守robots.txt：检查 suggestqueries.google.com/robots.txt 的规则
• 数据用途：仅用于SEO研究和内容规划，不要用于垃圾信息发送
• GDPR合规：如果处理涉及个人数据的结果，需遵守相关隐私法规

实践建议：在生产环境中，将请求间隔设置为 1-2秒，使用断路器模式在连续失败超过5次时暂停抓取，并记录所有请求的响应状态码以便后续分析。

关键要点总结

• Google自动补全、PAA和相关搜索是三个免费关键词金矿，分别映射不同的搜索意图
• a-z前缀扩展法可以将一个种子词扩展为数百个长尾词查询
• PAA递归展开能生成深度问题树，适合FAQ页面规划
• 住宅代理配合国家/城市级地理定位是获取本地化建议的关键
• ProxyHat的 gate.proxyhat.com:8080 网关支持用户名中直接配置地理定位和会话参数
• 始终实施速率限制、重试逻辑和断路器模式来保证抓取稳定性
• 去重和意图聚类是将原始数据转化为可操作关键词策略的关键步骤