【测试】selenium拦截Ajax(XHR)等异步请求数据

这篇具有很好参考价值的文章主要介绍了【测试】selenium拦截Ajax(XHR)等异步请求数据。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

1.说明

在爬取某个网站的时候遇到加密参数,由于js代码经过混淆编译不好破解,所以采用selenium的方式获取参数,但是我们获取selenium的数据基本上都是基于页面的,对于网站发起的异步请求,我们可以从日志中提取

2.设置driver参数

我们首先要通过Option对象(比如说ChromeOptions)设置监控浏览器日志,旧版本的Selenium是通过DesiredCapabilities设置的,下面是新版本的写法

from selenium.webdriver import Chrome, ChromeOptions
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.common.by import By
from selenium.webdriver.support import expected_conditions
from selenium.webdriver.support.wait import WebDriverWait

options = ChromeOptions()
options.add_argument("--no-sandbox")
options.add_argument("--allow-running-insecure-content")
options.add_argument("--ignore-certificate-errors")
options.add_argument("--disable-single-click-autofill")
options.add_argument("--disable-autofill-keyboard-accessory-view[8]")
options.add_argument("--disable-full-form-autofill-ios")
options.add_experimental_option('perfLoggingPrefs', {
    'enableNetwork': True,
    'enablePage': False,
})
options.set_capability("goog:loggingPrefs", {
    'browser': 'ALL',
    'performance': 'ALL',
})
options.set_capability("goog:perfLoggingPrefs", {
    'enableNetwork': True,
    'enablePage': False,
    'enableTimeline': False
})
3.请求网页

现在实例化一个driver,发起一个网页请求,我这里使用WebDriverWait显式等待的方式等待某个元素出现,你也可以隐式等待或者直接sleep,如果你不等待,异步请求还没加载完就开始获取,你可能会拿不到想要的数据

service = Service(executable_path=executable_path)
driver = Chrome(service=service, options=options)
driver.execute_cdp_cmd("Page.addScriptToEvaluateOnNewDocument",
                       {"source": """Object.defineProperty(navigator, 'webdriver', {get: () => undefined})"""})
driver.get(page_url)
wait = WebDriverWait(driver, 15, 0.5)
try:
    wait.until(expected_conditions.presence_of_element_located((By.CLASS_NAME, "item ")))
except Exception as e:
    print("WebDriverWait.until timeout error: {}".format(e))
html = driver.execute_script("return document.documentElement.outerHTML")
4.处理日志

访问一下driver的log_types属性可以获取到所有日志类型,遍历它,通过get_log()方法获取对应的日志,之后再过滤出自己想要的日志就行。

比如说,我这里是过滤出所有Network.requestWillBeSent的日志,即发送异步请求的数据,因为我需要该请求的请求头,如果是响应类型的日志(Network.responseReceived),它只包含响应头。具体支持的类型可以参考谷歌devtools的文档

如果需要过滤出Ajax(XHR)请求,可以根据日志的params里的type进行判断,也可以通过它判断

sign_dict = dict()  # 用来存储自己想要的数据
for log_type in driver.log_types:
    perf_list = driver.get_log(log_type)
    for row_log in perf_list:
        try:
            log_json = json.loads(row_log['message'])
            message_log = log_json['message']
        except Exception as e:
            print(e)
            continue
        if message_log.get('method') != 'Network.requestWillBeSent':
            continue
        if message_log.get("params", {}).get("type", "").upper() != "XHR":
            continue
        headers = message_log['params'].get('request', {}).get('headers')
        if not headers:
            continue
        x_sign = headers.get('X-Sign')
        if not x_sign:
            continue
        x_app_id = headers.get('X-AppID')
        x_ts = headers.get('X-Ts')
        print("success:", x_sign, x_app_id, x_ts)
        req_url = message_log['params'].get('request', {}).get('url')
        key = os.path.split(req_url.split("?")[0])[1]
        sign_dict[key] = {"X-AppID": x_app_id, "X-Sign": x_sign, "X-Ts": x_ts}

注意,如果你想要响应体,Network.responseReceived类型的日志的response字段是没有响应体的,你需要通过params字段里的requestId获取,参考代码如下

res_body_dict = dict()
for log_type in driver.log_types:
    perf_list = driver.get_log(log_type)
    for row_log in perf_list:
        try:
            log_json = json.loads(row_log['message'])
            message_log = log_json['message']
        except Exception as e:
            print(e)
            continue
        if message_log.get('method') != 'Network.responseReceived':
            continue
        if message_log.get("params", {}).get("type", "").upper() != "XHR":
            continue
        request_id = message_log['params'].get("requestId")
        if not request_id:
            continue
        req_url = message_log['params'].get('response', {}).get('url')
        key = os.path.split(req_url.split("?")[0])[1]
        content = driver.execute_cdp_cmd('Network.getResponseBody', {'requestId': request_id})
        body = None
        try:
            body = json.loads(content["body"])
        except Exception as e:
            print("get_unisat_data_by_selenium() json loads error: {}, content:{}".format(e, content))
        res_body_dict[key] = body
5.完整代码

上面的完整参考代码如下

import json
import os.path

from selenium.webdriver import Chrome, ChromeOptions
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.common.by import By
from selenium.webdriver.support import expected_conditions
from selenium.webdriver.support.wait import WebDriverWait


def get_selenium_driver(executable_path=r"E:\webdriver\chromedriver.exe"):
    options = ChromeOptions()
    options.add_argument("--no-sandbox")
    options.add_argument("--allow-running-insecure-content")
    options.add_argument("--ignore-certificate-errors")
    options.add_argument("--disable-single-click-autofill")
    options.add_argument("--disable-autofill-keyboard-accessory-view[8]")
    options.add_argument("--disable-full-form-autofill-ios")
    options.add_experimental_option('perfLoggingPrefs', {
        'enableNetwork': True,
        'enablePage': False,
    })
    options.set_capability("goog:loggingPrefs", {
        'browser': 'ALL',
        'performance': 'ALL',
    })
    options.set_capability("goog:perfLoggingPrefs", {
        'enableNetwork': True,
        'enablePage': False,
        'enableTimeline': False
    })
    service = Service(executable_path=executable_path)
    driver = Chrome(service=service, options=options)
    driver.execute_cdp_cmd("Page.addScriptToEvaluateOnNewDocument",
                           {"source": """Object.defineProperty(navigator, 'webdriver', {get: () => undefined})"""})

    return driver


def get_sign_by_selenium(page_url):
    driver = get_selenium_driver()
    driver.execute_cdp_cmd("Page.addScriptToEvaluateOnNewDocument",
                           {"source": """Object.defineProperty(navigator, 'webdriver', {get: () => undefined})"""})
    driver.get(page_url)
    wait = WebDriverWait(driver, 15, 0.5)
    try:
        wait.until(expected_conditions.presence_of_element_located((By.CLASS_NAME, "item ")))
    except Exception as e:
        print("WebDriverWait.until timeout error: {}".format(e))
    # html = driver.execute_script("return document.documentElement.outerHTML")
    # with open(r"C:\Users\admin\Desktop\test\test.html", "w") as f:
    #     f.write(html)
    # time.sleep(10)
    sign_dict = dict()
    for log_type in driver.log_types:
        perf_list = driver.get_log(log_type)
        for row_log in perf_list:
            try:
                log_json = json.loads(row_log['message'])
                message_log = log_json['message']
            except Exception as e:
                print(e)
                continue
            if message_log.get('method') != 'Network.requestWillBeSent':
                continue
            if message_log.get("params", {}).get("type", "").upper() != "XHR":
                continue
            headers = message_log['params'].get('request', {}).get('headers')
            if not headers:
                continue
            x_sign = headers.get('X-Sign')
            if not x_sign:
                continue
            x_app_id = headers.get('X-AppID')
            x_ts = headers.get('X-Ts')
            print("success:", x_sign, x_app_id, x_ts)
            req_url = message_log['params'].get('request', {}).get('url')
            key = os.path.split(req_url.split("?")[0])[1]
            sign_dict[key] = {"X-AppID": x_app_id, "X-Sign": x_sign, "X-Ts": x_ts}
    return sign_dict


def get_unisat_data_by_selenium(page_url):
    driver = get_selenium_driver()
    driver.execute_cdp_cmd("Page.addScriptToEvaluateOnNewDocument",
                           {"source": """Object.defineProperty(navigator, 'webdriver', {get: () => undefined})"""})
    driver.get(page_url)
    wait = WebDriverWait(driver, 15, 0.5)
    try:
        wait.until(expected_conditions.presence_of_element_located((By.CLASS_NAME, "item ")))
    except Exception as e:
        print("WebDriverWait.until timeout error: {}".format(e))
    res_body_dict = dict()
    for log_type in driver.log_types:
        perf_list = driver.get_log(log_type)
        for row_log in perf_list:
            try:
                log_json = json.loads(row_log['message'])
                message_log = log_json['message']
            except Exception as e:
                print(e)
                continue
            if message_log.get('method') != 'Network.responseReceived':
                continue
            if message_log.get("params", {}).get("type", "").upper() != "XHR":
                continue
            request_id = message_log['params'].get("requestId")
            if not request_id:
                continue
            req_url = message_log['params'].get('response', {}).get('url')
            key = os.path.split(req_url.split("?")[0])[1]
            content = driver.execute_cdp_cmd('Network.getResponseBody', {'requestId': request_id})
            body = None
            try:
                body = json.loads(content["body"])
            except Exception as e:
                print("get_unisat_data_by_selenium() json loads error: {}, content:{}".format(e, content))
            res_body_dict[key] = body
    return res_body_dict


if __name__ == '__main__':
    url = "https://unisat.io/brc20?q=bc1pkmnh3nj89uns3yp2mtqqxjns65vy6ca6n5jvp4s8ua8nke69cnjs987vtp"
    print("get_sign_by_selenium(url):", get_sign_by_selenium(url))
    # print("get_unisat_data_by_selenium(url):", get_unisat_data_by_selenium(url))

附:关于selenium的使用可以参考之前的文章
【测试】Selenium的使用(常用属性方法、元素等待、操作cookie、操作元素、无头模式、获取HTML源码)
【测试】selenium反爬操作
【测试】修改selenium选项配置参数优化性能
【测试】在Linux(CentOS、Ubuntu)无界面服务器使用selenium
【测试】Selenium操作Cookie文章来源地址https://www.toymoban.com/news/detail-641561.html

到了这里,关于【测试】selenium拦截Ajax(XHR)等异步请求数据的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • 异步请求-AJAX

    什么是同步交互 首先用户向HTTP服务器提交一个处理请求。接着服务器端接收到请求后,按照预先编写好的程序中的业务逻辑进行处理,比如和数据库服务器进行数据信息交换。最后,服务器对请求进行响应,将结果返回给客户端,返回一个HTML在浏览器中显示,通常会有CSS样

    2024年02月06日
    浏览(33)
  • AJAX 异步请求详细教程

    $.ajax() – Jquery提供的 ajax 函数 注册验证用户名是否可用 register 页面 servlet 类 service 层 dao 层 $.get() 与 $.post() Ajax 返回数据类型 json 简介 json 全称为 JavaScript Object Notation – js 对象简谱,是一种轻量级的数据 交换格式;它是基于 ECMAScript,采用独立的编程语言的文本格式进行储

    2024年02月02日
    浏览(35)
  • Ajax 与 Axios 异步请求

    1. 网页中如何请求数据 2. 资源的请求方式 客户端请求服务器时,请求的方式有很多种,最常见的两种请求方式分别为 get 和 post 请求。 get 请求通常用于获取服务端资源(向服务器要资源) post 请求通常用于向服务器提交数据(往服务器发送资源) 1. 什么是Ajax 2. 为什么要学

    2024年03月28日
    浏览(51)
  • 异步请求(Ajax,axios,json)

    同步/异步请求 表单(前端)向后端发送请求,属于同步请求 同步 : 发一个请求, 给一个回应, 会用回应的内容 覆盖 掉浏览器中内容,这样会打断前端其他的正常操作,在现在的前端中,显得不太友好。 异步 : 不同步 前端正常输入时,可以同时与后端进行交互,后端响应的数据

    2024年02月12日
    浏览(53)
  • Selenium处理异步加载请求获取XHR消息体的2种方法

    目录 通过Log读取XHR 简单使用示例 异步加载情况下,不涉及浏览器全局的加载,因此selenium会直接往下执行,这就导致异步结果还没返回,脚本就继续执行了。 构造chrome driver: 通过log来获取xhr: 其中,上述中“message”的消息如下: 通过requestId可以获得详细的消息体: Git

    2023年04月08日
    浏览(33)
  • javascript二维数组(21)执行异步HTTP(Ajax)请求的方法($.get、$.post、$getJSON、$ajax)

    . g e t 、 .get、 . g e t 、 .post、 g e t J S O N 、 getJSON、 g e t J SON 、 ajax都是jQuery提供的用于执行异步HTTP(Ajax)请求的方法。每个方法都有其特定的用途和区别。 . g e t :这个方法使用 G E T 方式来进行异步请求。其语法结构为: .get:这个方法使用GET方式来进行异步请求。其语

    2024年02月07日
    浏览(47)
  • 同步_异步请求和Ajax并利用axios框架简化

    目录 同步和异步 原生的Ajax 创建XMLHttpRequest对象 常用方法 常用属性 axios框架 同步请求:发送请求后,会做出回应,回应的内容会覆盖浏览器中的内容,这样会打断其他正常的操作,显得不太友好,并且请求时会携带所有的信息。 异步请求:前端正常输入时,可以同时与后端

    2024年02月13日
    浏览(35)
  • Java网络开发(Asynchronous异步)—— 从 Jsp 到 Ajax 的 axios 到 vue & 同步请求 到 异步请求

    如果想做bilibili那样的边看视频边评论怎么搞?; 之前用jsp的方式,是无法实现这个需求的,因为每次评论后提交了评论,会把整个页面全部刷新,导致视频也回到未播放的初始状态,如下所示: 代码为: 这是因为,在每次浏览器请求后,只能等待服务器的响应,即这种方

    2024年02月09日
    浏览(59)
  • AJAX 使用 JavaScript 的 `XMLHttpRequest` 对象来向服务器发送异步请求

    AJAX 是一种使用异步 HTTP (Ajax) 请求获取和发送数据的技术。它使得网页能够进行异步更新,而不需要重新加载整个页面。通过使用 AJAX,可以在不重新加载整个页面的情况下,与服务器交换数据并更新部分网页内容。 AJAX 使用 JavaScript 的 XMLHttpRequest 对象来向服务器发送异步请

    2024年01月16日
    浏览(53)
  • Ajax XHR响应

    如需获得来自服务器的响应,请使用 XMLHttpRequest 对象的 responseText 或 responseXML 属性。 属性 描述 responseText 获得字符串形式的响应数据。 responseXML 获得 XML 形式的响应数据。 如果来自服务器的响应并非 XML,请使用 responseText 属性。 responseText 属性返回字符串形式的响应,因此

    2023年04月27日
    浏览(34)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包