如何用Pyppeteer打造高并发无头浏览器采集方案

原创

jackcode

发布于 2025-05-12 11:02:08

24210

代码可运行

文章被收录于专栏：爬虫资料爬虫资料

运行总次数：0

代码可运行

以下文章将从行业痛点出发，结合 Pyppeteer 高并发无头浏览器技术，讲解如何在 Python 中打造一个可配置代理的高效采集方案，以采集 Amazon 今日特价商品并分析优惠价格与评分。文章按照“行业问题 → 技术灵感 → 构思实现 → 验证实验 → 潜在价值”五大板块展开，代码示例中集成了爬虫代理，并附有详尽注释，帮助读者快速上手。

一、行业问题

在电商竞争日益激烈的背景下，商家与分析师需要实时监控平台特价活动，以便快速调整定价策略和营销方案。然而，Amazon 等大型电商网站通常采用动态加载、反爬机制及 IP 限制策略，传统静态爬虫难以稳定获取高质量数据。

此外，单线程抓取速度受限，无法满足海量商品监测需求，亟需高并发、分布式的浏览器级采集方案来绕过限制并提升效率。

二、技术灵感

借鉴 Puppeteer 在 Node.js 中的成熟实践，Pyppeteer 作为其 Python 移植版本，可通过 Chrome DevTools 协议实现浏览器自动化，天然支持 headless 模式，灵活性与可控性优于 Selenium、。

结合爬虫代理提供的智能动态代理服务（域名、端口、用户名、密码），我们可在每个浏览器实例或页面上下文中注入独立代理，突破单 IP 限制，并通过模拟真实用户行为，进一步降低被识别风险、。

三、构思实现

3.1 系统架构

控制层：采用 asyncio 事件循环与信号量限制并发量；
浏览器层：利用 Pyppeteer 启动 Chromium，传入 --proxy-server 代理配置；
页面层：每个页面设置自定义 User-Agent、注入 Cookie，导航至今日特价页，执行 DOM 抓取；
数据层：将采集结果汇总至 Python 列表，再导出为 JSON/CSV 供后续分析。

3.2 关键代码

import asyncio
import json
from pyppeteer import launch

# 代理配置：亿牛云爬虫代理示例 16yun.cn
PROXY_HOST = 'proxy.16yun.cn'
PROXY_PORT = '12345'
PROXY_USER = '16YUN'
PROXY_PASS = '16IP'

# 并发控制信号量
SEMAPHORE = asyncio.Semaphore(5)

async def fetch_deals(browser, semaphore):
    async with semaphore:
        # 启动新页面并设置代理
        page = await browser.newPage()
        await page.setUserAgent(
            'Mozilla/5.0 (Windows NT 10.0; Win64; x64) '
            'AppleWebKit/537.36 (KHTML, like Gecko) '
            'Chrome/90.0.4430.93 Safari/537.36'
        )  # 自定义 User-Agent
        # 设置登录态 Cookie（可选）
        await page.setCookie({
            'name': 'session-id',
            'value': 'YOUR_SESSION_VALUE',
            'domain': '.amazon.com'
        })
        # 页面代理认证
        await page.authenticate({'username': PROXY_USER, 'password': PROXY_PASS})
        # 跳转至今日特价页面
        await page.goto('https://www.amazon.com/gp/goldbox', {'waitUntil': 'networkidle2'})
        # 抓取特价商品列表
        items = await page.querySelectorAll('.DealGridItem-module__dealItem')
        results = []
        for item in items:
            title = await page.evaluate('(el) => el.querySelector("h2").innerText', item)
            price = await page.evaluate(
                '(el) => el.querySelector(".a-price-whole")?.innerText', item
            )
            original = await page.evaluate(
                '(el) => el.querySelector(".a-price.a-text-price")?.innerText', item
            )
            rating = await page.evaluate(
                '(el) => el.querySelector(".a-icon-alt")?.innerText', item
            )
            results.append({
                'title': title.strip(),
                'deal_price': price,
                'original_price': original,
                'rating': rating
            })
        await page.close()
        return results

async def main():
    # 启动带代理的 headless 浏览器实例
    browser = await launch({
        'headless': True,
        'args': [
            f'--proxy-server=http://{PROXY_HOST}:{PROXY_PORT}',
            '--no-sandbox', '--disable-setuid-sandbox'
        ]
    })
    # 并发执行抓取任务
    tasks = [fetch_deals(browser, SEMAPHORE) for _ in range(5)]
    all_results = await asyncio.gather(*tasks)
    # 合并结果并存储
    merged = [item for sub in all_results for item in sub]
    with open('amazon_deals.json', 'w', encoding='utf-8') as f:
        json.dump(merged, f, ensure_ascii=False, indent=2)
    await browser.close()

if __name__ == '__main__':
    asyncio.get_event_loop().run_until_complete(main())

以上代码示例中：