Pypputeer自动化

Pyppeteer简介

pyppeteer 是 Python 语言的一个库，它是对 Puppeteer 的一个非官方端口，Puppeteer 是一个 Node 库，Puppeteer是Google基于Node.js开发的一个工具，它提供了一种高层次的 API 来通过 DevTools 协议控制 Chrome 或 Chromium。pyppeteer 可以用来进行网页自动化处理，支持页面抓取、表单提交、UI测试、JavaScript执行等功能，非常适合用于网页爬虫或自动化测试。

在pyppeter中，实际上它背后有一个类似Chrome浏览器的Chromium浏览器在执行一些动作进行网页渲染。

Chrome与Chromium渊源。两款浏览器内核是一样的，实现方式也是一样，可以认为是开发版和正式版的区别，功能基本没有太大的区别。

环境安装

pip install pyppeteer

注意：支持异步需要3.5以上的解释器

import pyppeteer
print(pyppeteer.executablePath()) #查看chromium存放路径
print(pyppeteer.__chromium_revision__) #查看版本号

官方文档：

API Reference — Pyppeteer 0.0.25 documentationhttps://miyakogi.github.io/pyppeteer/reference.html

测试样例

from pyppeteer import launch
import asyncio
import time
async def main():# 启动一个浏览器(headless默认是无头即无界面浏览器,改为false有界面)browser = await launch(headless=False,args=['--disable-infobars','--window-size=1920,1080'])# 创建一个页面page = await browser.newPage()# 设置页面视图大小await page.setViewport({'width':1900,'height':1080})# 跳转到百度await page.goto('https://www.baidu.com')# 输入要查询的关键字,type第一个参数是元素的selector(css),第二个是要输入的关键字await page.type('#kw','pyppeteer')# 点击提交按钮await page.click('#su')time.sleep(30)await browser.close()
# 启动异步任务
asyncio.get_event_loop().run_until_complete(main())

基本配置

基本参数

params = {# 关闭无头浏览器"headless":False,"dumpio":True,#防止浏览器卡住r"userDataDir":"./cache-data",  #用户文件地址"args":['--disable-infobars',       #关闭自动化提示框'--window-size=1920,1080',  #设置窗口大小'--log-level=30',           #日志保存等级,建议设置越小越好,要不然生成的日志占用的空间会很大30为waring级别'--user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36','--no-sandbox',             #关闭沙盒模式'--start-maximized',        #窗口最大化模式'--proxy-server=http://localhost:1080' #代理]
}

设置窗口

#UI模式 闭频警告
browser = await launch(headless = False,args=['--disable-infobars'])
page = await browser.newPage()
await page.setViewport({'width':1200,'height':800})

添加头部

网页截图

page.screenshot(path='example.png')

伪装浏览器绕过检测

object.defineProperty()方法会直接在一个对象上定义一个新属性，或者修改一个对象的现有属性，并返回此对象。

await page.evaluateOnNewDocument('()=>{Object.defineProperty(navigator, "webdriver", { get: () => false }); }');

案例演示触发JS

async def main():# 启动一个浏览器browser = await pyppeteer.launch(headless = False,args = ['--disable-infobars','--window-size=1920,1080'])# 打开一个新页面page = await browser.newPage()# 添加用户代理await page.setUserAgent('Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36')await page.evaluateOnNewDocument('()=>{Object.defineProperty(navigator, "webdriver", { get: () => false }); }')await page.goto('https://www.zhipin.com/web/geek/job?query=%E6%95%B0%E6%8D%AE%E5%88%86%E6%9E%90&city=100010000&page=')dimensions = await page.evaluate('() => ({ cookie: document.cookie })')headers = {'User-agent':'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36','Cookie':dimensions['cookie']}url = 'https://www.zhipin.com/web/geek/job?query=%E6%95%B0%E6%8D%AE%E5%88%86%E6%9E%90&city=100010000&page='resp = requests.get(url=url,headers=headers)print(resp.text)
# 启动异步任务
asyncio.get_event_loop().run_until_complete(main())

滚动到页面底部

await page.evaluate(window.scrollBy(0,document.body.scrollHeight))

进阶使用

数据提取

获取属性

登录案例

import asyncio
from pyppeteer import launchasync def main():# 启动浏览器，headless=False 表示非无头模式，也就是浏览器界面是可见的browser = await launch(headless=False, args=['--disable-infobars', '--window-size=1920,1080'])# 开启一个新的浏览器标签页page = await browser.newPage()# 访问指定的URLawait page.goto('https://www.captainbi.com/amz_login.html')# 设置视窗大小await page.setViewport(viewport={'width': 1356, 'height': 768})# 输入用户名await page.type('#username', '123456')# 输入密码，假定密码输入框的ID为'password'await page.type('#password', '123456')  # 请确保选择器正确对应到密码输入框# 单击登录按钮，假定按钮的ID为'submit'# 如果按钮没有ID，则需要提供正确的CSS选择器await page.click('#submit', options={'timeout': 3000})# 运行 main 协程
asyncio.run(main())

综合案例

'''
抓取唯品会关于女性口红等数据
1搜索入口抓口红数据
2根据品牌做检索
3字段 原价-折扣价-品牌
4翻页
5保存入库
根据观察数据是动态加载。所以要使用自动化技术 把动态变静态 结合requests
'''
import requests
from lxml import etree
import pandas as pd
import asyncio
from pyppeteer import launch
from loguru import loggerclass Wph(object):def __init__(self,url,name):self.url = urlself.name = nameself.headers = {'User-Agent':'aaqabbbccc'}self.session = requests.session()self.hadInone = lambda x:x[0] if x else ''self.browser = Noneasync def main(self,url):# 打开一个浏览器self.browser = await launch()# 创建一个窗口page = await self.browser.newPage()# 访问对应的urlawait page.goto(url)text = await page.content()  # 返回页面htmlreturn textdef spider(self):df = pd.DataFrame(columns=['品牌','标题','原价','现价','折扣'])# 发起请求res = self.session.get(self.url,params={'keyword':self.name},headers=self.headers,verify=False)html = etree.HTML(res.text)url_list = html.xpath('.//div[@class="c-filter-group-content"]/div[contains(@class,"c-filter-group-scroll-brand")]/ul/li/a/@href')# 迭代品牌URL地址for i in url_list:# 驱动浏览器请求page_html = asyncio.get_event_loop().run_until_complete(self.main('http:'+i))# 获取网页源代码page = etree.HTML(page_html)htmls = page.xpath('//section[@id="J_searchCatList"]/div')for h in htmls[1:]:# 品牌pinpai = self.hadInone(h.xpath('//div[contains(@class,"c-goods-item__name--two-line")]/text()'))# 标题title = self.hadInone(h.xpath('//div[contains(@class,"c-goods-item__name--two-line")]/text()'))# 原价y_price = self.hadInone(h.xpath('//div[contains(@class,"J-goods-item__market-price")]/text()'))# 卖价x_price = self.hadInone(h.xpath('//div[contains(@class,"c-goods-item__sale-price")]/text()'))# 折扣zk = self.hadInone(h.xpath('div//div[contains(@class,"c-goods-item__discount")]/text()'))logger.info(f'品牌{pinpai},标题{title},原价{y_price},现价{x_price},折扣{zk}')pro = {'品牌':pinpai,'标题':title,'原价':y_price,'现价':x_price,'折扣':zk,}df = df.append([pro])print(pro)# df.to_excel('唯品会数据.xlsx',index=False)return df# def __del__(self):#     if self.browser:#         asyncio.get_event_loop().run_until_complete(self.browser.close())if __name__=='__main__':url = 'https://category.vip.com/suggest.php'name = '香水'w = Wph(url,name)w.spider()