piaoquan_crawler/dev/dev_script/mitmproxy_test.py

# -*- coding: utf-8 -*-
# @Author: wangkun
# @Time: 2023/7/24
import os
import time

from mitmproxy import http, proxy, options, proxyconfig
from mitmproxy.proxy.config import ProxyConfig
from mitmproxy.tools.dump import DumpMaster
from selenium import webdriver
from selenium.webdriver import DesiredCapabilities
from selenium.webdriver.chrome.service import Service


class ProxyData:
    requests_data = []
    response_data = []

    @classmethod
    def start_proxy(cls):
        # 创建代理配置选项
        opts = options.Options(listen_host='0.0.0.0', listen_port=8888)

        # 创建代理配置
        config = ProxyConfig(opts)

        # 创建DumpMaster实例
        master = DumpMaster(opts)
        master.server = config

        # 启动代理
        print("Proxy started")
        master.run()

    @classmethod
    def intercept_request(cls, flow: http.HTTPFlow):
        # 拦截请求
        request_data = {
            'url': flow.request.url,
            'method': flow.request.method,
            'headers': dict(flow.request.headers),
            'content': flow.request.content.decode('utf-8')
        }
        cls.requests_data.append(request_data)

    @classmethod
    def intercept_response(cls, flow: http.HTTPFlow):
        # 拦截响应
        response_data = {
            'url': flow.request.url,
            'status_code': flow.response.status_code,
            'headers': dict(flow.response.headers),
            'content': flow.response.content.decode('utf-8')
        }
        cls.response_data.append(response_data)

    @classmethod
    def start_selenium(cls):
        quit_cmd = "ps aux | grep Chrome | grep -v grep | awk '{print $2}' | xargs kill -9"
        os.system(quit_cmd)
        time.sleep(1)
        # 启动 Chrome，指定端口号:8888
        cmd = 'open -a "Google Chrome" --args --remote-debugging-port=8888'
        os.system(cmd)
        # 打印请求配置
        ca = DesiredCapabilities.CHROME
        ca["goog:loggingPrefs"] = {"performance": "ALL"}
        # 配置 chromedriver
        chromedriver = "/Users/wangkun/Downloads/chromedriver/chromedriver_v114/chromedriver"
        # 初始化浏览器
        browser = webdriver.ChromeOptions()
        browser.add_experimental_option("debuggerAddress", "127.0.0.1:8888")
        # driver初始化
        driver = webdriver.Chrome(desired_capabilities=ca, options=browser, service=Service(chromedriver))
        driver.implicitly_wait(10)
        print("打开抖音推荐页")
        driver.get(f"https://www.douyin.com/")


if __name__ == "__main__":
    ProxyData.start_proxy()
    ProxyData.start_selenium()
    print("requests_data:", ProxyData.requests_data)
    print("response_data:", ProxyData.response_data)

    # 分析包含链接 www.douyin.com 的响应数据
    for response in ProxyData.response_data:
        if "www.douyin.com" in response['url']:
            print("Douyin response:", response)