Pyppeteer多页爬取时page.click与waitForNavigation适配Python3.11方案

问题根源 #

Python3.8+ 开始弃用 asyncio.wait 直接接收协程对象的用法，要求必须将协程通过 asyncio.create_task() 包装为 Task 对象后再传入，否则会触发警告。结合Pyppeteer中page.click与page.waitForNavigation的协同需求，以下是适配Python3.11的完整优化实现：

优化后代码示例 #

import asyncio
from pyppeteer import launch

async def crawl_ecommerce_pages():
    # 初始化浏览器，配置无头模式及性能参数
    browser = await launch(
        headless=True,
        args=['--no-sandbox', '--disable-dev-shm-usage', '--disable-gpu']
    )
    page = await browser.newPage()
    await page.setViewport({'width': 1200, 'height': 800})
    # 设置模拟UA，规避基础反爬
    await page.setUserAgent('Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/118.0.0.0 Safari/537.36')

    try:
        # 进入目标列表页，设置超时避免卡死
        await page.goto('https://example-ecommerce-site.com/product-list', timeout=60000)

        while True:
            # 提取当前页商品数据（需根据目标网站DOM结构修改选择器）
            products = await page.evaluate('''
                () => Array.from(document.querySelectorAll('.product-card')).map(card => ({
                    title: card.querySelector('.product-title').textContent.trim(),
                    price: card.querySelector('.product-price').innerText.trim(),
                    link: card.querySelector('.product-link').href
                }))
            ''')
            print(f"当前页爬取到 {len(products)} 条商品数据")
            # 此处可添加数据持久化逻辑（写入文件/数据库）

            # 查找下一页按钮
            next_btn = await page.querySelector('.pagination-next')
            if not next_btn:
                print("已爬取全部页面")
                break

            # 关键优化：用create_task包装协程，消除Python3.8+警告
            click_task = asyncio.create_task(next_btn.click())
            nav_task = asyncio.create_task(
                page.waitForNavigation(
                    waitUntil='networkidle2',  # 等待网络空闲后再继续
                    timeout=30000
                )
            )

            # 等待两个任务完成，确保点击和导航协同执行
            done, pending = await asyncio.wait(
                [click_task, nav_task],
                return_when=asyncio.ALL_COMPLETED
            )

            # 检查任务异常，主动抛出避免静默失败
            for task in done:
                if task.exception():
                    raise task.exception()

    except Exception as e:
        print(f"爬取出错: {str(e)}")
    finally:
        # 确保浏览器资源被释放
        await browser.close()

if __name__ == '__main__':
    # Python3.7+推荐入口，Python3.11对其有性能优化
    asyncio.run(crawl_ecommerce_pages())

核心优化点 #

1. 消除asyncio.wait警告：用asyncio.create_task()分别包装page.click()和page.waitForNavigation()，将协程转为Task对象后传入asyncio.wait，完全符合Python3.8+的规范。
2. 协同逻辑稳定性：通过asyncio.wait等待两个任务全部完成，避免点击与导航的竞态问题。
3. 异常与超时处理：为页面跳转和导航设置超时时间，主动捕获并抛出任务异常，避免程序静默挂死。
4. Python3.11适配：使用asyncio.run()作为异步程序入口，该方法在Python3.11中得到性能优化，同时兼容所有3.7+版本。

注意事项 #

• 根据目标电商网站的DOM结构，修改商品数据提取、下一页按钮的选择器。
• 若遇到反爬，可添加随机延迟（await asyncio.sleep(random.uniform(1,3))）、IP代理等策略。
• 确保Pyppeteer为最新版本：执行pip install --upgrade pyppeteer更新。

内容的提问来源于stack exchange，提问作者arzak