使用puppeteer库编写的爬虫程序

原创

华科云商小徐

发布于 2025-04-02 03:34:26

13000

代码可运行

文章被收录于专栏：小徐学爬虫小徐学爬虫

运行总次数：0

代码可运行

Embassy是基于aiohttp的异步库，适合处理普通的HTTP请求，而Puppeteer是控制Headless Chrome的，适合需要处理JavaScript渲染的页面。用户可能遇到了需要抓取动态生成内容的情况，或者之前的Embassy示例无法获取到完整的数据，所以转向Puppeteer。

以下是一个使用 Pyppeteer (Python 版本的 Puppeteer) 编写的网络爬虫示例，用于处理需要 JavaScript 渲染的页面：

先安装依赖

pip install pyppeteer

示例代码：动态页面爬虫

import asyncio
from urllib.parse import urljoin
from pyppeteer import launch
from pyppeteer.errors import TimeoutError

class PuppeteerCrawler:
    def __init__(self, headless=True, timeout=30):
        self.headless = headless  # 是否无头模式
        self.timeout = timeout    # 页面加载超时时间
        self.browser = None       # 浏览器实例
        self.visited = set()      # 已访问的URL

    async def init_browser(self):
        """启动浏览器实例"""
        self.browser = await launch(
            headless=self.headless,
            args=['--no-sandbox', '--disable-setuid-sandbox']
        )

    async def crawl(self, start_url):
        """启动爬虫"""
        if not self.browser:
            await self.init_browser()
        
        page = await self.browser.newPage()
        await page.setUserAgent('Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/90.0.4430.212 Safari/537.36')
        
        try:
            await self.process_page(page, start_url)
        finally:
            await page.close()
            await self.browser.close()

    async def process_page(self, page, url):
        """处理单个页面"""
        if url in self.visited:
            return
        self.visited.add(url)

        try:
            # 导航到页面
            await page.goto(url, {'timeout': self.timeout * 1000})
            await page.waitForSelector('body', timeout=self.timeout*1000)  # 等待页面加载

            # 获取页面内容
            content = await page.content()
            
            # 示例：提取标题
            title = await page.title()
            print(f'Scraped: {url}')
            print(f'Title: {title}')

            # 提取链接
            links = await self.extract_links(page, url)
            for link in links:
                if link not in self.visited:
                    new_page = await self.browser.newPage()
                    await self.process_page(new_page, link)
                    await new_page.close()

        except TimeoutError:
            print(f'Timeout loading {url}')
        except Exception as e:
            print(f'Error processing {url}: {str(e)}')

    async def extract_links(self, page, base_url):
        """提取页面中的链接"""
        return await page.evaluate('''(base_url) => {
            return Array.from(document.querySelectorAll('a'))
                .map(a => new URL(a.href, base_url).href)
                .filter(href => href.startsWith('http'));
        }''', base_url)

async def main():
    crawler = PuppeteerCrawler(headless=True)  # headless=False 可以显示浏览器
    await crawler.crawl('https://example.com')

if __name__ == '__main__':
    asyncio.run(main())

代码说明

核心功能：
- 使用 pyppeteer 控制 Chromium 浏览器
- 支持 JavaScript 渲染的动态页面
- 自动等待页面元素加载 (waitForSelector)
- 支持递归爬取页面链接
关键配置： launch( headless=True, # 无头模式（不显示浏览器界面） args=[ '--no-sandbox', '--disable-setuid-sandbox' # Linux 系统需要 ] )
优势场景：
- 需要执行 JavaScript 的 SPA (单页应用)
- 需要处理懒加载内容
- 需要模拟用户交互（点击、滚动等）
- 需要生成页面截图/PDF

高级功能扩展

1. 页面截图

await page.screenshot({'path': 'screenshot.png', 'fullPage': True})

2. 执行自定义 JavaScript

dimensions = await page.evaluate('''() => {
    return {
        width: document.documentElement.clientWidth,
        height: document.documentElement.clientHeight
    }
}''')

3. 处理弹窗

page.on('dialog', lambda dialog: dialog.dismiss())

4. 设置代理

await page.authenticate({
    'username': 'proxy_user',
    'password': 'proxy_pass'
})

注意事项

性能优化：
- 复用浏览器实例 (browser.newPage() 而不是频繁启动)
- 禁用图片加载： await page.setRequestInterception(True) page.on('request', lambda req: req.abort() if req.resourceType() == 'image' else req.continue_())
反反爬策略：
- 使用随机 User-Agent
- 添加随机延迟 (await asyncio.sleep(random.uniform(1, 3)))
- 使用代理池
资源管理：
- 及时关闭页面 (await page.close())
- 处理内存泄漏
- 监控浏览器进程