pyppeteer

提起 selenium 想必大家都不陌生，作为一款知名的 Web 自动化测试框架，selenium 支持多款主流浏览器，提供了功能丰富的API 接口，经常被我们用作爬虫工具来使用。但是 selenium 的缺点也很明显，比如速度太慢、对版本配置要求严苛，最麻烦是经常要更新对应的驱动。还有些网页是可以检测到是否是使用了selenium 。并且selenium 所谓的保护机制不允许跨域 cookies 保存以及登录的时候必须先打开网页然后后加载 cookies 再刷新的方式很不友好。那么pyppeteer便成为你的不二之选。

安装

pip install pypeteer 
pip install asyncio

pypeteer 常见操作

• 启动浏览器实例

browser = await launch()

• 打开一个空白页

page = await browser.newPage()

• 在地址栏输入网址并等待加载

await page.goto('https://example.com')

• 网页截图

await page.screenshot({path: 'example.png'})

• 添加useragent

await page.setUserAgent('Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.36 Edge/16.16299')

• 打开百度

await page.goto('https://www.baidu.com')

• 获取输入框焦点并输入文字

await page.type('#kw', keyword, {'delay': 100}) 获取输入框焦点并输入文字{'delay': input_time_random() - 50}

• 鼠标点击

# await page.click() 点击一个元素
await page.click('#su')

• 等待页面加载出来

await page.waitForNavigation({'waitUntil': 'load'})
# 等待页面加载出来，等同于window.onload

• 关掉浏览器

await browser.close()

• 获取网页内容

content = await page.content()

• 获取cookies

await get_cookie(page)
cookies = await page.cookies()

• 在网页中执行js代码

await page.evaluate(js1)

• 模拟键盘按下某个按键

await page.keyboard.press 

• 页面等待，可以是时间、某个元素、某个函数

await page.waitFor(10000)

• 获得当前访问的url

await page.url 

• 获取当前页面所有的 iframe，然后根据 iframe 的名字精确获取某个想要的 iframe

await page.frames() 

• 获取 iframe 中的某个元素

await iframe.$('.srchsongst') 

• 在浏览器中执行函数，相当于在控制台中执行函数，返回一个 Promise

await iframe.evaluate() 

• 将类数组对象转化为对象

Array.from 

• 相当于在 iframe 中运行 document.queryselector 获取指定元素，并将其作为第一个参数传递

await iframe.$eval() 

• 相当于在 iframe 中运行 document.querySelectorAll 获取指定元素数组，并将其作为第一个参数传递

await iframe.$$eval 

• 设置页面大小

await self.page.setViewport(viewport={'width': width, 'height': height})

代码演示

from pyppeteer import launch
import asyncio
import copy


class TestSpider(object):
    def __init__(self):
        self.chrome_extension = "C:\\Program Files\\Google\\Chrome\\Application\\chrome.exe"
        self.launch_kwargs = {
          # 控制是否为无头模式
        'executablePath': chrome_extension,  # chrome 文件目录位置
        "headless": True,  # 控制是否为无头模式
        "dumpio": True,  # 当界面开多了时会卡住，设置这个参数就不会了
        "userDataDir": r"./ceshi",  # 用户数据保存目录 这个最好也自己指定一个目录
        # 如果不指定的话，chrome会自动新建一个临时目录使用，在浏览器退出的时候会自动删除临时目录
        # 在删除的时候可能会删除失败（不知道为什么会出现权限问题，我用的windows） 导致浏览器退出失败
        # 然后chrome进程就会一直没有退出 CPU就会狂飙到99%
        'autoClose': True,
        # chrome启动命令行参数
        "args": [
            # 浏览器代理 配合某些中间人代理使用
            # "--proxy-server=http://127.0.0.1:8008",
            # 最大化窗口
            "--start-maximized",
            # 窗口大小
            '--window-size=1366,768'，
            # 取消沙盒模式 沙盒模式下权限太小
            "--no-sandbox",
            # 不显示信息栏  比如 chrome正在受到自动测试软件的控制 ...
            "--disable-infobars",
            # log等级设置 在某些不是那么完整的系统里 如果使用默认的日志等级 可能会出现一大堆的warning信息
            "--log-level=3",
            # 设置UA
            "--user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/71.0.3578.98 Safari/537.36",
        ],

    }

    def __del__(self):
        asyncio.gather(self.browser.close())  # 爬虫程序结束，关闭 browser对象
        print("-------程序结束")

    async def run(self): 
        item = dict()
        url_list = ["https://www.jianshu.com/u/93d4bac0fa23"]  # 此处链接脱敏处理
        temp_url = "https://www.jianshu.com/u/93d4bac0fa23/{}.html" # 此处链接脱敏处理
        for i in range(2, 10):
            url_list.append(temp_url.format(i))
        self.browser = await launch(self.launch_kwargs)
        page = await self.browser.newPage()
        await page.evaluateOnNewDocument(
            '() =>{ Object.defineProperties(navigator,''{ webdriver:{ get: () => false } }) }')
        await page.setUserAgent(
            'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/92.0.4515.159 Safari/537.36')
        for url in url_list:
            await page.goto(url, waitUntil='networkidle0')  # 等待网页加载完成
            item_list = []
            item["detail_url"] = url
            a_list = await page.xpath("//ul[@class='list']//li/a")  # 利用xpath语法提取数据
            for i in a_list:
                title = await (await i.getProperty("textContent")).jsonValue()  # 提取文本内容
                get_url = await (await i.getProperty("href")).jsonValue() 
                item["title"], item["get_url"] = str(title).strip(), get_url
                item_list.append(copy.deepcopy(item))
            if item_list:
                await self.get_detail_parse(page, copy.deepcopy(item_list))

    async def get_detail_parse(self, page, item_list):
        for item in item_list:
            await page.goto(item["get_url"], waitUntil='networkidle0')  # 等待网页加载完成
            a_list = await page.xpath("//div[@class='content']//a")
            content_data = await page.content()
            print(content_data)
            for i in a_list:
                item["name"] = await (await i.getProperty("textContent")).jsonValue()
                item["url"] = await (await i.getProperty("href")).jsonValue()
                print(item)


if __name__ == '__main__':
    func = TestSpider()
    asyncio.run(func.run())  # 运行协程程序