pyppeteer github地址:https://github.com/miyakogi/pyppeteer

pyppeteer 英文文檔地址:?https://miyakogi.github.io/pyppeteer/

pyppeteer 官方文檔 API Reference:

https://miyakogi.github.io/pyppeteer/reference.html

puppeteer(Nodejs版 selenium)快速入門:

https://blog.csdn.net/freeking101/article/details/91542887

爬蟲界又出神器 | 一款比selenium更高效的利器:

https://blog.csdn.net/chen801090/article/details/93216278

python爬蟲利器 pyppeteer(模擬瀏覽器)實戰:

https://blog.csdn.net/xiaoming0018/article/details/89841728

Pyppeteer 簡介

?

提起 selenium 想必大家都不陌生，作為一款知名的 Web 自動化測試框架，selenium 支持多款主流瀏覽器，提供了功能豐富的API 接口，經常被我們用作爬蟲工具來使用。但是 selenium 的缺點也很明顯，比如速度太慢、對版本配置要求嚴苛，最麻煩是經常要更新對應的驅動。還有些網頁是可以檢測到是否是使用了selenium。并且selenium 所謂的保護機制不允許跨域 cookies 保存以及登錄的時候必須先打開網頁然后后加載 cookies 再刷新的方式很不友好。

今天就給大家介紹另一款 web 自動化測試工具?Pyppeteer，雖然支持的瀏覽器比較單一，但在安裝配置的便利性和運行效率方面都要遠勝 selenium。

介紹 Pyppeteer 之前先說一下 Puppeteer，Puppeteer 是 Google 基于 Node.js 開發的一個工具，主要是用來操縱 Chrome? 瀏覽器的 API，通過 Javascript 代碼來操縱 Chrome 瀏覽器的一些操作，用作網絡爬蟲完成數據爬取、Web 程序自動測試等任務。其 API 極其完善，功能非常強大。 而 Pyppeteer 又是什么呢？它實際上是 Puppeteer 的 Python 版本的實現，但他不是 Google 開發的，是一位來自于日本的工程師依據 Puppeteer 的一些功能開發出來的非官方版本。

Pyppeteer 其實是 Puppeteer 的 Python 版本。pyppeteer 模塊看不懂就去看puppeteer文檔，pyppeteer 只是在 puppeteer之上稍微包裝了下而已 。

注意：本來 chrome 就問題多多，puppeteer 也是各種坑，加上 pyppeteer 是基于前者的改編 python 版本，也就是產生了只要前兩個有一個有 bug，那么 pyppeteer 就會原封不動的繼承下來，本來這沒什么，但是現在遇到的問題就是 pyppeteer 這個項目從2018年9月份之后幾乎沒更新過，前兩者都在不斷的更新迭代，而 pyppeteer 一直不更新，導致很多 bug 根本沒人修復。

?

下面簡單介紹下Pyppeteer的兩大特點:chromium瀏覽器和asyncio框架:

1).chromium

Chromium 是一款獨立的瀏覽器，是 Google 為發展自家的瀏覽器 Google Chrome 而開啟的計劃，相當于 Chrome的實驗版，且 Chromium 是完全開源的。二者基于相同的源代碼構建，Chrome 所有的新功能都會先在 Chromium 上實現，待驗證穩定后才會移植，因此 Chromium 的版本更新頻率更高，也會包含很多新的功能，但作為一款獨立的瀏覽器，Chromium 的用戶群體要小眾得多。兩款瀏覽器“同根同源”，它們有著同樣的 Logo，但配色不同，Chrome 由藍紅綠黃四種顏色組成，而 Chromium 由不同深度的藍色構成。

Pyppeteer 的 web 自動化是基于 chromium 來實現的，由于 chromium 中某些特性的關系，Pyppeteer 的安裝配置非常簡單，關于這一點稍后我們會詳細介紹。

?

2).asyncio

asyncio 是 Python 的一個異步協程庫，自3.4版本引入的標準庫，直接內置了對異步IO的支持，號稱是Python最有野心的庫，官網上有非常詳細的介紹：https://docs.python.org/3/library/asyncio.html

安裝與使用

?

由于 Pyppeteer 采用了 Python 的 async 機制，所以其運行要求的?Python 版本為 3.5 及以上。

?

1).極簡安裝

使用?pip3 install pyppeteer?命令就能完成 pyppeteer 庫的安裝，至于 chromium 瀏覽器，只需要一條 pyppeteer-install 命令就會自動下載對應的最新版本 chromium 瀏覽器到 pyppeteer 的默認位置。

window 下 安裝完?pyppeteer ，會在 python 安裝目錄下的?Scripts 目錄下 有?pyppeteer-install.exe?和?pyppeteer-install-script.py 兩個文件，執行 任意一個都可以安裝?chromium 瀏覽器到 pyppeteer 的默認位置。

運行?pyppeteer-install.exe ：

如果不運行 pyppeteer-install 命令，在第一次使用 pyppeteer 的時候也會自動下載并安裝 chromium 瀏覽器，效果是一樣的。總的來說，pyppeteer 比起 selenium 省去了 driver 配置的環節。

當然，出于某種原因（需要梯子，或者科學上網），也可能會出現chromium自動安裝無法順利完成的情況，這時可以考慮手動安裝：首先，從下列網址中找到自己系統的對應版本，下載chromium壓縮包；

'linux': 'https://storage.googleapis.com/chromium-browser-snapshots/Linux_x64/575458/chrome-linux.zip'
'mac': 'https://storage.googleapis.com/chromium-browser-snapshots/Mac/575458/chrome-mac.zip'
'win32': 'https://storage.googleapis.com/chromium-browser-snapshots/Win/575458/chrome-win32.zip'
'win64': 'https://storage.googleapis.com/chromium-browser-snapshots/Win_x64/575458/chrome-win32.zip'

然后，將壓縮包放到pyppeteer的指定目錄下解壓縮，windows系統的默認目錄。

其他系統下的默認目錄可以參照下面：

• Windows:?C:\Users\<username>\AppData\Local\pyppeteer
• OS X:?/Users/<username>/Library/Application?Support/pyppeteer
• Linux:?/home/<username>/.local/share/pyppeteer
  • or in?$XDG_DATA_HOME/pyppeteer?if?$XDG_DATA_HOME?is defined.

Details see?appdirs’s?user_data_dir.

好了，安裝完成之后我們命令行下測試下：
>>> import pyppeteer
如果沒有報錯，那么就證明安裝成功了。

?

2).使用

Pyppeteer 是一款非常高效的 web 自動化測試工具，由于 Pyppeteer 是基于 asyncio 構建的，它的所有 屬性 和方法 幾乎都是 coroutine?(協程)?對象，因此在構建異步程序的時候非常方便，天生就支持異步運行。

程序構建的基本思路是新建?一個 browser 瀏覽器?和?一個 頁面 page。

看下面這段代碼，在 main 函數中，先是建立一個瀏覽器對象，然后打開新的標簽頁，訪問百度主頁，對當前頁面截圖并保存為“example.png”，最后關閉瀏覽器。前文也提到過，pyppeteer 是基于 asyncio 構建的，所以在使用的時候需要用到?async/await?結構。

import asyncio from pyppeteer import launchasync def main():browser = await launch()page = await browser.newPage()await page.goto('http://baidu.com')await page.screenshot({'path': 'example.png'})await browser.close()asyncio.get_event_loop().run_until_complete(main())

運行上面這段代碼會發現并沒有瀏覽器彈出運行，這是因為 Pyppeteer 默認使用的是無頭瀏覽器，如果想要瀏覽器顯示，需要在launch 函數中設置參數 “headless =False”，程序運行結束后在同一目錄下會出現截取到的網頁圖片：

?

遇到的錯誤

?

• 1）pyppeteer.errors.NetworkError: Protocol error Network.getCookies: Target close

控制訪問指定 url 之后 await page.goto(url)，會遇到上面的錯誤，如果這時候使用了 sleep 之類的延時也會出現這個錯誤或者類似的 time out。?
這個問題是 puppeteer 的 bug，但是對方已經修復了，而 pyppeteer 遲遲沒更新，就只能靠自己了，搜了很多人的文章，例如：https://github.com/miyakogi/pyppeteer/issues/171?，但是我按照這個并沒有成功。也有人增加一個函數，但調用這個參數依然沒解決問題。

async def scroll_page(page):cur_dist = 0height = await page.evaluate("() => document.body.scrollHeight")while True:if cur_dist < height:await page.evaluate("window.scrollBy(0, 500);")await asyncio.sleep(0.1)cur_dist += 500else:break

可以把 python 第三方庫 websockets 版本 7.0 改為 6.0 就可以了，親測可用。

pip uninstall websockets #卸載websocketspip install websockets==6.0 或者 pip install websockets==6.0 --force-reinstall #指定安裝6.0版本

• 2）chromium瀏覽器多開頁面卡死問題。

解決這個問題的方法就是瀏覽器初始化的時候添加'dumpio':True。

# 啟動 pyppeteer 屬于內存中實現交互的模擬器

browser = await launch({'headless': False, 'args': ['--no-sandbox'], 'dumpio': True})

• 3）瀏覽器窗口很大，內容顯示很小。

需要設置瀏覽器顯示大小，默認就是無法正常顯示。可以看到頁面左側右側都是空白，網站內容并沒有完整鋪滿chrome.

# Pyppeteer支持字典和關鍵字傳參，Puppeteer只支持字典傳參 # 這里使用字典傳參 browser = await launch({'headless': False,'dumpio': True,'autoClose': False,'args': ['--no-sandbox','--window-size=1366,850']} ) await page.setViewport({'width': 1366, 'height': 768})

通過上面設置Windows-size和Viewport大小來實現網頁完整顯示。

但是對于那種向下無線加載的長網頁這種情況如果瀏覽器是可見狀態會顯示不全，針對這種情況的解決方法就是賦值當前網頁新開一個標簽頁粘貼進去就正常了。

Pyppeteer 和 Puppeteer 的 不同點

• Pyppeteer支持字典和關鍵字傳參，Puppeteer只支持字典傳參

# Puppeteer只支持字典傳參 browser = await launch({'headless': True})# Pyppeteer支持字典和關鍵字傳參 browser = await launch({'headless': True}) browser = await launch(headless=True)

• 元素選擇器方法名 $變為querySelector

# Puppeteer使用$符號 Page.$()/Page.$$()/Page.$x()# Pyppeteer使用Python風格的函數名 Page.querySelector()/Page.querySelectorAll()/Page.xpath() # 簡寫方式為: Page.J(), Page.JJ(), and Page.Jx()

• Page.evaluate() 和 Page.querySelectorEval()的參數

Puppeteer的evaluate()方法使用JavaScript原生函數或JavaScript表達式字符串。Pyppeteer的evaluate()方法只使用JavaScript字符串，該字符串可以是函數也可以是表達式，Pyppeteer會進行自動判斷。但有時會判斷錯誤，如果字符串被判斷成了函數，并且報錯，可以添加選項force_expr=True, 強制Pyppeteer作為表達式處理。

獲取頁面內容：

content = await page.evalute('document.body.textContent', force_expr=True)

獲取元素的內部文字:

element = await page.querySelector('h1') title = await page.evaluate('(element) => element.textContent', element)

基礎用法

?

抓取內容 ?可以使用 xpath 表達式

""" # Pyppeteer三種解析方式 Page.querySelector() # 選擇器 Page.querySelectorAll() Page.xpath() # xpath表達式 # 簡寫方式為: Page.J(), Page.JJ(), and Page.Jx() """

示例1:

import asyncio from pyppeteer import launchasync def main():# headless參數設為False，則變成有頭模式# Pyppeteer支持字典和關鍵字傳參，Puppeteer只支持字典傳參# 指定引擎路徑# exepath = r'C:\Users\Administrator\AppData\Local\pyppeteer\pyppeteer\local-chromium\575458\chrome-win32/chrome.exe'# browser = await launch({'executablePath': exepath, 'headless': False, 'slowMo': 30})browser = await launch(# headless=False,{'headless': False})page = await browser.newPage()# 設置頁面視圖大小await page.setViewport(viewport={'width': 1280, 'height': 800})# 是否啟用JS, enabled設置為False, 則無渲染效果await page.setJavaScriptEnabled(enabled=True)# 超時間見 1000 毫秒res = await page.goto('https://www.toutial.com/', options={'timeout': 1000})resp_headers = res.headers # 響應頭resp_status = res.status # 響應狀態# 等待await asyncio.sleep(2)# 第二種方法，在while循環里強行查詢某元素進行等待while not await page.querySelector('.t'):pass# 滾動到頁面底部await page.evaluate('window.scrollBy(0, document.body.scrollHeight)')await asyncio.sleep(2)# 截圖 保存圖片await page.screenshot({'path': 'toutiao.png'})# 打印頁面cookiesprint(await page.cookies())""" 打印頁面文本 """# 獲取所有html內容print(await page.content())# 在網頁上執行js 腳本dimensions = await page.evaluate(pageFunction='''() => {return {width: document.documentElement.clientWidth, // 頁面寬度height: document.documentElement.clientHeight, // 頁面高度deviceScaleFactor: window.devicePixelRatio, //像素比 1.0000000149011612}}''', force_expr=False) # force_expr=False 執行的是函數print(dimensions)# 只獲取文本執行js腳本force_expr 為True則執行的是表達式content = await page.evaluate(pageFunction='document.body.textContent', force_expr=True)print(content)# 打印當前頁標題print(await page.title())# 抓取新聞內容 可以使用xpath表達式"""# Pyppeteer 三種解析方式Page.querySelector() # 選擇器Page.querySelectorAll()Page.xpath() # xpath表達式# 簡寫方式為:Page.J(), Page.JJ(), and Page.Jx()"""element = await page.querySelector(".feed-infinite-wrapper > ul>li") # 只抓取一個print(element)# 獲取所有文本內容 執行jscontent = await page.evaluate('(element) => element.textContent', element)print(content)# elements = await page.xpath('//div[@class="title-box"]/a')elements = await page.querySelectorAll(".title-box a")for item in elements:print(await item.getProperty('textContent'))# <pyppeteer.execution_context.JSHandle object at 0x000002220E7FE518># 獲取文本title_str = await(await item.getProperty('textContent')).jsonValue()# 獲取鏈接title_link = await(await item.getProperty('herf')).jsonValue()print(title_str)print(title_link)# 關閉瀏覽器await browser.close()asyncio.get_event_loop().run_until_complete(main())

示例2:

import asyncio import pyppeteer from collections import namedtupleheaders = {'date': "Sun, 28 Apr 2019 06:50:20 GMT','server': "Cmcc','x-frame-options': "SAMORIGIN\nSAMEORIGIN','last-modified': 'Fir, 26 Apr 2019 09:58:09 GMT','accept-ranges': 'bytes','cache-control': 'max-age=42300','expires': 'Sun, 28 Apr 2019 18:50:20 GMT','vary': 'Accept-Encoding,User-Agent','content-encoding": 'gzip','content-length': 'gzip','content-length': '19823','connection': 'Keep-alive','via': '1.1 ID-0314217270751344 uproxy-17' }Response = namedtuple("rs", "title url html cookies headers history status")async def get_html(url):browser = await pyppeteer.launch(headless=True, args=['--no-sandbox'])page = await browser.newPage()res = await page.goto(url, options={'timeout': 10000})data = await page.content()title = await page.title()resp_cookies = await page.cookies() # cookieresp_headers = res.headers # 響應頭resp_status = res.status # 響應狀態print(data)print(title)print(resp_headers)print(resp_status)return titleif __name_ '__main__':url_list = ["https://www.toutiao.com","http://jandan.net/ooxx/page-8#comments","https://www.12306.cn/index"]task = [get_html(url) for url in url_list]loop = asyncio.get_event_loop()results = loop.run_until_complete(asyncio.gather(*task))for res in results:print(res)

模擬輸入

模擬輸入文本：

# 模擬輸入 賬號密碼 {'delay': rand_int()} 為輸入時間 await page.type('#TPL_username_1', "sadfsdfsadf") await page.type('#TPL_password_1', "123456789",)await page.waitFor(1000) await page.click("#J_SubmitStatic")

使用tkinter獲取頁面高度 寬度

def screen_size():"""使用tkinter獲取屏幕大小"""import tkintertk = tkinter.Tk()width = tk.winfo_screenwidth()height = tk.winfo_screenheight()tk.quit()return width, height

爬取京東商城

示例代碼：

import requests from bs4 import BeautifulSoup from pyppeteer import launch import asynciodef screen_size():"""使用tkinter獲取屏幕大小"""import tkintertk = tkinter.Tk()width = tk.winfo_screenwidth()height = tk.winfo_screenheight()tk.quit()return width, heightasync def main(url):# browser = await launch({'headless': Flase, 'args': ['--no-sandbox'], })browser = await launch({'args': ['--no-sandbox'], })page = await browser.newPage()width, height = screen_size()await page.setViewport(viewport={"width": width, "height": height})await page.setJavaScriptEnabled(enabled=True)await page.setUserAgent('Mozilla/5.0 (windows NT 10.0; Win64; x64) AppleWebKit/537.36 ''(KHTML, like Geck) Chrome/58.0.3029.110 Safri/537.36 Edge/16.16299')await page.goto(url)# await asyncio.sleep(2)await page.evaluate('window.scrollBy(0, document.body.scrollHeight)')await asyncio.sleep(1)# content = await page.content()li_list = await page.xpath('//*[@id="J_goodsList"]/url/li')# print(li_list)item_list = []for li in li_list:a = await li.xpath('.//div[@class="p-img"]/a')detail_url = await (await a[0].getProperty("href")).jsonValue()promo_words = await (await a[0].getProperty("title")).jsonValue()a_ = await li.xpath('.//div[@class="p-commit"]/strong/a')p_commit = await (await a_[0].getProperty("textContent")).jsonValue()i = await li.xpath('./div/div[3]/strong/i')price = await (await i[0].getProperty("textContent")).jsonValue()em = await li.xpath('./div/div[4]/a/em')title = await (await em[0].getProperty("textContent")).jsonValue()item = {"title": title,"detail_url": detail_url,"promo_words": promo_words,'p_commit': p_commit,'price': price}item_list.append(item)# print(item)# break# print(content)await page_close(browser)return item_listasync def page_close(browser):for _page in await browser.pages():await _page.close()await browser.close()msg = "手機"url = "https://search.jd.com/Search?keyword={}&enc=utf-8&qrst=1&rt=1&stop=1&vt=2&wq={}&cid2=653&cid3=655&page={}" task_list = [] for i in range(1, 6):page = i * 2 - 1url = url.format(msg, msg, page)task_list.append(main(url))loop = asyncio.get_event_loop() results = loop.run_until_complete(asyncio.gather(*task_list)) # print(results, len(results)) for i in results:print(i, len(i))print('*' * 100) # soup = BeautifulSoup(content, 'lxml') # div = soup.find('div', id='J_goodsList') # for i, li in enumerate(div.find_all('li', class_='gl-item')): # if li.select('.p-img a'): # print(li.select('.p-img a')[0]['href'], i) # print(li.select('.p-price i')[0].get_text(), i) # print(li.select('.p-name em')[0].text, i) # else: # print("#" * 200) # print(li)

抓取淘寶

示例代碼：

# -*- coding: urf-8 -*-import time import random import asyncio from retrying import retry # 錯誤自動重試 from pyppeteer.launcher import launchjs1 = '''() =>{Object.defineProperties(navigator, {webdriver:{ get: () => false}})}''' js2 = '''() => {alert(window.navigator.webdriver)}''' js3 = '''() => {window.navigator.chrome = {runtime: {}, }; }''' js4 = '''() => {Object.defineProperty(navigator, 'languages', {get: () => ['en-US', 'en']});}''' js5 = '''() => {Object.defineProperty(navigator, 'plugins', {get: () => [1, 2, 3, 4, 5, 6],});}'''def retry_if_result_none(result):return result is None@retry(retry_on_result=retry_if_result_none, ) async def mouse_slide(page=None):await asyncio.sleep(3)try:await page.hover('#nc_1_n1z')await page.mouse.down()await page.mouse.move(2000, 0, {'delay': random.randint(1000, 2000)})await page.mouse.up()except Exception as e:print(e, ' : slide login False')return Noneelse:await asyncio.sleep(3)slider_again = await page.Jeval('.nc-lang-cnt', 'node => node.textContent')if slider_again != '驗證通過':return Noneelse:await page.screenshot({'path': './headless-slide-result.png'})print('驗證通過')return 1 def input_time_random():return random.randint(100, 151)def screen_size():"""使用tkinter獲取屏幕大小"""import tkintertk = tkinter.Tk()width = tk.winfo_screenwidth()height = tk.winfo_screenheight()tk.quit()return width, heightasync def main(username, pwd, url):browser = await launch({'headless': False, 'args': ['--no-sandbox'], },userDataDir = './userdata',args=['--window-size=1366,768'])page = await browser.newPage()width, height = screen_size()await page.setViewport(viewport={"width": width, "height": height})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(url)await page.evaluate(js1)await page.evaluate(js3)await page.evaluate(js4)await page.evaluate(js5)pwd_login = await page.querySelector('.J_Quick2Static')# print(await (await pwd_Login.getProperty('textContent')).jsonValue())await pwd_login.click()await page.type('#TPL_username_1', username, {'delay': input_time_random() - 50})await page.type('#TPL_password_1', pwd, {'delay': input_time_random()})await page.screenshot({'path': './headless-test-result.png'})time.sleep(2)slider = await page.Jeval('#nocaptcha', 'node => node.style') # 是否有滑塊if slider:print('出現滑塊情況判定')await page.screenshot({'path': './headless-login-slide.png'})flag = await mouse_slide(page=page)if flag:print(page.url)await page.keyboard.press('Enter')await get_cookie(page)else:await page.keyboard.press('Enter')await page.waitFor(20)await page.waitForNavigation()try:global errorerror = await page.Jeval('.error', 'node => node.textContent')except Exception as e:error = Noneprint(e, "錯啦")finally:if error:print('確保賬戶安全重新輸入')else:print(page.url)# 可繼續網頁跳轉 已經攜帶 cookie# await get_search(page)await get_cookie(page)await page_close(browser)async def page_close(browser):for _page in await browser.pages():await _page.close()await browser.close()async def get_search(page):# https://s.taobao.com/search?q={查詢的條件}&p4ppushleft=1%2C48&s={每頁 44 條 第一頁 0 第二頁 44}&sort=sale-descawait page.goto("https://s.taobao.com/search?q=氣球")await asyncio.sleep(5)# print(await page.content())# 獲取登錄后cookie async def get_cookie(name):res = await page.content()cookies_list = await page.cookies()cookies = ''for cookie in cookies_list:str_cookie = '{0}={1};'str_cookie = str_cookie.format(cookie.get('name'), cookie.get('value'))cookies += str_cookieprint(cookies)# 將coookie放入cookie池 以便多次請求 封賬號 利用cookie對搜索內容進行爬取return cookiesif __name__ == '__main__':tb_username = '淘寶用戶名'tb_pwd = '淘寶密碼'tb_url = "https://login.taobao.com/member/login.jhtml"loop = asyncio.get_event_loop()loop.run_until_complete(main(tb_username, tb_pwd, tb_url))

利用 上面 獲取到的 cookie 爬取搜索內容

示例代碼：

import json import requests import re# 設置cookie 池 隨機發送請求 通過pyppeteer獲取cookie cookie = '_tb_token_=edd7e354dee53;t=fed8f4ca1946ca1e73223cfae04bc589;sg=20f;cna=2uJSFdQGmDMCAbfFWXWAC4Jv;cookie2=1db6cd63ad358170ea13319f7a862c33;_l_g_=Ug%3D%3D;v=0;unb=3150916610;skt=49cbfd5e01d1b550;cookie1=BxVRmD3sh19TaAU6lH88bHw5oq%2BgcAGcRe229Hj5DTA%3D;csg=cf45a9e2;uc3=vt3=F8dByEazRMnQZDe%2F9qI%3D&id2=UNGTqfZ61Z3rsA%3D%3D&nk2=oicxO%2BHX4Pg%3D&lg2=U%2BGCWk%2F75gdr5Q%3D%3D;existShop=MTU1Njg3MDM3MA%3D%3D;tracknick=%5Cu7433150322;lgc=%5Cu7433150322;_cc_=V32FPkk%2Fhw%3D%3D;mt=ci=86_1;dnk=%5Cu7433150322;_nk_=%5Cu7433150322;cookie17=UNGTqfZ61Z3rsA%3D%3D;tg=0;enc=tThHs6Sn3BAl8v1fu3J4tMpgzA1n%2BLzxjib0vDAtGsXJCb4hqQZ7Z9fHIzsN0WghdcKEsoeKz6mBwPUpyzLOZw%3D%3D;JSESSIONID=B3F383B3467EC60F8CA425935232D395;l=bBMspAhrveV5732DBOCanurza77OSIRYYuPzaNbMi_5pm6T_G4QOlC03xF96VjfRswYBqh6Mygv9-etuZ;hng=CN%7Czh-CN%7CCNY%7C156;isg=BLi41Q8PENDal3xUVsA-aPbfiWaKiRzB6vcTu_IpBPOmDVj3mjHsO86vxUQYW9SD;uc1=cookie16=W5iHLLyFPlMGbLDwA%2BdvAGZqLg%3D%3D&cookie21=W5iHLLyFeYZ1WM9hVnmS&cookie15=UIHiLt3xD8xYTw%3D%3D&existShop=false&pas=0&cookie14=UoTZ4ttqLhxJww%3D%3D&tag=8&lng=zh_CN;thw=cn;x=e%3D1%26p%3D*%26s%3D0%26c%3D0%26f%3D0%26g%3D0%26t%3D0;swfstore=34617;'headers = {'cookie': cookie,"user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/73.0.3683.103 Safari/537.36" }rep = requests.get('https://s.taobao.com/search?q=手機&p4ppushleft=1%2C48&s=0&sort=sale-desc ', headers=headers) rep.encoding = 'utf-8' res = rep.text print(res)r = re.compile(r'g_page_config = (.*?)g_srp_loadCss', re.S) res = r.findall(res)data = res[0].strip().rstrip(';') dic_data = json.loads(data) auctions = dic_data.get('mods')['itemlist']['data']['auctions']# print(auctions, len(auctions)) for item in auctions[1:]:print(item)break

針對iframe 的操作

• page.frames 獲取所有的 iframe 列表 需要判斷操作的是哪一個 iframe 跟操作 page 一樣操作

from pyppeteer import launch import asyncioasync def main(url):w = await launch({'headless': False, 'args': ['--no-sandbox'], })page = await w.newPage()await page.setViewport({"width": 1366, 'height': 800})await page.goto(url)try:await asyncio.sleep(1)frame = page.framesprint(frame) # 需要找到哪一個frametitle = await frame[1].title()print(title)await asyncio.sleep(1)login = await frame[1].querySelector('#switcher_plogin')print(login)await login.click()await asyncio.sleep(20)except Exception as e:print(e, "EEEEEEEEE")for _page in await w.pages():await _page.close()await w.close()asyncio.get_event_loop().run_until_complete(main("https://i.qq.com/?rd=1")) # asyncio.get_event_loop().run_until_complete(main("https://www.gushici.com/"))

與 scrapy 的整合

加入downloadmiddleware

from scrapy import singals from scrapy.downloadermiddlewares.useragent import UserAgentMiddleware import random import pyppeteer import asyncio import os from scrapy.http import HtmlResponsepyppeteer.DEBUG = Falseclass FundscrapyDownloaderMiddleware(object):# Not all methods nned to be defined. If a method is not defined,# scrapy acts as if the downloader middleware does not modify the # passed objects.def __init__(self):print("Init downloaderMiddleware use pypputeer.")os.environ['PYPPETEER_CHROMIUM_REVISION'] = '588429'# pyppeteer.DEBUG = Falseprint(os.environ.get('PYPPETEER_CHROMIUM_REVISION'))loop = asyncio.get_event_loop()task = asyncio.ensure_future(self.getbrowser())loop.run_until_complete(task)# self.browser = task.result()print(self.browser)print(self.page)# self.page = await browser.newPage()async def getbrowser(self):self.browser = await pyppeteer.launch()self.page = await self.browser.newPage()# return await pyppeteer.launch()async def getnewpage(self):return await self.browser.newPage()@classmethoddef from_crawler(cls, crawler):# This method is used by Scrapy to create your spiders.s = cls()crawler.singlas.connect(s.spider_opened, signal=signals.spider_opened)return sdef process_request(self, request, spider):# Called for each request that goes through the downloader# middleware# Must either:# - return None: continue processing this request# - or return a Response object# - or return a Request object# - or raise IgnoreRequest: process_exception() methods of# installed downloader middleware will be calledloop = asyncio.get_event_loop()task = asyncio.ensure_future(self.usePypuppeteer(request))loop.run_until_complete(task)# return task.result()return HtmlResponse(url=request.url, body=task.result(), encoding="utf-8",request=request)async def usePypuppeteer(self, request):print(request.url)# page = await self.browser.newPage()await self.page.goto(request.url)content = await self.page.content()return content def process_response(self, request, response, spider):# Called with the response returned from the downloader.# Must either:# - return a Response object# - return a Request object# - or raise IgnoreRequestreturn responsedef process_exception(self, request, exception, spider):# Called when a download handler or a process_request()# (from other downloader middleware) raises an exception.# Must either:# - return None: continue processing this exception# - return a Response object: stops process_exception() chain# - return a Request object: stops process_exception() chainpassdef spider_opened(self, spider):spider.logger.info('Spider opened: %s' %spider.name)

實戰 異步爬取

?

示例 1 ：快速上手

接下來我們測試下基本的頁面渲染操作，這里我們選用的網址為:http://quotes.toscrape.com/js/,這個頁面是JavaScript渲染而成的，用基本的requests庫請求得到的HTML結果里面是不包含頁面中所見的條目內容的。

為了證明requests無法完成正常的抓取，我們可以先用如下代碼來測試一下:

import requests from pyquery import PyQuery as pqurl = 'http://quotes.toscrape.com/js/' response = requests.get(url=url) doc = pq(response.text) print('Quotes : {0}'.format(doc('.quote').length))# 結果 # Quotes : 0

這里首先使用requests來請求網頁內容，然后使用pyquery來解析頁面中的每一個條目。觀察源碼之后我們發現每個條目的class名為quote，所以這里選用了.quote這個CSS選擇器來選擇，最后輸出條目數量。

運行結果: Quotes: 0

結果是0，這就證明使用requests是無法正常抓取到相關數據的。

為什么？

因為這個頁面是JavaScript渲染而成的，我們所看到的內容都是網頁加載后又執行了JavaScript之后才呈現出來的，因此這些條目數據并不存在于原始HTML代碼中，而requests僅僅抓取的是原始HTML代碼。

好的，所以遇到這種類型的網站我們應該怎么辦呢?

其實答案有很多:

分析網頁源代碼數據，如果數據是隱藏在 HTML 中的其他地方，以 JavaScript 變量的形式存在，直接提取就好了。

分析 Ajax，很多數據可能是經過 Ajax 請求時候獲取的，所以可以分析其接口。

模擬 JavaScript 渲染過程，直接抓取渲染后的結果。

而 Pyppeteer 和 Selenium 就是用的第三種方法，下面我們再用 Pyppeteer 來試試，如果用 Pyppeteer 實現如上頁面的抓取的話，代碼就可以寫為如下形式：

import asyncio from pyppeteer import launch from pyquery import PyQuery as pqasync def main():browser = await launch()page = await browser.newPage()url = 'http://quotes.toscrape.com/js/'await page.goto(url=url)doc = pq(await page.content())print('Quotes : {0}'.format(doc('.quote').length))await browser.close()asyncio.get_event_loop().run_until_complete(main())

運行結果：Quotes: 10
看運行結果，這說明我們就成功匹配出來了 class 為 quote 的條目，總數為 10 條，具體的內容可以進一步使用 pyquery 解析查看。

那么這里面的過程發生了什么？

實際上，Pyppeteer 整個流程就完成了瀏覽器的開啟、新建頁面、頁面加載等操作。另外 Pyppeteer 里面進行了異步操作，所以需要配合 async/await 關鍵詞來實現。首先， launch 方法會新建一個 Browser 對象，然后賦值給 browser，然后調用 newPage 方法相當于瀏覽器中新建了一個選項卡，同時新建了一個 Page 對象。然后 Page 對象調用了 goto 方法就相當于在瀏覽器中輸入了這個 URL，瀏覽器跳轉到了對應的頁面進行加載，加載完成之后再調用 content 方法，返回當前瀏覽器頁面的源代碼。然后進一步地，我們用 pyquery 進行同樣地解析，就可以得到 JavaScript 渲染的結果了。另外其他的一些方法如調用 asyncio 的 get_event_loop 等方法的相關操作則屬于 Python 異步 async 相關的內容了，大家如果不熟悉可以了解下 Python 的 async/await 的相關知識。好，通過上面的代碼，我們就可以完成 JavaScript 渲染頁面的爬取了。

模擬網頁截圖，保存 PDF，執行自定義的 JavaScript 獲得特定的內容

接下來我們再看看另外一個例子，這個例子可以模擬網頁截圖，保存PDF，另外還可以執行自定義的JavaScript獲得特定的內容，代碼如下:

import asyncio from pyppeteer import launchasync def main():browser = await launch()page = await browser.newPage()url = 'http://quotes.toscrape.com/js/'await page.goto(url=url)await page.screenshot(path='test_screenshot.png')await page.pdf(path='test_pdf.pdf')# 在網頁上執行js 腳本dimensions = await page.evaluate(pageFunction='''() => {return {width: document.documentElement.clientWidth, //頁面寬度height: document.documentElement.clientHeight, //頁面高度deviceScaleFactor: window.devicePixelRatio, // 像素比 1.0000000149011612}}''', force_expr=False) # force_expr=False 執行的是函數print(dimensions)await browser.close()asyncio.get_event_loop().run_until_complete(main())# 結果 # {'width': 800, 'height': 600,'deviceScaleFactor': 1}

這里我們又用到了幾個新的 API，完成了網頁截圖保存、網頁導出 PDF 保存、執行 JavaScript 并返回對應數據。

?evaluate 方法執行了一些 JavaScript，JavaScript 傳入的是一個函數，使用 return 方法返回了網頁的寬高、像素大小比率三個值，最后得到的是一個 JSON 格式的對象。

總之利用 Pyppeteer 我們可以控制瀏覽器執行幾乎所有動作，想要的操作和功能基本都可以實現，用它來自由地控制爬蟲當然就不在話下了。

了解了基本的實例之后，我們再來梳理一下 Pyppeteer 的一些基本和常用操作。Pyppeteer 的幾乎所有功能都能在其官方文檔的 API Reference 里面找到，鏈接為：https://miyakogi.github.io/pyppeteer/reference.html，用到哪個方法就來這里查詢就好了，參數不必死記硬背，即用即查就好。

登錄淘寶 （打開網頁后，手動輸入用戶名和密碼，可以看到正常跳轉到登錄后的頁面）：

import asyncio from pyppeteer import launchwidth, height = 1366, 768js1 = '''() =>{Object.defineProperties(navigator,{ webdriver:{ get: () => false}})}''' js2 = '''() => {alert(window.navigator.webdriver)}''' js3 = '''() => {window.navigator.chrome = {runtime: {}, }; }''' js4 = '''() =>{Object.defineProperty(navigator, 'languages', {get: () => ['en-US', 'en']});}''' js5 = '''() =>{Object.defineProperty(navigator, 'plugins', {get: () => [1, 2, 3, 4, 5,6],});}'''async def page_evaluate(page):# 替換淘寶在檢測瀏覽時采集的一些參數# 需要注意，在測試的過程中發現登陸成功后頁面的該屬性又會變成True# 所以在每次重新加載頁面后要重新設置該屬性的值。await page.evaluate('''() =>{ Object.defineProperties(navigator,{ webdriver:{ get: () => false } }) }''')await page.evaluate('''() =>{ window.navigator.chrome = { runtime: {}, }; }''')await page.evaluate('''() =>{ Object.defineProperty(navigator, 'languages', { get: () => ['en-US', 'en'] }); }''')await page.evaluate('''() =>{ Object.defineProperty(navigator, 'plugins', { get: () => [1, 2, 3, 4, 5,6], }); }''')async def main():browser = await launch(headless=False,# userDataDir='./userdata',args=['--disable-infobars', f'--window-size={width},{height}', '--no-sandbox'])page = await browser.newPage()await page.setViewport({"width": width,"height": height})url = 'https://www.taobao.com'await page.goto(url=url)# await page.evaluate(js1)# await page.evaluate(js3)# await page.evaluate(js4)# await page.evaluate(js5)await page_evaluate(page)await asyncio.sleep(100)# await browser.close()asyncio.get_event_loop().run_until_complete(main())

如果把上面 js 去掉，發現淘寶可以檢測出來， 跳轉不到登錄后的頁面。

window.navigator 對象包含有關訪問者瀏覽器的信息：https://www.runoob.com/js/js-window-navigator.html

js 主要需要修改瀏覽器的 window.navigator.webdriver、window.navigator.languages等值。

打開正常的瀏覽器可以看到：

window.navigator.webdriver的值為undefined，而通過pyppeteer控制打開的瀏覽器該值為True，當被檢測到該值為True的時候，則滑動會一直失敗，所以我們需要修改該屬性。需要注意，在測試的過程中發現登陸成功后頁面的該屬性又會變成True，所以在每次重新加載頁面后要重新設置該屬性的值。

async def page_evaluate(page):# 替換淘寶在檢測瀏覽時采集的一些參數await page.evaluate('''() =>{ Object.defineProperties(navigator,{ webdriver:{ get: () => false } }) }''')await page.evaluate('''() =>{ window.navigator.chrome = { runtime: {}, }; }''')await page.evaluate('''() =>{ Object.defineProperty(navigator, 'languages', { get: () => ['en-US', 'en'] }); }''')await page.evaluate('''() =>{ Object.defineProperty(navigator, 'plugins', { get: () => [1, 2, 3, 4, 5,6], }); }''')

另一種方法可以進一步免去淘寶登錄的煩惱，那就是設置用戶目錄。

平時我們已經注意到，當我們登錄淘寶之后，如果下次再次打開瀏覽器發現還是登錄的狀態。這是因為淘寶的一些關鍵Cookies已經保存到本地了，下次登錄的時候可以直接讀取并保持登錄狀態。

那么這些信息保存在哪里了呢?其實就是保存在用戶目錄下了，里面不僅包含了瀏覽器的基本配置信息，還有一些Cache、Cookies等各種信息都在里面，如果我們能在瀏覽器啟動的時候讀取這些信息，那么啟動的時候就可以恢復一些歷史記錄甚至一些登錄狀態信息了。

這也就解決了一個問題:很多朋友在每次啟動 Selenium 或 Pyppeteer 的時候總是是一個全新的瀏覽器，那就是沒有設置用戶目錄，如果設置了它，每次打開就不再是一個全新的瀏覽器了，它可以恢復之前的歷史記錄，也可以恢復很多網站的登錄信息。

當然可能時間太久了，Cookies 都過期了，那還是需要登錄的。

那么這個怎么來做呢？很簡單，在啟動的時候設置 userDataDir 就好了，示例如下：

browser = await launch(headless=False,userDataDir='./userdata',args=['--disable-infobars', f'--window-size={width},{height}'])

好，這里就是加了一個 userDataDir 的屬性，值為 userdata，即當前目錄的 userdata 文件夾。我們可以首先運行一下，然后登錄一次淘寶，這時候我們同時可以觀察到在當前運行目錄下又多了一個 userdata 的文件夾：

用戶文件夾

具體的介紹可以看官方的一些說明，如：https://chromium.googlesource.com/chromium/src/+/master/docs/user_data_dir.md?這里面介紹了 userdatadir 的相關內容。

?

命令行啟動 chrome 并進入指定的 URL：chrome.exe --disable-infobars --user-data-dir="./userdatadir" --new-window https://login.taobao.com/member/login.jhtml

執行完后會打開 淘寶的登錄頁面，登錄淘寶，然后保存用戶名密碼，這樣登錄信息就保存在?userdatadir?目錄下了

在執行?chrome.exe --disable-infobars --user-data-dir="./userdatadir" --new-window https://www.taobao.com

可以看到已經時登錄狀態了。

示例 2 ：爬取今日頭條

# -*- coding: utf-8 -*- # @Author : # @File : toutiao.py # @Software: PyCharm # @description : XXXimport asyncio from pyppeteer import launch from pyquery import PyQuery as pqdef screen_size():"""使用tkinter獲取屏幕大小"""import tkintertk = tkinter.Tk()width = tk.winfo_screenwidth()height = tk.winfo_screenheight()tk.quit()return width, heightasync def main():width, heigth = screen_size()print(f'screen : [width:{width}, height:{height} ]')browser = await launch(headless=False, args=[f'--window-size={width},{height}'])page = await browser.newPage()await page.setViewport({'width': width, 'height': height})# 是否啟用JS, enabled設為False,則無渲染結果await page.setJavaScriptEnabled(enabled=True)await page.goto('https://www.toutiao.com')await asyncio.sleep(5)print(await page.cookies()) # 打印頁面cookiesprint(await page.content()) # 打印頁面文本print(await page.title()) # 打印當前頁標題# 抓取新聞標題title_elements = await page.xpath('//div[@class="title-box"]/a')for item in title_elements:# 獲取文本title_str = await (await item.getProperty('textContent')).jsonValue()print(await item.getProperty('textContent'))# 獲取鏈接title_link = await (await item.getProperty('href')).jsonValue()print(title_str)print(title_link)# 在搜索框中輸入pythonawait page.type(input.tt-input__inner', 'python')# 點擊搜索按鈕await page.click('button.tt-button')await asyncio.sleep(5)# print(page.url)# 今日頭條點擊后新開一個頁面，通過打印url可以看出page還停留在原頁面# 以下用于切換至新頁面pages = await browser.pages()page = pages[-1]# print(page.url)page_source = await page.content()text = pq(page_source)await page.goto(url="https://www.toutiao.com/api/search/content/?""aid=24&app_name=web_search&offset=60&format=json""&keyword=python&autoload=true&count=20&en_qc=1""&cur_tab=1&from=search_tab&pd=synthesis&timestamp=1555589585193")for i in range(1, 10):print(text("#J_section_{} > div > div > div.normal.rbox > div > div.title-box > a > span".format(i)).text())# 關閉瀏覽器await browser.close()asyncio.get_event_loop().run_until_complete(main())

示例代碼2:

import asyncio from pyppeteer import launchasync def main():# headless參數設為False,則變成有頭模式browser = await launch(# headless=False)page = await browser.newPage()# 設置頁面視圖大小await page.setViewport(viewport={'width': 1280, 'height': 800})# 是否啟用JS，enabled設為False，則無渲染效果await page.setJavaScriptEnabled(enabled=True)await page.goto('https://www.toutiao.com/')# 打印頁面cookiesprint(await page.cookies())# 打印頁面文本print(await page.content())# 打印當前頁標題print(await page.title())# 抓取新聞標題title_elements = await page.xpath('//div[@class="title-box"]/a')for item in title_elements:# 獲取文本title_str = await (await item.getProperty('textContent')).jsonValue()print(await item.getProperty('textContent'))# 獲取鏈接title_link = await (await item.getProperty('href')).jsonValue()print(title_str)print(title_link)# 關閉瀏覽器await browser.close()asyncio.get_event_loop().run_until_complete(main())

百度首頁交互

示例代碼：

import time import asyncio from pyppeteer import launchasync def main():browser = await launch(headless=False)page = await browser.newPage()await page.setViewport({'width': 1200, 'height': 800})await page.goto('https://www.baidu.com')# 在搜索框中輸入pythonawait page.type('input#kw.s_ipt', 'python')# 點擊搜索按鈕await page.click('input#su')# 等待元素加載，第一種辦法，強行等待5秒# await asyncio.sleep(5)# 第二種方法，在while循環里強行查詢某元素進行等待while not await page.querySelector('.t'):pass# 滾動到頁面底部await page.evaluate('window.scrollBy(0, window.innerHeight)')# 這些等待方法都不好用# await page.waitForXPath('h3', timeout=300)# await page.waitForNavigation(waitUntil="networkidle0")# await page.waitForFunction('document.getElementByTag("h3")')# await page.waitForSelector('.t')# await page.waitFor('document.querySelector("#t")')# await page.waitForNavigation(waitUntil='networkidle0')# await page.waitForFunction('document.querySelector("").inner??Text.length == 7')title_elements = await page.xpath('//h3[contains(@class, "t")]/a')for item in title_elements:title_str = await (await item.getProperty('textContent')).jsonValue()print(title_str)await browser.close()asyncio.get_event_loop().run_until_complete(main())

示例:

import asyncio import pyppeteer from collections import namedtupleResponse = namedtuple("rs", "title url html cookies headers history status")async def get_html(url, timeout=30):# 默認30sbrowser = await pyppeteer.launch(headless=True, args=['--no-sandbox'])page = await browser.newPage()res = await page.goto(url, options={'timeout': int(timeout * 1000)})data = await page.content()title = await page.title()resp_cookies = await page.cookies()resp_headers = res.headersresp_history = Noneresp_status = res.statusresponse = Response(title=title,url=url,html=data,cookies=resp_cookies,headers=resp_headers,history=resp_history,status=resp_status)return responseif __name__ == '__main__':url_list = ["http://www.10086.cn/index/tj/index_220_220.html","http://www.10010.com/net5/011/",# "http://python.jobbole.com/87541/"]task = (get_html(url) for url in url_list)loop - asyncio.get_event_loop()results = loop.run_until_complete(asyncio.gather(*task))for res in results:print(res.title)

?

總結

以上是生活随笔為你收集整理的pyppeteer:比selenium更高效的爬虫利器的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。