Skip to content

Instantly share code, notes, and snippets.

@rkdgusrnrlrl
Last active November 28, 2019 05:46
Show Gist options
  • Save rkdgusrnrlrl/315bd50aabe18d010eb0fc0d3111f285 to your computer and use it in GitHub Desktop.
Save rkdgusrnrlrl/315bd50aabe18d010eb0fc0d3111f285 to your computer and use it in GitHub Desktop.
add how get ws url
import websocket
# pip install websocket-client
import json
try:
import thread
except ImportError:
import _thread as thread
import time
import requests
# pip install requests
mm = {
'id': 0,
'method': 'Log.enable'
}
mm2 = {
'id': 1,
'method': 'Network.enable'
}
method_id = 4
requset_dict = dict()
response_dict = dict()
request_id_dict = dict()
response_body_dict = dict()
# http://localhost:9222/json -> webSocketDebuggerUrl
websocket_url = 'ws://localhost:9222/devtools/page/3F16DFD0A8C59E8B860643E970CB2C1B'
def on_message(ws, message):
ee = json.loads(message)
# save log(need fix)
if 'method' in ee and ee['method'] == 'Log.entryAdded':
eee = ee['params']['entry']
print('[%s] %s' % (eee['level'], eee['text']))
# save response
elif 'method' in ee and ee['method'] == 'Network.responseReceived':
id = ee['params']['requestId']
response_dict[id] = ee['params']['response']
get_response_body(ws, id)
# save request
elif 'method' in ee and ee['method'] == 'Network.requestWillBeSent':
id = ee['params']['requestId']
ee['params']['request']['id'] = id
requset_dict[id] = ee['params']['request']
# integrate request, response, body
elif 'id' in ee and 'body' in ee['result']:
m_id = ee['id']
request_id = request_id_dict[m_id]
body = ee['result']['body']
response_dict[request_id]['body'] = body
print('--------------------request--------------------')
print(requset_dict[request_id])
print('--------------------response--------------------')
print(response_dict[request_id])
def get_response_body(ws, request_id):
global method_id
method_id += 1
get_response_body_method = {
'id' : method_id,
'method': 'Network.getResponseBody',
'params': {
'requestId' : request_id
}
}
ws.send(json.dumps(get_response_body_method))
request_id_dict[method_id] = request_id
def on_error(ws, error):
print(error)
def on_close(ws):
print("### closed ###")
def on_open(ws):
def run(*args):
ws.send(json.dumps(mm))
time.sleep(1)
ws.send(json.dumps(mm2))
time.sleep(1)
thread.start_new_thread(run, ())
if __name__ == "__main__":
r = requests.get('http://localhost:9222/json')
dd = r.json()
websocket_url = dd[0]['webSocketDebuggerUrl']
websocket.enableTrace(True)
ws = websocket.WebSocketApp(websocket_url,
on_message = on_message,
on_error = on_error,
on_close = on_close)
ws.on_open = on_open
ws.run_forever()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment