scrape dynamic web pages

要抓取动态网页,就必须处理首次请求下载的HTML中不存在的内容。现代网站使用JavaScript在浏览器中构建商品网格、价格、评论和信息流,因此普通的HTTP请求往往只会返回一个本应承载数据的空壳。

本指南将全面介绍Python中的工作流程:先通过比对原始响应与渲染后的DOM识别动态内容,再按成本和复杂度由低到高介绍各种提取方案:找到隐藏的JSON API、驱动Playwright或Selenium、处理无限滚动和「加载更多」按钮、减少代理流量、添加重试机制、并发抓取,以及导出干净的JSON和CSV。

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

要点

  • 动态抓取:JavaScript构建的内容不在原始HTML中,因此你要么调用页面所获取的底层JSON API,要么在无头浏览器中渲染页面。
  • 最佳代理类型:轮换住宅代理,提供可通过检测的真实消费者IP。
  • 价格:每GB 1美元起,按量付费,流量永不过期,无需订阅。
  • 覆盖范围:遍布195个国家、超过9,000万个符合道德规范来源的IP。
  • 可靠性:成功率99.51%,在G2上获得4.8/5的评分。
  • 协议与定向:HTTP、HTTPS和SOCKS5,并包含国家级定向。
抓取动态网页的方法优先级

网页为何会成为动态页面?

当有意义的内容是由浏览器中的JavaScript生成、而不是直接包含在初始HTML响应中时,页面就是动态的。服务器发送一个轻量的骨架,随后客户端代码会在页面打开后获取数据并注入DOM。

最快的手动检查方法是对比同一页面的两种视图。右键选择「查看源代码」查看服务器返回的原始HTML,再打开开发者工具的 Elements 面板,查看脚本运行后的实时DOM。如果所需的价格或列表出现在 Elements 面板中,却在源代码里找不到,那么该内容就是动态的,单次请求无法捕获它。

  • 静态内容存在于源代码中,可以直接从HTTP响应中解析。
  • 动态内容只出现在渲染后的DOM中,通过后台请求载入。

你可以用代码自动完成这项诊断。首先,统计普通HTTP客户端下载的原始HTML中目标元素的数量:

import requests
from bs4 import BeautifulSoup

url = "https://example.com/products"
headers = {"User-Agent": "Mozilla/5.0"}

resp = requests.get(url, headers=headers, timeout=30)
soup = BeautifulSoup(resp.text, "html.parser")

raw_cards = soup.select("div.product-card")
print("Cards in raw HTML:", len(raw_cards))

# If this prints 0 but the page clearly shows products in a browser,
# the content is injected by JavaScript and the page is dynamic.

然后在无头浏览器中渲染同一个URL并再次统计。两个数字之间的差异即可证明页面包含动态内容:

from playwright.sync_api import sync_playwright

with sync_playwright() as p:
    browser = p.chromium.launch(headless=True)
    page = browser.new_page()
    page.goto("https://example.com/products", wait_until="networkidle")
    rendered_cards = page.query_selector_all("div.product-card")
    print("Cards after JavaScript runs:", len(rendered_cards))
    browser.close()

# Raw HTML 0, rendered DOM 48 means the page builds its content dynamically.

在编写完整的抓取程序前,也应确认目标网站是否允许自动化访问。我们关于如何检查网站是否允许抓取的指南介绍了如何阅读robots规则和服务条款。

抓取动态页面应该使用哪种方法?

应选择能可靠返回数据的最轻量方案:有隐藏的JSON API时直接调用,只有在没有可用API时才改用无头浏览器。每种方案都需要在速度、成本和页面还原能力之间权衡。

下表比较了四种常见方法,方便你在写代码前做选择。实际上,大多数项目会组合使用其中两种:用隐藏的API做批量加载,用浏览器处理难以直接抓取的页面。

方法 速度 成本 最适合 主要限制
隐藏的JSON API 最快 最低 位于干净XHR调用背后的数据 端点可能变更或需要令牌
Playwright 中等 较高 完整渲染、现代异步控制 占用CPU、内存和带宽
Selenium 中等 较高 遗留技术栈、广泛的语言支持 配置较慢、等待代码更冗长
requests-html 简单页面上的轻量JavaScript 基本无人维护、渲染不稳定

成本很重要,因为经按量计费代理访问的无头浏览器,下载的数据量远大于单次API调用。大规模抓取时,所选方法对账单的影响不亚于代码本身。关于节奏控制和请求头这类更广泛的规范,请参阅我们的网页抓取最佳实践

如何找到动态页面背后隐藏的API?

寻找页面为获取数据而发出的后台请求,因为该端点通常返回你可以直接调用的干净JSON。这是最快、最可靠的方法,并且完全避免运行浏览器。

打开开发者工具,切换到 Network 标签,按 Fetch/XHR 筛选,然后重新加载页面。随着内容出现,留意返回所需数据的JSON请求。检查请求的URL、方法、查询参数以及它发送的任何请求头或令牌,然后右键将它复制为 cURL,以查看完整结构。一旦你能用HTTP客户端重现该调用,就完全跳过了渲染,以极低的成本获取结构化数据。

下面是一个最小示例,它通过代理调用一个JSON端点并直接读取字段:

import requests

proxy = "http://USERNAME:[email protected]:823"
proxies = {"http": proxy, "https": proxy}

headers = {
    "User-Agent": "Mozilla/5.0",
    "Accept": "application/json",
    "Referer": "https://example.com/products",
}
params = {"category": "laptops", "page": 1}

r = requests.get(
    "https://example.com/api/v2/products",
    headers=headers,
    params=params,
    proxies=proxies,
    timeout=30,
)
r.raise_for_status()
data = r.json()

for item in data["items"]:
    print(item["name"], item["price"])

如果端点是分页的,就递增page参数,或跟随响应返回的下一个游标。当某个请求需要令牌时,捕获页面首次签发它的位置,并在同一会话内重放。通过DataImpulse转发请求后,端点看到的是住宅IP,而不是你的服务器地址,这在API按来源限速时很重要。如果代理本身的认证出现问题,我们关于代理认证的说明介绍了凭据格式。

如何用Playwright抓取动态页面?

没有可直接调用的干净API时,可以使用 Playwright 等无头浏览器,让JavaScript像面对真实用户一样完整执行,然后读取完成后的DOM。Playwright是现代的默认之选,因为它自带可靠的自动等待、出色的异步支持和简洁的代理配置。

先安装它并下载一个浏览器二进制文件:

pip install playwright
playwright install chromium

核心原则是等待你需要的特定元素,而不是用固定的休眠去猜测。下面的示例启动Chromium,并通过字典配置代理,等待商品网格,然后提取渲染后的值:

from playwright.sync_api import sync_playwright

PROXY = {
    "server": "http://gw.dataimpulse.com:823",
    "username": "USERNAME",
    "password": "PASSWORD",
}

with sync_playwright() as p:
    browser = p.chromium.launch(headless=True, proxy=PROXY)
    context = browser.new_context(user_agent="Mozilla/5.0")
    page = context.new_page()
    page.goto("https://example.com/products", wait_until="domcontentloaded")

    page.wait_for_selector("div.product-card", timeout=15000)

    cards = page.query_selector_all("div.product-card")
    for card in cards:
        name = card.query_selector(".title").inner_text()
        price = card.query_selector(".price").inner_text()
        print(name, price)

    browser.close()

等待策略往往决定动态抓取程序的成败,因此了解这四种策略以及各自的适用场景很有帮助。相比 networkidle,应优先等待具体选择器或条件,因为在保持长连接的页面上,networkidle 可能会一直等待:

# Wait for a specific element (most reliable)
page.wait_for_selector("div.product-card", timeout=15000)

# Wait for the initial DOM to be built, before every script finishes
page.goto(url, wait_until="domcontentloaded")

# Wait until there are no network connections for 500 ms
page.goto(url, wait_until="networkidle")

# Wait for a custom condition, such as a minimum item count
page.wait_for_function(
    "document.querySelectorAll('div.product-card').length > 20"
)

如果你的目标严重依赖在浏览器中运行的脚本,我们的配套文章用JavaScript进行网页抓取更深入地讲解了让这些页面变难的客户端渲染。

如何处理无限滚动和「加载更多」按钮?

触发页面加载更多内容时使用的事件,然后在读取之前等待每一批新内容渲染完成。无限滚动和「加载更多」按钮只会在用户操作后获取数据,因此单次页面加载只能捕获第一屏。

在动用浏览器之前,再次查看 Network 标签,因为无限滚动几乎总是由分页的XHR调用驱动,使用上述 requests 方法遍历该端点,比滚动页面的成本低得多。如果你必须滚动,就在一个受控循环中进行,一旦没有新项目出现就停止:

def scroll_until_stable(page, item_selector, max_rounds=30):
    seen = 0
    for _ in range(max_rounds):
        page.evaluate("window.scrollTo(0, document.body.scrollHeight)")
        page.wait_for_timeout(1500)
        count = len(page.query_selector_all(item_selector))
        if count == seen:
            break            # no new items loaded, stop
        seen = count
    return seen

使用显式按钮而非滚动的页面需要一个点击循环。在每次点击前读取项目数量,点击按钮,然后等待数量增长,这样就不会在渲染完成前继续操作:

def click_load_more(page, button_selector, item_selector, max_clicks=50):
    for _ in range(max_clicks):
        button = page.query_selector(button_selector)
        if button is None or not button.is_visible():
            break
        before = len(page.query_selector_all(item_selector))
        button.click()
        page.wait_for_function(
            "([sel, n]) => document.querySelectorAll(sel).length > n",
            arg=[item_selector, before],
        )
    return len(page.query_selector_all(item_selector))

采集时要同步去重,因为每轮重复读取相同的节点很常见。维护一个唯一ID或URL集合,只添加尚未见过的记录。

可以用Selenium代替Playwright吗?

可以。Selenium同样能驱动真实浏览器,当你的团队已经在用它、或需要它广泛的语言支持时,它是一个可靠的选择。其模式与Playwright基本相同:加载页面,用WebDriverWait等待某个条件,然后从渲染后的DOM中读取元素。

from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC

options = Options()
options.add_argument("--headless=new")
options.add_argument("--proxy-server=http://gw.dataimpulse.com:823")

driver = webdriver.Chrome(options=options)
driver.get("https://example.com/products")

WebDriverWait(driver, 15).until(
    EC.presence_of_element_located((By.CSS_SELECTOR, "div.product-card"))
)

for card in driver.find_elements(By.CSS_SELECTOR, "div.product-card"):
    name = card.find_element(By.CSS_SELECTOR, ".title").text
    price = card.find_element(By.CSS_SELECTOR, ".price").text
    print(name, price)

driver.quit()

有一点需要注意:单纯的–代理-server标志不接受用户名和密码,因此需要认证的代理则需要借助辅助工具。Selenium Wire库能够干净地注入凭据,我们关于Selenium Wire 代理配置的教程展示了确切的设置。Selenium比Playwright更难配置,其等待代码也更冗长,因此除非有理由继续使用Selenium,新项目往往从Playwright起步。

渲染页面时如何削减代理带宽?

屏蔽不需要的资源,避免浏览器下载不会解析的图片、字体和媒体。无头浏览器默认会请求所有资源,而经由按GB计费的代理访问时,每个字节都会产生费用。

Playwright允许你拦截请求并中止与提取无关的请求。屏蔽图片、媒体和字体可以大幅减少代理流量,同时保留你真正读取的JSON和HTML不受影响:

BLOCK = {"image", "media", "font"}

def block_heavy(route):
    if route.request.resource_type in BLOCK:
        route.abort()
    else:
        route.continue_()

context = browser.new_context()
context.route("**/*", block_heavy)
page = context.new_page()
page.goto("https://example.com/products", wait_until="domcontentloaded")

你可以扩展同一个处理器,跳过那些只增加负载却不增加数据的分析和广告域名。DataImpulse采用按量付费模式,流量每GB 1美元起且永不过期,因此削减浪费的字节会直接降低每次抓取的成本。在相关页面之间复用同一个浏览器上下文还能避免重复的启动开销,最省钱的请求,仍是通过改用隐藏API而省去的那一次。

如何加入错误处理和重试?

为每次导航加入带指数退避的重试循环,这样一次加载缓慢或短暂封锁就不会毁掉整个运行。动态页面会间歇性失败,在大规模下,一个在首次超时就放弃的抓取程序会丢失大量记录。

捕获Playwright抛出的超时,等待一个逐渐增长的间隔,并在有限次数内重试,然后记录失败并继续:

import time
from playwright.sync_api import TimeoutError as PlaywrightTimeout

def fetch_with_retries(page, url, selector, retries=3, backoff=2):
    for attempt in range(1, retries + 1):
        try:
            page.goto(url, wait_until="domcontentloaded", timeout=30000)
            page.wait_for_selector(selector, timeout=15000)
            return page.query_selector_all(selector)
        except PlaywrightTimeout:
            wait = backoff ** attempt
            print(f"Attempt {attempt} timed out, retrying in {wait}s")
            time.sleep(wait)
    raise RuntimeError(f"Failed to load {url} after {retries} attempts")

将重试与代理轮换结合使用,让每次尝试都从一个新的IP发出。如果某个出口被限速,下一次请求会落在另一个地址上,往往就能成功。住宅代理移动代理的轮换池会为每个请求自动分配一个新IP,从而把许多硬性封锁变成一次简单的重试。针对代理层面的失败,我们关于HTTP错误407的说明介绍了最常见的认证响应。

如何并发抓取大量动态页面?

结合使用Playwright异步API与asyncio信号量,让多个页面同时渲染,而不会压垮你的机器或目标站点。并发能让动态抓取重新获得吞吐量,因为每个浏览器页面大部分时间都在等待网络。

信号量限制了并行运行的页面数量。在各任务之间共享一个浏览器和上下文,为每个URL打开一个页面,然后汇总结果:

import asyncio
from playwright.async_api import async_playwright

PROXY = {
    "server": "http://gw.dataimpulse.com:823",
    "username": "USERNAME",
    "password": "PASSWORD",
}

async def scrape_one(context, url, sem):
    async with sem:
        page = await context.new_page()
        try:
            await page.goto(url, wait_until="domcontentloaded")
            await page.wait_for_selector("div.product-card", timeout=15000)
            cards = await page.query_selector_all("div.product-card")
            return [await c.inner_text() for c in cards]
        finally:
            await page.close()

async def main(urls):
    sem = asyncio.Semaphore(5)   # at most 5 pages at once
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=True, proxy=PROXY)
        context = await browser.new_context()
        tasks = [scrape_one(context, u, sem) for u in urls]
        results = await asyncio.gather(*tasks)
        await browser.close()
    return results

urls = ["https://example.com/products?page=%d" % i for i in range(1, 21)]
data = asyncio.run(main(urls))

把并发上限保持适度。过多的并行页面会耗尽内存并触发限速,由此增加的重试成本会超过并行带来的收益。可从5个左右开始,只有在成功率保持较高时才提高并发数。

如何将抓取的数据导出为JSON和CSV?

将每个页面的结果收集为字典列表,然后将该列表写入JSON以保存嵌套数据,或写入CSV以生成扁平、适合电子表格的表格。将提取与导出分开,能让抓取程序更易于测试和重跑。

import json
import csv

records = [
    {"name": "Laptop A", "price": "999"},
    {"name": "Laptop B", "price": "1299"},
]

# Export to JSON
with open("products.json", "w", encoding="utf-8") as f:
    json.dump(records, f, ensure_ascii=False, indent=2)

# Export to CSV
with open("products.csv", "w", newline="", encoding="utf-8") as f:
    writer = csv.DictWriter(f, fieldnames=["name", "price"])
    writer.writeheader()
    writer.writerows(records)

记录包含嵌套列表或对象时,应选择JSON,当每条记录都共享相同的扁平字段、且分析师会用电子表格打开它时选择CSV。在写入前对字段进行规范化,使每条记录都有相同的键,这样无论动态页面如何渲染,导出步骤都能保持简单。从这里开始,同一份JSON无需进一步解析即可用于数据库加载或分析任务。

针对延迟加载内容的等待策略

常见问题

在编写抓取程序之前,我如何判断一个页面是否是动态的?

在开发者工具中对比「查看源代码」与 Elements 面板,或统计原始HTML与渲染后DOM中的某个目标元素数量。如果数据只有在JavaScript运行后才出现,那么该页面就是动态的,需要调用API或使用无头浏览器。

调用隐藏API和使用无头浏览器,哪个更好?

能找到隐藏的API时就调用它,因为它能以远低于渲染的成本和更快的速度返回干净的JSON。只有在不存在可重现的API、或数据依赖复杂的客户端脚本时,才使用无头浏览器。

动态抓取应该用Playwright还是Selenium?

Playwright凭借可靠的自动等待和内置的异步支持成为现代的默认选择,因此新项目通常从它起步。当你的技术栈已经在用Selenium、或你需要它更广泛的语言支持时,Selenium是很合适的。

使用无头浏览器时,如何减少代理流量?

拦截请求并中止你不解析的图片、字体、媒体和分析请求。由于代理费用按GB计量,这能显著削减流量,同时也加快每次页面加载。

如何抓取通过无限滚动加载的内容?

先在Network 标签中查找滚动背后的分页XHR调用,并直接遍历该端点。如果你必须滚动,就在一个受控循环中进行,并在读取之前等待每一批新内容渲染完成。

什么情况下DataImpulse并不合适?

如果你需要静态ISP代理、完全托管的抓取API,或访问银行和政府网站,那么DataImpulse并不是合适的工具。它专注于用于采集公开数据和访问内容的轮换住宅、移动和数据中心代理。

使用可靠代理抓取动态页面

如果你的抓取程序需要行为像真实访客的住宅、移动或数据中心IP,你可以从每GB 1美元起的按量付费流量开始。创建一个DataImpulse账户,通过轮换或粘性会话转发动态抓取请求。


Share article: