Docker コンテナ環境で Selenium を使用して Web スクレイピングを行う際、特に docker-standalone-chrome などのリモート WebDriver を利用する場合、同時に複数の Chrome インスタンスが起動・実行されるわけではなく、単一のインスタンスを共有する必要が生じることがあります。 この制約に対処するため、スレッドセーフな方法で単一のブラウザインスタンスを管理するクラスを実装しました。 以下にそのコードを示します。

import threading
import traceback
import json
import time

from loguru import logger
from selenium import webdriver
from selenium.common import WebDriverException
from selenium.webdriver.common.by import By
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.support.wait import WebDriverWait

# 仮のモジュールまたは定数定義
# from utils.configs import ScreenshotDir, ChromeUserDataDir, REMOTE_DRIVER
ScreenshotDir = "./screenshots"  # 例
ChromeUserDataDir = "./user_data"  # 例
REMOTE_DRIVER = "http://localhost:4444/wd/hub"  # 例

def configure_chrome_options():
    """Chrome の起動オプションを設定します。"""
    options = webdriver.ChromeOptions()
    options.add_argument(f"user-data-dir={ChromeUserDataDir}")
    options.page_load_strategy = 'eager'
    options.add_experimental_option('excludeSwitches', ['enable-automation'])
    options.add_argument("--disable-popup-blocking")
    options.add_argument("--blink-settings=imagesEnabled=false")
    options.add_argument("--window-size=1920,1080")
    options.add_argument("--no-sandbox")
    options.add_argument("--disable-gpu")
    # options.add_argument('--headless') # ヘッドレスモードの場合
    # options.add_experimental_option('debuggerAddress', '127.0.0.1:9222') # デバッガーアドレス指定の場合
    return options

class BrowserInstance:
    """
    個々のブラウザセッションを管理するクラス。
    このクラス自体はスレッドセーフではありません。
    """
    def __init__(self, options=None):
        if options is None:
            options = configure_chrome_options()
        try:
            # Docker Chrome の場合、Remote WebDriver を使用
            self.driver = webdriver.Remote(command_executor=REMOTE_DRIVER, options=options)
            # ローカル Chrome の場合（テスト用）
            # self.driver = webdriver.Chrome(options=options)
            self.session_id = self.driver.session_id
            logger.info(f"Browser instance created with session ID: {self.session_id}")
        except Exception as e:
            logger.error(f"Failed to initialize browser instance: {e}")
            raise

    def navigate_to(self, url):
        """指定された URL に移動します。"""
        logger.info(f"Navigating to: {url}")
        self.driver.get(url)
        logger.info(f"Successfully navigated to: {url}")

    def wait_for_element_by_xpath(self, xpath_locator, timeout=10):
        """XPath で要素が表示されるまで待機します。"""
        try:
            WebDriverWait(self.driver, timeout, 1, ignored_exceptions=(WebDriverException,)).until(
                EC.element_to_be_clickable((By.XPATH, xpath_locator))
            )
        except Exception:
            logger.warning(f"Element not found by XPath: {xpath_locator} within {timeout} seconds.")

    def wait_for_element_by_id(self, element_id, timeout=10):
        """ID で要素が表示されるまで待機します。"""
        try:
            WebDriverWait(self.driver, timeout, 1, ignored_exceptions=(WebDriverException,)).until(
                EC.presence_of_element_located((By.ID, element_id))
            )
        except Exception:
            logger.warning(f"Element not found by ID: {element_id} within {timeout} seconds.")

    def get_current_page_source(self) -> str:
        """現在のページのソースコードを取得します。"""
        return self.driver.page_source

    def save_page_source(self, file_path):
        """現在のページのソースコードを指定されたファイルに保存します。"""
        with open(file_path, 'w', encoding='utf-8') as f:
            f.write(self.get_current_page_source())

    def click_element_by_xpath(self, xpath_locator):
        """XPath で指定された要素をクリックし、新しいタブに切り替えます。"""
        try:
            element = self.driver.find_element(By.XPATH, xpath_locator)
            if element:
                element.click()
                time.sleep(2) # クリック後の処理待機
                # 新しいタブに切り替わることを想定
                handles = self.driver.window_handles
                if handles:
                    self.driver.switch_to.window(handles[-1])
                return self.driver.current_url
            else:
                logger.warning(f"Element not found for click by XPath: {xpath_locator}")
                return ""
        except Exception as e:
            logger.error(f"Error clicking element by XPath {xpath_locator}: {e}")
            return ""

    def take_screenshot(self, filename="screenshot.png"):
        """スクリーンショットを指定されたファイル名で保存します。"""
        full_path = f"{ScreenshotDir}/{filename}"
        try:
            self.driver.save_screenshot(full_path)
            logger.info(f"Screenshot saved to: {full_path}")
        except Exception as e:
            logger.error(f"Failed to save screenshot to {full_path}: {e}")

    def execute_cdp_command(self, command, params):
        """Chrome DevTools Protocol (CDP) コマンドを実行します。"""
        resource = f"/session/{self.session_id}/chromium/send_command_and_get_result"
        url = self.driver.command_executor._url + resource
        body = json.dumps({'cmd': command, 'params': params})
        response = self.driver.command_executor._request('POST', url, body)
        return response.get('value')

    def close(self):
        """ブラウザインスタンスを閉じます。"""
        if self.driver:
            try:
                self.driver.quit()
                self.driver = None
                logger.info(f"Browser instance (session ID: {self.session_id}) closed.")
            except Exception as e:
                logger.error(f"Error while quitting browser instance {self.session_id}: {e}\n{traceback.format_exc()}")
                self.driver = None

class ThreadSafeBrowserManager:
    """
    スレッドセーフな単一 Selenium ブラウザインスタンスを管理するコンテキストマネージャー。
    `with ThreadSafeBrowserManager() as browser:` の形式で使用します。
    """
    _instance_lock = threading.Lock()
    _is_in_use = threading.Event()
    _browser_instance = None

    def __init__(self, options=None):
        self.options = options

    def __enter__(self):
        """コンテキストマネージャーのエントリポイント。ブラウザインスタンスを取得または作成します。"""
        with ThreadSafeBrowserManager._instance_lock:
            # インスタンスが使用中かチェックし、使用可能になるまで待機
            while ThreadSafeBrowserManager._is_in_use.is_set():
                logger.debug("Browser instance is in use. Waiting...")
                # Event.wait() はタイムアウトを設定可能
                if not ThreadSafeBrowserManager._is_in_use.wait(timeout=10):
                    logger.warning("Timeout while waiting for browser instance.")
                    continue # タイムアウトした場合、再度チェック
                # wait() が成功した場合（Event が set された場合）、ループを抜ける

            ThreadSafeBrowserManager._is_in_use.set() # インスタンスを使用中に設定
            try:
                if ThreadSafeBrowserManager._browser_instance is None:
                    logger.info("Creating a new browser instance...")
                    ThreadSafeBrowserManager._browser_instance = BrowserInstance(self.options)
                logger.info(f"Acquired browser instance (Session ID: {ThreadSafeBrowserManager._browser_instance.session_id}) for thread {threading.current_thread().name}.")
                return ThreadSafeBrowserManager._browser_instance
            except Exception as e:
                ThreadSafeBrowserManager._is_in_use.clear() # エラー発生時は使用フラグをクリア
                raise RuntimeError(f"Failed to acquire browser instance: {e}") from e

    def __exit__(self, exc_type, exc_value, traceback):
        """コンテキストマネージャーの終了ポイント。ブラウザインスタンスを解放します。"""
        with ThreadSafeBrowserManager._instance_lock:
            if ThreadSafeBrowserManager._browser_instance:
                # 必要であれば、ここでブラウザインスタンスのクリーンアップ処理（例：ページ遷移リセットなど）を行う
                # 例: ThreadSafeBrowserManager._browser_instance.navigate_to("about:blank")
                pass
            ThreadSafeBrowserManager._is_in_use.clear() # インスタンスの使用フラグをクリア
            ThreadSafeBrowserManager._instance_lock.notify_all() # 他のスレッドに通知
            logger.info(f"Released browser instance for thread {threading.current_thread().name}.")
            # 注意: ここでは BrowserInstance.close() は呼び出しません。
            # 実際には、アプリケーション終了時など、グローバルなリソース解放のタイミングで
            # BrowserInstance.close() を呼び出すのが適切です。
            # そうしないと、コンテキストを抜けるたびにブラウザが終了してしまいます。

# アプリケーション終了時にブラウザインスタンスをクリーンアップするための関数（例）
def cleanup_browser_instance():
    with ThreadSafeBrowserManager._instance_lock:
        if ThreadSafeBrowserManager._browser_instance:
            logger.info("Cleaning up global browser instance...")
            ThreadSafeBrowserManager._browser_instance.close()
            ThreadSafeBrowserManager._browser_instance = None
            ThreadSafeBrowserManager._is_in_use.clear()

# 例: イベントハンドラなどでアプリケーション終了を検知し、cleanup_browser_instance() を呼び出す
# import atexit
# atexit.register(cleanup_browser_instance)

  

使用例

この ThreadSafeBrowserManager クラスは、コンテキストマネージャーとして以下のように簡単に利用できます。

# 例：並列処理内で使用
def scrape_task(url):
    try:
        with ThreadSafeBrowserManager() as browser:
            browser.navigate_to(url)
            page_content = browser.get_current_page_source()
            # page_content を使った処理...
            logger.info(f"Successfully scraped {url} with thread {threading.current_thread().name}")
            return page_content
    except Exception as e:
        logger.error(f"Error in scraping task for {url}: {e}")
        return None

# 実際のアプリケーションでは、スレッドプールなどを使って scrape_task を実行します。
# 例：
# urls_to_scrape = ["http://example.com", "http://example.org"]
# results = []
# with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
#     futures = [executor.submit(scrape_task, url) for url in urls_to_scrape]
#     for future in concurrent.futures.as_completed(futures):
#         results.append(future.result())

# アプリケーション終了時のクリーンアップ
# cleanup_browser_instance()
  

この実装では、threading.Lock と threading.Event を使用して、複数のスレッドからのアクセスを同期し、単一の BrowserInstance オブジェクトが同時に複数のスレッドから操作されないように制御しています。 __enter__ メソッドでブラウザインスタンスの取得または作成を行い、__exit__ メソッドでインスタンスの使用終了を通知します。 ただし、__exit__ での quit() の呼び出しは、インスタンスがアプリケーション全体で共有されることを考慮し、コメントアウトしています。 実際のクリーンアップは、アプリケーションのライフサイクルに合わせて別途行うのが一般的です。