如何使用代理抓取谷歌购物数据

抓取谷歌购物是指从谷歌购物结果中大规模采集结构化商品数据，例如标题、价格、卖家、评分和配送信息。由于这些结果受到严格保护，且会随地区变化，要可靠地完成这项工作，就需要轮换IP并正确进行地理定位，而不能只在单一连接上运行一个抓取器。

本指南以代码为主线，依次介绍地理定位请求、通过国家定向代理路由请求、解析谷歌刻意混淆的商品卡片、分页、遇到速率限制时的重试与退避、处理同意对话框的无头浏览器备用方案、可导出CSV的清晰数据模型、价格历史追踪器以及监控计划。同时也会如实说明反自动化机制的现实、自建抓取器与付费SERP API之间的取舍，以及相关法律边界。

DataImpulse是一家注重合规与道德采购的代理服务商，在195个国家提供超过9000万个住宅、移动和数据中心IP地址。它采用按量付费模式，每GB起价1美元，流量永不过期，被广泛用于网页抓取、广告验证、价格监控、市场调研和多账号管理。

可以从谷歌购物抓取哪些数据?

谷歌购物展示的商品列表包含多个结构化字段，对定价和市场调研很有用。每条结果通常带有一组一致的属性，可以提取并归一化为数据行。

• 商品详情: 标题、描述摘要、品牌、型号和商品图片URL。
• 价格: 标价、货币，有时还包括不同卖家之间的价格区间。
• 卖家: 商家名称，以及商品详情页中销售同款商品的其他卖家列表。
• 评分和评论: 综合星级评分和评论数量(如有)。
• 库存和配送: 库存状态和因地区而异的配送说明。

每当谷歌更新其页面布局时，具体字段就会发生变化，因此任何解析器都需要容忍缺失值和变化的标记结构。应将每个字段都视为可选项，并在提取后验证类型，而不是假设评分或卖家信息始终存在。由于同一商品在不同国家可能显示不同价格，从一开始就应围绕(商品、国家、时间戳)这一键设计数据结构，而不是事后再加入地理维度。

如何向谷歌购物发送地理定位请求?

要发送地理定位请求，可携带购物垂直搜索标志和地区参数访问谷歌搜索端点，并模拟真实购物者会发送的IP和参数信号。最重要的两个参数是 gl (国家，例如 gl=de 表示德国)，以及 hl (界面语言，例如 hl=de)。第三个有用的参数是类似 location 的位置术语，可将其加入查询上下文，不过 gl 加上匹配的出口IP通常已能满足大多数需求。

仅设置参数是不够的，因为谷歌还会读取请求的IP来决定显示哪些价格和卖家。下面这个最简请求发送真实的请求头和地区参数组合。它刻意不使用代理，以便你在添加IP层之前先看到基础调用。

import requests

def build_params(query, country="us", lang="en", start=0):
    return {
        "q": query,
        "tbm": "shop",   # shopping vertical
        "gl": country,    # country bias, e.g. de, gb, fr
        "hl": lang,       # interface language
        "num": 40,        # results per page (soft cap)
        "start": start,   # pagination offset
    }

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

resp = requests.get(
    "https://www.google.com/search",
    params=build_params("wireless headphones", country="us", lang="en"),
    headers=HEADERS,
    timeout=30,
)
print(resp.status_code, len(resp.text))

保持 Accept-Language 请求头与 hl 一致。用英语请求头请求德语价格页面是一种不匹配，这既显得不自然，还可能返回错误的货币。当你扩大规模时，这个基础函数就成为决定地区设置的唯一位置，从而保持所有后续调用的一致性。

如何通过国家定向代理路由请求?

通过国家定向代理路由请求的方式是，将 requests 指向单一网关，并将目标国家编码在代理用户名中。使用DataImpulse时网关是 gw.dataimpulse.com:823，在用户名末尾添加 __cr.us或__cr.de 即可选择出口国家。国家定位包含在基础价格内，而州、城市、邮编和ASN定位则是付费附加项。如果你对代理身份验证还不熟悉，我们关于 代理身份验证指南详细介绍了用户名和密码的机制。

import requests

USER = "your_login"
PASS = "your_password"
GATEWAY = "gw.dataimpulse.com:823"

def proxy_for(country):
    # __cr. pins the exit IP to that country
    cred = f"{USER}__cr.{country}:{PASS}"
    url = f"http://{cred}@{GATEWAY}"
    return {"http": url, "https": url}

resp = requests.get(
    "https://www.google.com/search",
    params=build_params("wireless headphones", country="de", lang="de"),
    headers=HEADERS,
    proxies=proxy_for("de"),
    timeout=30,
)
print(resp.status_code, len(resp.text))

关键原则是同时切换代理国家与 gl/hl 参数。要构建跨市场价格比较，需遍历国家列表，并让出口IP与地区设置保持同步，使每个响应都反映单一、一致的购物者所在位置。

MARKETS = [
    {"country": "us", "lang": "en"},
    {"country": "de", "lang": "de"},
    {"country": "gb", "lang": "en"},
    {"country": "fr", "lang": "fr"},
]

def fetch_market(query, market):
    c, lang = market["country"], market["lang"]
    return requests.get(
        "https://www.google.com/search",
        params=build_params(query, country=c, lang=lang),
        headers={**HEADERS, "Accept-Language": f"{lang};q=0.9"},
        proxies=proxy_for(c),
        timeout=30,
    )

for m in MARKETS:
    r = fetch_market("wireless headphones", m)
    print(m["country"], r.status_code, len(r.text))

DataImpulse提供的是IP层，而不是托管抓取服务，因此抓取器需要由你自行维护。这里通常适合使用轮换住宅代理，因为它们来自真实消费者设备。数据中心代理价格更低，但更容易被识别；面对最难访问的目标时，则可使用带有运营商级IP的移动代理。

如何解析谷歌购物的商品卡片?

解析商品卡片时应使用容错的HTML库，并依赖稳定的文本模式，而不是脆弱的深层选择器，因为谷歌的类名是混淆的且经常变化。你今天从浏览器开发者工具复制的选择器可能几周内就会失效，因此应把解析器设计成能优雅降级，而不是在页面布局变化时崩溃。

一种稳健的做法是读取每个候选卡片，从标签结构中提取标题和图片，并用支持货币识别的正则表达式提取价格。下面的选择器只是贴近实际的示例，并非保证有效；应预期需要更新它们，并将正则表达式作为可靠的后备方案。

from bs4 import BeautifulSoup
import re

# Currency-aware price matcher: symbol optional, thousands and decimals tolerated.
PRICE_RE = re.compile(r"(?:[$€£]|USD|EUR|GBP)\s?\d[\d.,]*")

def parse_cards(html):
    soup = BeautifulSoup(html, "html.parser")
    rows = []
    # These class names ROTATE; treat them as hints, keep the fallbacks.
    candidates = soup.select("div.sh-dgr__content, div.sh-dlr__list-result, div[data-docid]")
    if not candidates:
        candidates = soup.find_all("div")  # last-resort sweep
    for card in candidates:
        text = card.get_text(" ", strip=True)
        price_match = PRICE_RE.search(text)
        if not price_match or len(text) > 400:
            continue
        img = card.find("img")
        link = card.find("a", href=True)
        rows.append({
            "title": _first_text(card, ["h3", "h4"]),
            "price_raw": price_match.group(0),
            "image": img.get("src") if img else None,
            "url": link["href"] if link else None,
        })
    return rows

def _first_text(card, tags):
    for t in tags:
        el = card.find(t)
        if el and el.get_text(strip=True):
            return el.get_text(strip=True)
    return None

CURRENCY = {"$": "USD", "€": "EUR", "£": "GBP"}

def normalize_price(raw):
    if not raw:
        return None, None
    symbol = next((s for s in CURRENCY if s in raw), None)
    code = CURRENCY.get(symbol, "USD")
    digits = re.sub(r"[^\d.,]", "", raw).replace(",", "")
    try:
        return float(digits), code
    except ValueError:
        return None, code

由于混淆标记在这里是常态，一些团队转而使用渲染后的DOM。关于静态HTML解析与渲染页面解析之间更广泛的取舍，请参阅我们的指南 抓取动态网页。

当谷歌返回429时，如何分页和退避?

分页时需推进 start 偏移量，步长应与每页大小一致；应对速率限制时，则应，将HTTP 429(以及空响应或验证码响应)视为需要等待并通过新IP重试的信号。谷歌会激进地限制重复的自动化请求，因此分页和退避应放在同一个循环中处理。

持续翻页，直到某次请求不再返回任何卡片，或达到页数上限。通过重新读取代理字典在各页之间轮换出口IP，并通过检查响应正文中的同意或验证码标记来检测软封锁，而不是仅依赖状态码。

BLOCK_MARKERS = ("unusual traffic", "/sorry/", "consent.google", "captcha")

def is_blocked(resp):
    if resp.status_code in (429, 503):
        return True
    body = resp.text.lower()
    return any(m in body for m in BLOCK_MARKERS)

def paginate(query, market, max_pages=5, page_size=40):
    all_rows = []
    for page in range(max_pages):
        params = build_params(query, market["country"], market["lang"],
                              start=page * page_size)
        resp = get_with_retry(
            "https://www.google.com/search",
            params=params, proxies=proxy_for(market["country"]),
        )
        if resp is None:
            break
        rows = parse_cards(resp.text)
        if not rows:
            break  # no more results
        all_rows.extend(rows)
    return all_rows

下面的重试辅助函数使用带抖动的指数退避，将重试分散开来，避免一批被封锁的工作进程在同一瞬间同时重试。每次尝试都会重新获取代理字典，因此轮换池会在下一次尝试时给你一个新的IP。

import time, random

def get_with_retry(url, params, proxies, max_attempts=4):
    for attempt in range(max_attempts):
        try:
            resp = requests.get(url, params=params, headers=HEADERS,
                                proxies=proxies, timeout=30)
        except requests.RequestException:
            resp = None
        if resp is not None and not is_blocked(resp):
            return resp
        # exponential backoff: 2, 4, 8 seconds, plus jitter
        wait = (2 ** (attempt + 1)) + random.uniform(0, 1.5)
        time.sleep(wait)
    return None  # exhausted; caller decides what to do

即使重试成功，也要保持请求速率合理。在各页之间设置礼貌的基础延迟，再结合轮换，对长期稳定性的帮助远胜于疯狂请求并依赖重试。 网页抓取最佳实践 相关的说明中更详细地介绍了请求节奏和请求头卫生。

如何使用无头浏览器渲染被封锁的页面?

当普通的HTTP请求不断遇到同意墙或JavaScript渲染的内容时，就回退到无头浏览器，让它像真实客户端一样执行页面，并在读取DOM之前关闭同意对话框。Playwright很适合这项工作，因为它支持按上下文设置代理，并具有可靠的等待机制。

下面这个脚本通过一个国家定向的DataImpulse代理启动Chromium，处理欧洲地区通常会显示的cookie或同意对话框，等待内容加载完成，并返回渲染后的HTML，以便你将其输入到同一个 parse_cards 函数中处理。

from playwright.sync_api import sync_playwright

def render_shopping(query, country="de", lang="de"):
    cred_user = f"{USER}__cr.{country}"
    params = f"?q={query.replace(' ', '+')}&tbm=shop&gl={country}&hl={lang}"
    url = "https://www.google.com/search" + params
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)
        context = browser.new_context(
            proxy={
                "server": "http://gw.dataimpulse.com:823",
                "username": cred_user,
                "password": PASS,
            },
            locale=f"{lang}-{country.upper()}",
            user_agent=HEADERS["User-Agent"],
        )
        page = context.new_page()
        page.goto(url, wait_until="domcontentloaded", timeout=45000)
        _dismiss_consent(page)
        page.wait_for_timeout(2000)
        html = page.content()
        browser.close()
        return html

def _dismiss_consent(page):
    # Consent dialogs vary by locale; try a few common accept buttons.
    for label in ("Accept all", "Alle akzeptieren", "Tout accepter",
                  "I agree", "Accept"):
        try:
            btn = page.get_by_role("button", name=label)
            if btn.count() > 0:
                btn.first.click(timeout=3000)
                page.wait_for_timeout(1000)
                return
        except Exception:
            continue

无头浏览器在CPU和内存上的开销比原始请求更大，因此应把它当作针对确实需要渲染的页面的备用方案，而不是默认路径。许多团队为了速度直接运行HTTP请求，只有在 is_blocked is_blocked不断触发时才升级到Playwright。如果你的技术栈是JavaScript而非Python，同样的模式也适用于Puppeteer；请参阅 使用JavaScript进行网页抓取.

如何组织和导出抓取到的数据?

将数据围绕一条始终包含市场和采集时间戳的稳定记录来组织，然后导出为CSV，这样输出结果便于比对、加载到电子表格或推送到数据仓库中。使用数据类可以获得类型提示，并对一行数据有一个清晰的定义。

from dataclasses import dataclass, asdict, field
from datetime import datetime, timezone
import csv

@dataclass
class Offer:
    query: str
    country: str
    title: str | None
    price: float | None
    currency: str | None
    seller: str | None = None
    url: str | None = None
    captured_at: str = field(
        default_factory=lambda: datetime.now(timezone.utc).isoformat()
    )

def to_offers(rows, query, country):
    offers = []
    for r in rows:
        price, currency = normalize_price(r.get("price_raw"))
        offers.append(Offer(
            query=query, country=country, title=r.get("title"),
            price=price, currency=currency, url=r.get("url"),
        ))
    return offers

写入CSV就变成了导出dataclass字段的简单操作。追加模式加上表头保护，可以让定时任务随时间不断扩充同一个文件，而无需重写它。

import os

def export_csv(offers, path="google_shopping.csv"):
    if not offers:
        return
    fieldnames = list(asdict(offers[0]).keys())
    write_header = not os.path.exists(path)
    with open(path, "a", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        if write_header:
            writer.writeheader()
        for o in offers:
            writer.writerow(asdict(o))

在每一行都保留国家和时间戳信息，日后就能回答诸如今天哪个市场某个SKU最便宜、或者本周价格如何变动之类的问题。这种地理与时间组合键正是下一节价格历史追踪器的基础。

如何随时间跟踪谷歌购物的价格历史?

要跟踪价格历史，可将每次观测结果按商品、国家和日期作为键，通过upsert写入小型数据库，这样重新运行抓取器只会更新当天那一行，而不会产生重复记录，同时保留之前各天的数据。SQLite足以应对数百万行数据，且无需服务器，使监控任务保持自包含。

下面的表结构使用了一个复合唯一键和一个 ON CONFLICT upsert 操作。同一天内运行两次抓取器会覆盖该(商品、国家、日期)组合的最新价格;而跨多天运行则会构建出之后可以绘制图表的时间序列。

import sqlite3

DDL = """
CREATE TABLE IF NOT EXISTS price_history (
    product   TEXT NOT NULL,
    country   TEXT NOT NULL,
    day       TEXT NOT NULL,
    price     REAL,
    currency  TEXT,
    seller    TEXT,
    captured_at TEXT,
    PRIMARY KEY (product, country, day)
);
"""

def init_db(path="prices.db"):
    conn = sqlite3.connect(path)
    conn.execute(DDL)
    conn.commit()
    return conn

def upsert_offers(conn, offers):
    sql = """
    INSERT INTO price_history
        (product, country, day, price, currency, seller, captured_at)
    VALUES (?, ?, ?, ?, ?, ?, ?)
    ON CONFLICT(product, country, day) DO UPDATE SET
        price=excluded.price,
        currency=excluded.currency,
        seller=excluded.seller,
        captured_at=excluded.captured_at;
    """
    for o in offers:
        day = o.captured_at[:10]  # YYYY-MM-DD
        conn.execute(sql, (o.title, o.country, day, o.price,
                           o.currency, o.seller, o.captured_at))
    conn.commit()

def price_trend(conn, product, country, days=14):
    cur = conn.execute(
        """SELECT day, price, currency FROM price_history
           WHERE product = ? AND country = ?
           ORDER BY day DESC LIMIT ?""",
        (product, country, days),
    )
    return cur.fetchall()

实际操作中，跨天匹配标题是最困难的部分，因为谷歌的标题在不同抓取之间会有细微差异。对于小型监控列表，建议在upsert之前，将原始标题映射到你自己掌控的规范商品ID，而不是把抓取到的标题当作稳定的键来信任。

如何安排持续的价格监控?

可将获取、解析和存储这几个步骤封装为一个入口函数，并用cron以固定节奏运行它来安排监控，这样价格就能自动刷新而无需手动运行。对于价格追踪来说，每天运行一次是明智的默认设置;更频繁的运行会同时增加成本和被封锁的风险，而对大多数商品目录而言并不会带来多少额外信号。

# run_monitor.py
def run_once(queries):
    conn = init_db()
    for query in queries:
        for market in MARKETS:
            rows = paginate(query, market, max_pages=3)
            offers = to_offers(rows, query, market["country"])
            export_csv(offers)
            upsert_offers(conn, offers)
            print(f"{query} / {market['country']}: {len(offers)} offers")
    conn.close()

if __name__ == "__main__":
    WATCHLIST = ["wireless headphones", "mechanical keyboard"]
    run_once(WATCHLIST)

然后用cron注册它。下面这条条目每天06:15运行监控任务，并记录输出以便调试，这一点很重要，因为否则计划任务型抓取器会悄无声息地失败。

# crontab -e
# minute hour day month weekday  command
15 6 * * * cd /opt/shopping && /usr/bin/python3 run_monitor.py >> monitor.log 2>&1

在任务中加入轻微的抖动(开始时随机休眠几分钟)，避免每天都在完全相同的时刻发起请求。按前文所述轮换出口国家并调节请求节奏，同时留意日志中封锁率是否上升，这是谷歌已经调整防御策略或你的模式变得过于规律的最早信号。

应选择哪种采集方式：自建、隐藏端点、SERP API还是无头浏览器?

选择依据在于你愿意自行承担多少脆弱性和工程投入，相对于你愿意为每次请求支付多少费用。没有唯一正确的答案；坦率地说，这是控制权和成本与维护负担和反机器人风险之间的权衡。

反机器人现实: 谷歌通过速率限制、验证码、同意墙、轮换的混淆标记以及行为信号积极防御这些结果。任何自建抓取器都会周期性地失效，需要持续维护。代理和退避可以减少封锁，但无法彻底消除封锁。任何承诺提供永久不会被封锁的谷歌抓取器的人都是在过度承诺。

SERP API与自建方案对比: 第三方SERP API会以JSON形式返回已解析的购物结果，并替你承担解析和解封工作，但按请求计费，存在供应商锁定，且你对实际获取的内容控制较少。使用代理自建抓取器的单次请求成本低得多，控制力也更强，但当谷歌发生变化时，解析器、重试机制和维护工作都需要由你负责。需要明确的是，DataImpulse在这些自建方案中提供的是底层代理服务，不是SERP API，也不是托管抓取服务。如果你希望完全免除解析工作，SERP API是更合适的选择；如果你想降低成本、保有控制力，并能维护抓取器，那么使用轮换代理自建抓取器在经济性上更优。隐藏JSON端点的成本可能更低，但也最脆弱，因为谷歌随时可能更改或移除它们。

抓取谷歌购物数据合法且合乎道德吗?

抓取公开可见的商品数据通常被认为风险低于访问私有或受限内容，但这并非完全没有规则约束，本文也不构成法律建议。谷歌购物上的价格和列表是公开的，但你如何采集和使用这些数据仍然很重要，具体案例应向合格律师确认。

• 服务条款: 自动化访问可能与谷歌的服务条款发生冲突，这是与版权或计算机访问法律不同的一个合同问题。
• 个人数据: 商品列表通常不属于个人数据，但应避免收集评论者姓名或其他身份标识信息，并在适用的情况下遵守GDPR。
• 速率与负载: 保持请求量合理，以免影响你所查询的服务的正常运行。
• 数据使用: 大规模重复使用受版权保护的图片或描述，其风险要高于将价格作为事实来分析。

在数据采集方面，道德实践首先取决于所使用IP的来源。DataImpulse的IP来自自愿加入并获得报酬的用户，符合GDPR要求，并提供数据处理协议，从而支撑起一条站得住脚的合规链条。关于同意、来源和负责任实践的更全面介绍，请参阅我们的指南 道德网页抓取。将合规来源的IP与礼貌的请求节奏以及狭窄、真实的数据用途结合起来，就能同时把法律风险和声誉风险控制在较低水平。

常见问题

相关指南

• 如何抓取谷歌趋势数据
• 如何抓取酒店列表数据
• 抓取动态网页

使用可靠的代理开始抓取谷歌购物数据

如果你准备在多个市场采集谷歌购物数据，DataImpulse可提供轮换和固定的住宅、移动及数据中心IP，包含国家定位功能，每GB起价1美元。 创建账户 并让你的第一个抓取器通过一个干净的IP池进行路由。