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 documentation
https://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 launch
async def main():
# 启动浏览器，headless=False 表示非无头模式，也就是浏览器界面是可见的
browser = await launch(headless=False, args=['--disable-infobars', '--window-size=1920,1080'])
# 开启一个新的浏览器标签页
page = await browser.newPage()
# 访问指定的URL
await 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 logger
class Wph(object):
def __init__(self,url,name):
self.url = url
self.name = name
self.headers = {
'User-Agent':'aaqabbbccc'
}
self.session = requests.session()
self.hadInone = lambda x:x[0] if x else ''
self.browser = None
async def main(self,url):
# 打开一个浏览器
self.browser = await launch()
# 创建一个窗口
page = await self.browser.newPage()
# 访问对应的url
await page.goto(url)
text = await page.content() # 返回页面html
return text
def 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()
