はじめに

ChromeのDevToolsなどで見るように、ブラウザ上で読み込んだファイルを読み込む方法を紹介します。
seleniumから直接読み込むことで、

リクエスト数を減らせる
適切な方法でリクエストできる (ヘッダ、Cookieなど)

といったメリットがあります。
特にリクエスト方法が複雑なサイトの場合、その条件をjsファイルなどから解析せずに自動化できるため便利です。

※スクレイピングが禁止されているサイトでの利用は控えてください

方法

以下の2つの方法があります。

selenium-wire を使う
- メリット
  - 簡単に使える
- デメリット
  - 実行中にエラーが出ることがある
  - 読み込んだデータが取得できないことがある
  - 「保護されていない通信」になる
自力で実装する
- メリット
  - 自分の使い方に合ったように作成できる
- デメリット
  - エラーなどには自分で対応する必要がある

(私は最初 selenium-wire を知らなかったため、自力で実装することになりました...)

画像の保存

それでは、qiitaのページにある公式イベントの画像を取得してみましょう。
ファイル名: original.jpg

1. `selenium-wire` を使う

seleniumwireのdriverにはrequestsというインスタンス変数があり、
そこにはブラウザで行った通信が保持されます。

selenium-wireを使った場合のコード

from seleniumwire import webdriver
from webdriver_manager.chrome import ChromeDriverManager

# driverの作成
driver = webdriver.Chrome(ChromeDriverManager().install())
driver.get('https://qiita.com/')

# URLに「original.jpg」が含まるリクエストを取得する
results = [item for item in driver.requests if 'original.jpg' in item.url]
if len(results) == 0:
    raise Exception('original.jpgが見つかりませんでした')

# リクエスト情報にある取得データを保存する
with open('original.jpg', 'wb') as f:
    f.write(results[0].response.body)

無事、画像が保存されていました。

2. 自力で実装する

実装部分は長くなってしまうため、selenium_loadというモジュールに書き、それを読み込んでいます。
1つ注意点として、使うseleniumのdriverには caps["goog:loggingPrefs"] = {"performance": "ALL"} という設定が必要です。

from selenium import webdriver
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.common.desired_capabilities import DesiredCapabilities
from webdriver_manager.chrome import ChromeDriverManager

import selenium_load

# driverの作成
caps = DesiredCapabilities.CHROME
caps["goog:loggingPrefs"] = {"performance": "ALL"}
service = Service(ChromeDriverManager().install())
driver = webdriver.Chrome(service=service, desired_capabilities=caps)
driver.get('https://qiita.com/')

# original.jpgの情報を取得
network_dict = selenium_load.get_network_dict(driver)
results = [network_info for url, network_info in network_dict.items() if 'original.jpg' in url]
if len(results) == 0:
    raise Exception('original.jpgが見つかりませんでした')

# 画像として保存
with open('original2.jpg', 'wb') as f:
    f.write(results[0].body)

こちらも問題なく保存されていました。

自作モジュール

driver.get_log('performance') をすることでdriverのログを取得することができます。
そのログを種類ごとに分類し、Network.getResponseBody というコマンドからbodyを取得しています。
(詳しい説明は後日追加予定です。)

selenium_load.py

import json

import cv2
import numpy as np
import selenium.common.exceptions


class NetworkInfo(object):
    def __init__(self, request_id):
        self.request_id = request_id

        self.url = None
        self.status_code = None
        self.content_type = None
        self.body = None
        self.request_headers = {}
        self.response_headers = {}
        self.is_full_request_headers = False
        self.is_full_response_headers = False

    def set_body(self, body):
        self.body = body

    def set_request_headers(self, headers, is_full):
        if not self.is_full_request_headers:
            self.request_headers = headers
            self.is_full_request_headers = is_full

    def set_response_headers(self, headers, is_full):
        if not self.is_full_response_headers:
            self.response_headers = headers
            self.content_type = self.get_content_type(headers)
            self.is_full_response_headers = is_full

    @staticmethod
    def get_content_type(response_headers):
        content_type_list = [value for key, value in response_headers.items() if key.lower() == 'content-type']
        if len(content_type_list) > 0:
            return content_type_list[0]
        else:
            return None


def get_network_dict(driver):
    events = _get_network_events(driver)
    network_dict = _get_network_dict(events, driver)
    return network_dict


def _get_network_events(driver):
    net_log = driver.get_log('performance')
    events = [json.loads(entry['message'])['message'] for entry in net_log]
    return events


def _get_network_dict(events, driver=None):
    results = {}
    for event in events:
        method = event['method']

        if method == 'Network.requestWillBeSent':
            request_id = event['params']['requestId']
            url = event['params']['request']['url']
            part_headers = _fix_headers_tuple(event['params']['request']['headers'])

            if request_id not in results.keys():
                results[request_id] = NetworkInfo(request_id)

            results[request_id].url = url
            results[request_id].set_request_headers(part_headers, False)
        elif method == 'Network.requestWillBeSentExtraInfo':
            request_id = event['params']['requestId']
            headers = _fix_headers_tuple(event['params']['headers'])

            if request_id not in results.keys():
                results[request_id] = NetworkInfo(request_id)

            results[request_id].set_request_headers(headers, True)
        elif method == 'Network.responseReceived':
            request_id = event['params']['requestId']
            url = event['params']['response']['url']
            part_headers = _fix_headers_tuple(event['params']['response']['headers'])

            if request_id not in results.keys():
                results[request_id] = NetworkInfo(request_id)

            results[request_id].url = url
            results[request_id].set_response_headers(part_headers, False)

            if driver is not None:
                try:
                    body_info = driver.execute_cdp_cmd(
                        'Network.getResponseBody', {'requestId': event['params']['requestId']})
                    results[request_id].set_body(body_info['body'])
                except selenium.common.exceptions.WebDriverException:
                    print(f"Failed to fetch: {url}")
                    pass

        elif method == 'Network.responseReceivedExtraInfo':
            request_id = event['params']['requestId']
            headers = _fix_headers_tuple(event['params']['headers'])
            status_code = event['params']['statusCode']

            if request_id not in results.keys():
                results[request_id] = NetworkInfo(request_id)

            results[request_id].status_code = status_code
            results[request_id].set_response_headers(headers, True)

    formatted_results = {result.url: result for result in results.values()}

    return formatted_results


def _fix_headers_tuple(headers):
    new_headers = {}
    for key in headers.keys():
        if type(headers[key]) is tuple:
            new_headers[key] = headers[key][0]
        else:
            new_headers[key] = headers[key]

    return new_headers

さいごに

今回はseleniumで読み込んだデータを抽出する方法について説明しました。
方法は selenium-wireを使った方法、自作モジュールを使った方法の2種類がありますが、
それぞれメリット・デメリットがあるので使い分けてみてください。
頻繁に使う場合は自作モジュールを使ったほうが調整できるので便利ですが、
たまに使う程度であれば selenium-wire のほうが楽で良い印象です。

それでは