Python cdp ( Chrome DevTools Protocol ) 爬虫
可以说,利用 Develop Tools 实现的功能我们都可以通过 Chrome DevTools Protocol 实现,Chrome 自己也内置了一个官方的实现,用 Chrome 直接访问页面信息的 devtoolsFrontendUrl 即可看到,和按 F12 调用出来的 Develop Tools 基本一模一样。如果你想看看页面在远程服务器的 Chrome 里渲染的结果,在开发者工具里切换
devtools-protocol:https://github.com/ChromeDevTools/devtools-protocol
Chrome DevTools Protocol:https://chromedevtools.github.io/devtools-protocol/
实现 CDP 协议的库:https://github.com/ChromeDevTools/awesome-chrome-devtools
cdp chrome:https://github.com/search?q=cdp+chrome&type=repositories&p=2
1、关于 Chrome Debugging Protocol
Chrome 提供了 websocket 调试接口用于对当前 Tab 内页面的 DOM、网络、性能、存储 等等进行调试,我们常用的开发者工具就是基于此接口,这个接口也支持远程调用,在启动参数中加上 --remote-debugging-port=9222 即可。
远程调试接口是一个 WebSocket 的接口,Chrome 提供的开发者工具是一种客户端,自己写代码调用也是一种客户端。DevTools 可以 attach 到一个远程运行的 Chrome 实例,来进行 debug,
- 启动一个 host Chrome 实例:chrome.exe --remote-debugging-port=9222
- 指定一个用户的 profile 启动一个 client 客户端 Chrome 实例:chrome.exe --user-data-dir=<some directory>
在客户端中输入 http://localhost:9222,你会看到当前 client 端的 DevTools 就像内嵌在电脑中Chrome 的 devtools 一样,然后你就可以通过 client 操作他了。
- 当用 client 连接到 9222 的时候,DevTools 前端 会被 host Chrome 实例 serve 为一个远程服务端的 web application,他会通过 HTTP 协议 fetch HTML,JavaScript,CSS,一旦加载,DevTools 会和 host 建立一个 ws 链接,并开始交换 JSON 信息
也可以用自己的实现来替代 DevTools 前端,除了 localhost:9222,
远程调试还提供了一个JSON接口,用于管理浏览器的 Tab 页面。( localhost:9222/json 来获取到 ws 通信的 JSONobject,并修改使用它们, HTTP Endpoints 章节中有详细内容。 )
2、实战 Chrome Headless 数据抓取
实战 Chrome Headless 数据抓取(上):https://blog.csdn.net/chixulu6723/article/details/100730003
启动 Chrome
启动之前,需要关闭所有 Chrome 窗口
Linux 中的 screen 命令使用:https://blog.csdn.net/han0373/article/details/81352663
如果在远程服务器上建议在 screen 里运行,一个小工具防止网络突然中断:$ screen -S chrome
然后会打开一个新的 shell,可以用 Ctrl + A + D 切出来,或者断开SSH 直接切出来。再进去只需要执行:$ screen -r chrome
然后在 screen 里面的 shell 执行( 本机 Windows 调试把 google-chrome-unstable 换成chrome.exe ):$ google-chrome-unstable --headless --remote-debugging-port=9222 --user-data-dir='/home/luke/chrome-data/baidu' --user-agent='Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/60.0.3080.5 Safari/537.36'
解释下参数:
- --headless:无头模式,就是无界面模式运行
- --remote-debugging-port:开启远程调试,端口9222和我们之前转发出来的端口一致
- --user-data-dir:设置独立的文件保存目录,建议一个网站一个目录
- --user-agent:伪装浏览器,默认User-Agent里的浏览器叫HeadlessChrome,很容易被发现
连接调试端口
启动之前,需要关闭所有 Chrome 窗口,然后在启动,
远程调试接口是一个 WebSocket 的接口,Chrome 提供的开发者工具是一种客户端,我们自己写代码调用也是一种客户端。这里先用开发者工具测试,后面会写代码来实现。 在本地浏览器开:http://服务器IP:9222/json ,本机测试的话就是 http://127.0.0.1:9222/json 。我已经用 SSH 把服务器的 9222 转发到本机的 9222 了,这是效果:
这里列出了当前远程浏览器内打开的Tab,每个页面一个UUID用以识别。已知接口:
- http://127.0.0.1:9222/json :查看已经打开的Tab列表
- http://127.0.0.1:9222/json/version : 查看浏览器版本信息
- http://127.0.0.1:9222/json/new?http://www.baidu.com : 新开Tab打开指定地址
- http://127.0.0.1:9222/json/close/ac5a6adb-bb53-44f1-a9e6-2354bd724924 : 关闭指定Tab
- http://127.0.0.1:9222/json/activate/69301801-d503-42a3-9335-3e448a780857 : 切换到目标Tab
抓取百度
远程打开,新开一个 Tab 打开百度首页,然后刷新 http://127.0.0.1:9222/json ,可以看到百度已经打开了:
注意到,有一个devtoolsFrontendUrl,那就是开发者工具的前端地址,就是一个html应用,url里面传过去WebSocket调试地址。打开这个地址就可以看到熟悉的开发者工具了!注意:这个窗口调试的是远程chrome上的页面。
如果你想看看页面在远程服务器的 Chrome 里渲染的结果,在开发者工具里切换到 Performance,勾选 Screenshots,点刷新图标,重新加载完成就可以看到逐帧加载的截图。
远程操作 DOM
思路:我们在 Elements 里面找到输入框的 ID,使用 JQuery 操作。百度首页已经有 JQuery 了,其他网站我们可以在 Console 里执行 JS,加载一个。 我们切换到 Console 里直接用 JS 操作DOM,执行:
$("input[name='wd']").val('测试');
$("form.fm").submit();
相当于在百度输入框里输入了测试并点击了“百度一下”按钮。
现在再打开 http://127.0.0.1:9222/json ,可以看到原来的页面标题已经变成了 “测试_百度搜索”,也就说明成功完成了搜索。
获取搜索结果
依然是在 Elements 里面找到结果列表的 ID,然后用 JS 获取内容。 获得结果数量:
console.log($("#container").find(".nums").text());
获得所有结果标题:
$("#container").find(".c-container").each(function(){console.log($(this).find("h3.t").text())});
常用指令
Chrome DevTools Protocol 的指令分为三十多个大类,每类又有若干个指令,这里不能一一介绍,只选择几个简单而常用的指令介绍一下:
- 跳转到指定页面:https://chromedevtools.github.io/devtools-protocol/tot/Page/
- 执行JS函数:https://chromedevtools.github.io/devtools-protocol/tot/Runtime/
- 获取资源树:https://chromedevtools.github.io/devtools-protocol/tot/Page/
- 获取资源:https://chromedevtools.github.io/devtools-protocol/tot/Page/
其中 Page.navigate 是必备指令,用于跳转页面。而 Runtime.evaluate 的效果等同于在 Develop Tools 的 Console 控制台执行指令,基本可以执行任何js指令,模拟输入,输出渲染后的 html 用它都可以轻松搞定,可以说是大杀器了。
获取资源树和获取资源指令则用于获取浏览器当前原始请求的数据,可以用它来构建Develop Tools的Source树。
可以说,利用 Develop Tools 实现的功能我们都可以通过 Chrome DevTools Protocol 实现,Chrome 自己也内置了一个官方的实现,用 Chrome 直接访问页面信息的 devtoolsFrontendUrl 即可看到,和按 F12 调用出来的 Develop Tools 基本一模一样。
3、chrome headless 爬虫抓取 websoket 数据
From:https://brucedone.com/archives/1201
要分析的网站:https://datacenter.jin10.com/price
Python 连接 websocket 代码:
# coding:utf-8
from websocket import create_connection
from websocket import ABNF
api = "wss://sshibikfdn.jin10.com:9084/socket.io/?EIO=3&transport=websocket&sid=VsJvZikGdc8spBaPAAMO"
headers = {
'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
'Cookie': 'UM_distinctid=16614315bf179b-0354a40c6714ff-34677908-232800-16614315bf2acb; XSRF-TOKEN=eyJpdiI6IkNZRU9uSmM1ZnY2M0VqNUttK1pxRGc9PSIsInZhbHVlIjoiRlJpNlRuekIxTDJZeVd3bHpvXC9OUEZGamw4VndZNEdXTEVsRjRMaFFyOEIxUHRtNDdTc1JaQ042eG4xdjlFeWJjWGlkcWFaeWl6NTRVUUlQMThaZmJ3PT0iLCJtYWMiOiJkYWU1MzQ2NjEyM2U3OTk0MzY5NWNjZTdhZmNlZjE0YTViMjc2YzBiYWM4YjhiMjNhZmRjMzU3YzliNDg3ZGIzIn0%3D; laravel_session=eyJpdiI6IjBRS3h0Y29XcGRBRlFIc0xIeWFiZGc9PSIsInZhbHVlIjoibVRLblpNTDJJa1JIN1ZJc0s5c2xrSkYzckNadDB6aGp0REd5SVJQTlkxNVAzajhvdXY5ZElSQ3VTcGVicjNiSXZ3NE9pZDZOdHJUM1d6WG1KQjZXNkE9PSIsIm1hYyI6Ijg3MWVkZDVlMDFjZDM2NDRjZmI2ZDhkNDJmZGI5MjNhMzk3MTViNmI1YTNmMDRmYWJjNzQ4ZGU2YWZhNzNhNzUifQ%3D%3D; io=VsJvZikGdc8spBaPAAMO',
'Host': 'sshibikfdn.jin10.com:9084',
'Connection': 'Upgrade',
'Origin': 'https://datacenter.jin10.com',
'Pragma': 'no-cache',
'Sec-WebSocket-Extensions': 'permessage-deflate; client_max_window_bits',
'Sec-WebSocket-Key': 'g4UA3smEJ0eGufMkyz7AOw==',
'Sec-WebSocket-Version': '13',
'Upgrade': 'websocket',
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_13_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/68.0.3440.106 Safari/537.36',
}
def get_web_socket():
start_message = "2probe"
ws = create_connection(
api,
header=headers,
cookie=headers['Cookie'],
#origin=headers['Origin'],
#host=headers['Host']
)
frame = ABNF.create_frame("2probe", ABNF.OPCODE_TEXT)
ws.send_frame(frame)
data = ws.recv_frame()
print(data)
if __name__ == '__main__':
get_web_socket()
使用 websocket 和 chrome header less 进行交互
import json
import time
import requests
import websocket
request_id = 0
target_url = 'https://datacenter.jin10.com/price'
def get_websocket_connection():
r = requests.get('http://10.10.2.42:9222/json') #这是开启docker chrome headless的机器地址
if r.status_code != 200:
raise ValueError("can not get the api ,please check if docker is ready")
conn_api = r.json()[0].get('webSocketDebuggerUrl')
return websocket.create_connection(conn_api)
def run_command(conn, method, **kwargs):
global request_id
request_id += 1
command = {'method': method,
'id': request_id,
'params': kwargs}
conn.send(json.dumps(command))
#while True:
msg = json.loads(conn.recv())
if msg.get('id') == request_id:
return msg
def get_element():
conn = get_websocket_connection()
msg = run_command(conn, 'Page.navigate', url=target_url)
time.sleep(5)
js = "var p = document.querySelector('.jin-pricewall_list-item_b').innerText ; p ;"
for _ in range(20):
time.sleep(1)
msg = run_command(conn, 'Runtime.evaluate', expression=js)
print(msg.get('result')['result']['value'])
if __name__ == '__main__':
get_element()
整体逻辑非常简单,打开指定页面,等待页面数据刷新,然后直接偷懒拿数据渲染之后的页面值,运行效果如下:
这里使用 chrome-headless 的相关渲染环境来解决了抓取数据的问题,并且使用 websocket api 来进一步操作,其实 google 官方有 sdk 进行操作,https://github.com/GoogleChrome/puppeteer ,渲染的终究不是高效的做法,但是对于这种单页面目的性很强的数据,可以尝试渲染大法
Pyppeteer 防止检测
4、chrome devtools protocol
chrome devtools protocol 允许第三方对基于 chrome 的 web 应用程序进行调试、分析等,它基于 WebSocket,利用 WebSocke t建立连接 DevTools 和浏览器内核的快速数据通道。一句话,有了这个协议就可以自己开发工具获取 chrome 的数据
协议详细内容看这里 chrome devtools protocol:Chrome DevTools Protocol
目前已经有很多大神针对这个协议封装出不同语言(nodejs,python,java...)的库。
awesome-chrome-devtools:https://github.com/ChromeDevTools/awesome-chrome-devtools
Protocol Driver Libraries
- JavaScript/Node.js: chrome-remote-interface - The most-used JavaScript API for the protocol
- TypeScript/Node.js: chrome-debugging-client
- Java: chrome-devtools-java-client
- Java: karate - Web-service testing framework with a Java API to automate Chrome
- Java: jvppeteer - Headless Chrome For Java
- Python: PyCDP - Pure-Python, sans-IO wrappers. See also the Trio CDP driver
- Python: chromewhip - drop-in replacement for the
splash
service - Python: pychrome - low level CDP transport handler
- Python: ChromeController - high-level browser mgmt
- Go: chromedp - High-level actions and tasks for driving browsers
- Go: cdp
- Go: gcd
- Go: godet
- Go: Rod
- C#/dotnet: chrome-dev-tools - Protocol wrapper generator that can be customized by editing handlebars templates. Includes .Net Core template.
- Ruby: Cuprite - Capybara driver
- Ruby: ChromeRemote
- Kotlin: chrome-reactive-kotlin - reactive (rxjava 2.x), low-level client library in Kotlin
- Kotlin: chrome-devtools-kotlin - A coroutine-based client library, providing low-level CDP primitives and high-level extensions.
- Clojure: clj-chrome-devtools - The CDP wrapper API is autogenerated and will be updated when CDP protocol changes.
- PHP: chrome-devtools-protocol - A PHP client library for the protocol.
PyChromeDevTools 爬虫 示例
PyChromeDevTools:https://github.com/marty90/PyChromeDevTools
PyChromeDevTools 是一个 Python 模块,允许在 Python 脚本中使用 Chrome DevTools 协议与 Google Chrome 交互。
页面加载时间
import PyChromeDevTools
import time
chrome = PyChromeDevTools.ChromeInterface()
chrome.Network.enable()
chrome.Page.enable()
start_time=time.time()
chrome.Page.navigate(url="http://www.google.com/")
chrome.wait_event("Page.loadEventFired", timeout=60)
end_time=time.time()
print ("Page Loading Time:", end_time-start_time)
打印所有安装的 cookies
import PyChromeDevTools
import time
chrome = PyChromeDevTools.ChromeInterface()
chrome.Network.enable()
chrome.Page.enable()
chrome.Page.navigate(url="http://www.nytimes.com/")
chrome.wait_event("Page.frameStoppedLoading", timeout=60)
#Wait last objects to load
time.sleep(5)
cookies,messages = chrome.Network.getCookies()
for cookie in cookies["result"]["cookies"]:
print ("Cookie:")
print ("\tDomain:", cookie["domain"])
print ("\tKey:", cookie["name"])
print ("\tValue:", cookie["value"])
print ("\n")
打印页面的所有对象 URL
import PyChromeDevTools
chrome = PyChromeDevTools.ChromeInterface()
chrome.Network.enable()
chrome.Page.enable()
chrome.Page.navigate(url="http://www.facebook.com")
event,messages=chrome.wait_event("Page.frameStoppedLoading", timeout=60)
for m in messages:
if "method" in m and m["method"] == "Network.responseReceived":
try:
url=m["params"]["response"]["url"]
print (url)
except:
pass
pychrome 爬虫 示例
Pychrome:能跟 chrome 开发者工具交流的 Python 包,查看 github 代码中 examples 文件夹,查看更多例子。pychrome github地址,使用方法很简单,直接看 github上它的 Demo
这个库依赖 websocket-client
python - pychrome 页面抓取测试:https://blog.csdn.net/max229max/article/details/91972429
使用 Chrome-headless 抓取页面内容,使用 python 的 pychrome 包。
要先开启浏览器,然后通过pychrome调用chrome dev protocol
# -*- coding: utf-8 -*-
# @Author :
# @Date : 2021/8/19
# @File : test.py
# @description : XXX
import pychrome
from concurrent.futures import ThreadPoolExecutor
def request_will_be_sent(**kwargs):
print("loading: %s" % kwargs.get('request').get('url'))
def start_chrome(chrome_url='http://127.0.0.1:9222'):
browser = pychrome.Browser(url=chrome_url)
tab = browser.new_tab()
tab.set_listener("Network.requestWillBeSent", request_will_be_sent)
tab.start()
tab.call_method("Network.enable")
tab.call_method("Page.navigate", url="https://detail.tmall.com/item.htm?id=612274990981", _timeout=10)
tab.wait(20)
tab.stop()
browser.close_tab(tab)
if __name__ == '__main__':
"""
chrome.exe --remote-debugging-port=9222 --user-data-dir=c:/user_data_dir_9222
chrome.exe --remote-debugging-port=9223 --user-data-dir=c:/user_data_dir_9223
"""
tem_list = ('http://127.0.0.1:9222', 'http://127.0.0.1:9223')
with ThreadPoolExecutor() as tp_executor:
tp_executor.submit(start_chrome, 'http://127.0.0.1:9222')
tp_executor.submit(start_chrome, 'http://127.0.0.1:9223')
# start_chrome()
pass
Web 性能自动化
chrome devtools protocol --- Web 性能自动化实践介绍 :https://testerhome.com/topics/15817
在测试 Web 页面加载时间时,可能会是这样的:
- 打开 chrome 浏览器。
- 按 F12 打开开发者工具。
- 在浏览器上打开要测试的页面
- 查看开发者工具中 Network 面板的页面性能数据并记录
- 或者 开发者 Console 面板运行 performance.timing 和 performance.getEntries() 收集数据
performance 相关信息看这里 https://www.w3.org/TR/navigation-timing-2/#the-performancetiming-interface
几十上百个页面,每个版本都这样来,估计疯了,所以就想怎么把它做成自动化呢?
获取 performance api 数据
这里使用 Runtime Domain 中运行 JavaScript 脚本的 API Runtime.evaluate
# 开始前先启动chrome,启动chrome必须带上参数`--remote-debugging-port=9222`开启远程调试否则无法与chrome交互
browser = pychrome.Browser('http://127.0.0.1:%d' % 9222)
tab = browser.new_tab()
tab.start()
tab.Runtime.enable()
tab.Page.navigate(url={你的页面地址})
# 设置等待页面加载完成的时间
tab.wait(10)
# 运行js脚本
timing_remote_object = tab.Runtime.evaluate(
expression='performance.timing'
)
# 获取performance.timing结果数据
timing_properties = tab.Runtime.getProperties(
objectId=timing_remote_object.get('result').get('objectId')
)
timing = {}
for item in timing_properties.get('result'):
if item.get('value', {}).get('type') == 'number':
timing[item.get('name')] = item.get('value').get('value')
# 获取performance.getEntries()数据
entries_remote_object = tab.Runtime.evaluate(
expression='performance.getEntries()'
)
entries_properties = tab.Runtime.getProperties(
objectId=entries_remote_object.get('result').get('objectId')
)
entries_values = []
for item in entries_properties.get('result'):
if item.get('name').isdigit():
url_timing_properties = tab.Runtime.getProperties(
objectId=item.get('value').get('objectId')
)
entries_value = {}
for son_item in url_timing_properties.get('result'):
if (son_item.get('value', {}).get('type') == 'number'or
son_item.get('value', {}).get('type') == 'string'):
entries_value[son_item.get('name')] = son_item.get('value').get('value')
entries_values.append(entries_value)
获取 Network 数据
实际上 performance.getEntries() 不会记录 404 的请求信息,另外当前页面通过 js 触发新 html 页面请求时它只会记录第一个页面的请求,在这些情况下就需要通过 Network Domain 的 API 来收集所有请求信息,先介绍用到的 API :
- Network.requestWillBeSent 每个 http 请求发送前回调
- Network.responseReceived 首次接送到 http 响应时回调
- Network.loadingFinished 请求加载完成时回调
- Network.loadingFailed 请求加载失败时回调
# 封装上面4个事件对应的回调方法
class NetworkAPIImplemention(object):
def __init__(self):
self.request_dict = {}
# 首个请求开始时间
self.start = None
def request_will_be_sent(self, **kwargs):
if self.start is None:
self.start = time.time()
dict_http = {
'url':kwargs.get('request').get('url'),
'start':kwargs.get('timestamp')
}
self.request_dict[kwargs.get('requestId')]=dict_http
#print "loading:%s" % kwargs.get('request').get('url')
def loading_finished(self, **kwargs):
# 服务器返回code 例如404也是finished
self.request_dict[kwargs.get('requestId')]['end'] = kwargs.get('timestamp')
self.request_dict[kwargs.get('requestId')]['size'] = kwargs.get('encodedDataLength')
def response_received(self, **kwargs):
self.request_dict[kwargs.get('requestId')]['type'] = kwargs.get('type')
self.request_dict[kwargs.get('requestId')]['response'] = kwargs.get('response')
def loading_failed(self, **kwargs):
self.request_dict[kwargs.get('requestId')]['end'] = kwargs.get('timestamp')
self.request_dict[kwargs.get('requestId')]['error_text'] = kwargs.get('errorText')
network_api = NetworkAPIImplemention()
browser = pychrome.Browser('http://127.0.0.1:%d' % 9222)
tab = browser.new_tab()
# 绑定回调函数
tab.Network.requestWillBeSent = network_api.request_will_be_sent
tab.Network.responseReceived = network_api.response_received
tab.Network.loadingFinished = network_api.loading_finished
tab.Network.loadingFailed = network_api.loading_failed
tab.start()
tab.Network.enable()
tab.Runtime.enable()
# 是否禁用缓存
if disable_cache:
tab.Network.setCacheDisabled(cacheDisabled=True)
tab.Page.navigate(url={你的页面地址})
tab.wait(10)
tab.stop()
self.browser.close_tab(tab)
# 获取的所有url详细信息
print network_api.request_dict
监听 页面 事件
有时候特别是一些复杂的页面,页面依赖 js 和 后端资源数据,并不是通常意义上页 loadEventEnd 事件触发完就表示页面加载完成,这种情况可能需要依赖开发打点。
这里以开发设计了一个 Loaded 事件为例:
# 具体事件注册方式和注册时机询问开发,所谓注册时机即要求在js对象生成后注册,我们项目中page是在一个js文件中声明的,需要等这个js文件请求完成后再注册
# 这边使用Promise方式,这种方式awaitPromise参数必须是True
js = """
new Promise((resolve, reject) => {
page.getController().getPageEvent().addEventListener("Loaded",
function(){
resolve(new Date().getTime());
});
});
"""
custom_result = tab.Runtime.evaluate(
expression=js,
awaitPromise=True,
timeout=timeout * 1000
)
print custom_result.get('result').get('value')
有个坑
peformance.now()
获取与 chrome 开发者工具协议一样类型的时间时,这个时间不准确,只好用new Date().getTime()
写在最后
一开始是使用 nodejs 的 chrome-remote-interface,但是发现 Page.loadEventFired 回调后不会再记录请求,事实上有些页面仍然有请求没有完成,不懂是不是我使用姿势不对。。。
附赠 W3C 的一幅图
![Logo](https://devpress.csdnimg.cn/79de2bf0b7994defa4242ef90d5513fa.jpg)
开放原子开发者工作坊旨在鼓励更多人参与开源活动,与志同道合的开发者们相互交流开发经验、分享开发心得、获取前沿技术趋势。工作坊有多种形式的开发者活动,如meetup、训练营等,主打技术交流,干货满满,真诚地邀请各位开发者共同参与!
更多推荐
所有评论(0)