pyppeteer
提起 selenium 想必大家都不陌生,作為一款知名的 Web 自動化測試框架,selenium 支持多款主流瀏覽器,提供了功能豐富的API 接口,經(jīng)常被我們用作爬蟲工具來使用。但是 selenium 的缺點(diǎn)也很明顯,比如速度太慢、對版本配置要求嚴(yán)苛柔袁,最麻煩是經(jīng)常要更新對應(yīng)的驅(qū)動腥例。還有些網(wǎng)頁是可以檢測到是否是使用了selenium 构回。并且selenium 所謂的保護(hù)機(jī)制不允許跨域 cookies 保存以及登錄的時(shí)候必須先打開網(wǎng)頁然后后加載 cookies 再刷新的方式很不友好政己。那么pyppeteer便成為你的不二之選。
安裝
pip install pypeteer
pip install asyncio
pypeteer 常見操作
- 啟動瀏覽器實(shí)例
browser = await launch()
- 打開一個空白頁
page = await browser.newPage()
- 在地址欄輸入網(wǎng)址并等待加載
await page.goto('https://example.com')
- 網(wǎng)頁截圖
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')
- 獲取輸入框焦點(diǎn)并輸入文字
await page.type('#kw', keyword, {'delay': 100}) 獲取輸入框焦點(diǎn)并輸入文字{'delay': input_time_random() - 50}
- 鼠標(biāo)點(diǎn)擊
# await page.click() 點(diǎn)擊一個元素
await page.click('#su')
- 等待頁面加載出來
await page.waitForNavigation({'waitUntil': 'load'})
# 等待頁面加載出來欠气,等同于window.onload
- 關(guān)掉瀏覽器
await browser.close()
- 獲取網(wǎng)頁內(nèi)容
content = await page.content()
- 獲取cookies
await get_cookie(page)
cookies = await page.cookies()
- 在網(wǎng)頁中執(zhí)行js代碼
await page.evaluate(js1)
- 模擬鍵盤按下某個按鍵
await page.keyboard.press
- 頁面等待宜鸯,可以是時(shí)間淋袖、某個元素焰情、某個函數(shù)
await page.waitFor(10000)
- 獲得當(dāng)前訪問的url
await page.url
- 獲取當(dāng)前頁面所有的 iframe裸准,然后根據(jù) iframe 的名字精確獲取某個想要的 iframe
await page.frames()
- 獲取 iframe 中的某個元素
await iframe.$('.srchsongst')
- 在瀏覽器中執(zhí)行函數(shù)炒俱,相當(dāng)于在控制臺中執(zhí)行函數(shù),返回一個 Promise
await iframe.evaluate()
- 將類數(shù)組對象轉(zhuǎn)化為對象
Array.from
- 相當(dāng)于在 iframe 中運(yùn)行 document.queryselector 獲取指定元素推盛,并將其作為第一個參數(shù)傳遞
await iframe.$eval()
- 相當(dāng)于在 iframe 中運(yùn)行 document.querySelectorAll 獲取指定元素?cái)?shù)組峦阁,并將其作為第一個參數(shù)傳遞
await iframe.$$eval
- 設(shè)置頁面大小
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, # 當(dāng)界面開多了時(shí)會卡住,設(shè)置這個參數(shù)就不會了
"userDataDir": r"./ceshi", # 用戶數(shù)據(jù)保存目錄 這個最好也自己指定一個目錄
# 如果不指定的話耘成,chrome會自動新建一個臨時(shí)目錄使用榔昔,在瀏覽器退出的時(shí)候會自動刪除臨時(shí)目錄
# 在刪除的時(shí)候可能會刪除失敗(不知道為什么會出現(xiàn)權(quán)限問題瘪菌,我用的windows) 導(dǎo)致瀏覽器退出失敗
# 然后chrome進(jìn)程就會一直沒有退出 CPU就會狂飆到99%
'autoClose': True,
# chrome啟動命令行參數(shù)
"args": [
# 瀏覽器代理 配合某些中間人代理使用
# "--proxy-server=http://127.0.0.1:8008",
# 最大化窗口
"--start-maximized",
# 窗口大小
'--window-size=1366,768'撒会,
# 取消沙盒模式 沙盒模式下權(quán)限太小
"--no-sandbox",
# 不顯示信息欄 比如 chrome正在受到自動測試軟件的控制 ...
"--disable-infobars",
# log等級設(shè)置 在某些不是那么完整的系統(tǒng)里 如果使用默認(rèn)的日志等級 可能會出現(xiàn)一大堆的warning信息
"--log-level=3",
# 設(shè)置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()) # 爬蟲程序結(jié)束,關(guān)閉 browser對象
print("-------程序結(jié)束")
async def run(self):
item = dict()
url_list = ["http://www.reibang.com/u/93d4bac0fa23"] # 此處鏈接脫敏處理
temp_url = "http://www.reibang.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') # 等待網(wǎng)頁加載完成
item_list = []
item["detail_url"] = url
a_list = await page.xpath("http://ul[@class='list']//li/a") # 利用xpath語法提取數(shù)據(jù)
for i in a_list:
title = await (await i.getProperty("textContent")).jsonValue() # 提取文本內(nèi)容
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') # 等待網(wǎng)頁加載完成
a_list = await page.xpath("http://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()) # 運(yùn)行協(xié)程程序