Python使用BrowserMob Proxy和Selenium获取浏览器请求和响应

BrowserMob Proxy介绍

BrowserMobProxy会提供一个ProxyServer用于做转发代理拦截，这个server可以是standalone部署支持远程，也可以embed进代码中。由于BrowserMob是Java开发的，因此JVM的可以支持真正的embedded，python等非JVM系的只能配置其执行路径通过子进程的方式来伪装embedded.

安装

pip install browsermob-proxy

下载 browsermob-proxy 的二进制文件

下载地址：https://github.com/lightbody/browsermob-proxy/releases

调试

from browsermobproxy import Server


## 启动代理, 修改下载的文件路径
server = Server(r'**\browsermob-proxy-2.1.4\bin\browser-mob-proxy.bat') server.start()proxy = server.create_proxy()print('proxy', proxy.proxy)

使用

from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from browsermobproxy import Server
import time
import requests
import json


class BaseFramework(object):

    def __init__(self):
        # browsermob-proxy.bat的文件路径
        self.server = Server(r'*\browsermob-proxy-2.1.4\bin\browsermob-proxy.bat')
        self.server.start()
        self.proxy = self.server.create_proxy()
        self.query_url = 'http://www.baidu.com'
        chrome_options = Options()
        chrome_options.add_argument('--ignore-certificate-errors')
        chrome_options.add_argument('--proxy-server={0}'.format(self.proxy.proxy))
        # chrome_options.add_argument('--headless')  # 无头模式
	# chromedriver.exe的文件路径
        self.browser = webdriver.Chrome(executable_path="*\chromedriver.exe", options=chrome_options)

    def process_request(self, request, response):
        pass

    def process_response(self, response, request):
        pass

    def run(self, func, *args):
        self.proxy.new_har(options={
            'captureContent': True,
            'captureHeaders': True
        })
        func(*args)
        result = self.proxy.har
        for entry in result['log']['entries']:
            request = entry['request']
            response = entry['response']
            self.process_request(request, response)
            self.process_response(response, request)
        self.proxy.close()
        self.browser.close()

    def check_params(self, params: list) -> str:
        pass

    def check_mongo(self, track_data: dict) -> str:
        pass


class Framework(BaseFramework):

    def load(self, url):
        self.browser.get(url)
        time.sleep(3)

    def process_request(self, request, response):
        pass

    def process_response(self, response, request):
        # print(request['url'])
	# 查找需要数据的URL后即可进行解析
	if '**' in request['url']:
		pass


if __name__ == '__main__':
    Framework = Framework()
    url = "**"
    Framework.run(Framework.load, url)