diff --git a/src/managers/driver/WebBrowserDetector.py b/src/managers/driver/WebBrowserDetector.py new file mode 100644 index 0000000..1da85d8 --- /dev/null +++ b/src/managers/driver/WebBrowserDetector.py @@ -0,0 +1,166 @@ +import platform +import installed_browsers + +from pathlib import Path +from enum import Enum +from dataclasses import dataclass + + +class WebBrowserType(Enum): + """ + Web browser type + """ + + CHROME = "chrome" + FIREFOX = "firefox" + EDGE = "edge" + + +class WebBrowserArch(Enum): + """ + Web browser architecture + """ + + WINX86_32 = 0 + WINX86_64 = 1 + WINARM = 2 + + LINUXX86_32 = 3 + LINUXX86_64 = 4 + LINUXARM = 5 + + MACX86_64 = 6 + MACARM = 7 + +@dataclass +class WebBrowserInfo: + """ + Web browser information + + Attributes: + browser_arch (WebBrowserArch): Web browser architecture + browser_type (WebBrowserType): Web browser type + browser_version (str): Web browser version + browser_path (Path): Web browser executable file path + """ + + browser_arch: WebBrowserArch + browser_type: WebBrowserType + browser_version: str + browser_path: Path + + +class WebBrowserArchDetector: + """ + Web browser architecture detector + """ + + def __init__( + self + ): + + pass + + + def detect( + self + ) -> WebBrowserArch: + """ + Detect system architecture + + Returns: + WebBrowserArch: System architecture + """ + + system = platform.system() + machine = platform.machine().lower() + if system == "Windows": + if machine in ["amd64", "x86_64"]: + return WebBrowserArch.WINX86_64 + elif machine in ["i386", "i686", "x86"]: + return WebBrowserArch.WINX86_32 + elif machine in ["arm64", "aarch64"]: + return WebBrowserArch.WINARM + else: + return WebBrowserArch.WINX86_64 + elif system == "Darwin": + if machine in ["arm64", "aarch64"]: + return WebBrowserArch.MACARM + else: + return WebBrowserArch.MACX86_64 + elif system == "Linux": + if machine in ["amd64", "x86_64"]: + return WebBrowserArch.LINUXX86_64 + elif machine in ["i386", "i686", "x86"]: + return WebBrowserArch.LINUXX86_32 + elif machine in ["arm64", "aarch64"]: + return WebBrowserArch.LINUXARM + elif machine.startswith("arm"): + return WebBrowserArch.LINUXARM + else: + return WebBrowserArch.LINUXX86_64 + raise ValueError(f"不支持的系统架构 : {system} {machine}") + + +class WebBrowserDetector: + """ + Web browser detector + """ + + def __init__( + self + ): + + self.browser_arch = WebBrowserArchDetector().detect() + self.browser_infos : list[WebBrowserInfo] = [] + + + def detect( + self + ) -> list[WebBrowserInfo]: + + """ + Detect installed web browsers on the system. + + Returns: + list[WebBrowserInfo]: List of detected browser information objects. + """ + + self.browser_infos = [] + try: + all_browsers = installed_browsers.browsers() + except Exception as e: + self.browser_infos = [] + return self.browser_infos + + # Mapping from internal library name to our enum + type_map = { + 'chrome': WebBrowserType.CHROME, + 'firefox': WebBrowserType.FIREFOX, + 'msedge': WebBrowserType.EDGE, + } + for browser in all_browsers: + internal_name = browser.get('name', '').lower() + if internal_name not in type_map: + continue # Not one of the browsers we care about + version = browser.get('version') + if not version: + # Skip browsers with no version info (unlikely, but defensive) + continue + exe_path = browser.get('location') + if not exe_path: + continue + try: + path = Path(exe_path) + if not path.is_file(): + continue + except Exception: + continue # Invalid path + info = WebBrowserInfo( + browser_arch=self.browser_arch, # Use system architecture as fallback + browser_type=type_map[internal_name], + browser_version=version, + browser_path=path, + ) + self.browser_infos.append(info) + return self.browser_infos \ No newline at end of file diff --git a/src/managers/driver/WebDriverDownloader.py b/src/managers/driver/WebDriverDownloader.py new file mode 100644 index 0000000..de3bf74 --- /dev/null +++ b/src/managers/driver/WebDriverDownloader.py @@ -0,0 +1,443 @@ +import os +import time +import shutil +import requests +import zipfile +import tarfile + +from enum import Enum +from pathlib import Path +from typing import Optional, Callable + + +class WebDriverType(Enum): + """ + Web driver type + """ + + CHROME = "chrome" + FIREFOX = "firefox" + EDGE = "edge" + + +class WebDriverArch(Enum): + """ + Web driver architecture + """ + + class Chrome(Enum): + """ + Chrome web driver architecture + """ + + WINX86_32 = "win32" + WINX86_64 = "win64" + + # LINUX86_32 : no support for linux 32bit + LINUX86_64 = "linux64" + # LINUXARM : no support for linux arm64 + + MACX86_64 = "mac-x64" + MACARM = "mac-arm64" + + class Firefox(Enum): + """ + Firefox web driver architecture + """ + + WINX86_32 = "win32" + WINX86_64 = "win64" + WINARM = "win-aarch64" + + LINUXX86_32 = "linux32" + LINUXX86_64 = "linux64" + LINUXARM = "linux-aarch64" + + MACX86_64 = "macos" + MACARM = "macos-aarch64" + + class Edge(Enum): + """ + Edge web driver architecture + """ + + WINX86_32 = "win32" + WINX86_64 = "win64" + WINARM = "arm64" + + # LINUX86_32 : no support for linux 32bit + LINUXX86_64 = "linux64" + # LINUXARM : no support for linux arm64 + + MACX86_64 = "mac64" + MACARM = "mac64_m1" + + +class WebDriverName: + """ + Web driver name + """ + + def __init__( + self, + driver_type: WebDriverType + ): + + self.driver_type = driver_type + + + def __str__( + self + ) -> str: + + match self.driver_type: + case WebDriverType.CHROME: + return "chromedriver" + case WebDriverType.FIREFOX: + return "geckodriver" + case WebDriverType.EDGE: + return "msedgedriver" + case _: + raise ValueError(f"不受支持的 web driver 类型 : {self.driver_type}") + + +class WebDriverExecName: + """ + Web driver executable file name + """ + + def __init__( + self, + driver_type: WebDriverType, + arch: WebDriverArch + ): + + self.driver_type = driver_type + self.arch = arch + + + def __str__( + self + ) -> str: + + is_win = True if self.arch is WebDriverArch.Chrome.WINX86_32 or\ + self.arch is WebDriverArch.Chrome.WINX86_64 or\ + self.arch is WebDriverArch.Firefox.WINX86_32 or\ + self.arch is WebDriverArch.Firefox.WINX86_64 or\ + self.arch is WebDriverArch.Edge.WINX86_32 or\ + self.arch is WebDriverArch.Edge.WINX86_64 else False + match self.driver_type: + case WebDriverType.CHROME: + return f"{WebDriverName(self.driver_type)}" + (".exe" if is_win else "") + case WebDriverType.FIREFOX: + return f"{WebDriverName(self.driver_type)}" + (".exe" if is_win else "") + case WebDriverType.EDGE: + return f"{WebDriverName(self.driver_type)}" + (".exe" if is_win else "") + case _: + raise ValueError(f"不受支持的 web driver 类型 : {self.driver_type}") + + +class WebDriverFileName: + """\ + Web driver compressed file name + """ + + def __init__( + self, + version: str, + driver_type: WebDriverType, + arch: WebDriverArch + ): + + self.version = version + self.driver_type = driver_type + self.arch = arch + + def __str__( + self + ) -> str: + + match self.driver_type: + case WebDriverType.CHROME: + return f"{WebDriverName(self.driver_type)}-{self.arch.value}.zip" + case WebDriverType.FIREFOX: + if self.arch is WebDriverArch.Firefox.WINX86_32 or\ + self.arch is WebDriverArch.Firefox.WINX86_64: + suffix = "zip" + else: + suffix = "tar.gz" + return f"{WebDriverName(self.driver_type)}-v{self.version}-{self.arch.value}.{suffix}" + case WebDriverType.EDGE: + return f"edgedriver_{self.arch.value}.zip" # Edge web driver file name is different + case _: + raise ValueError(f"不受支持的 web driver 类型 : {self.driver_type}") + + +class WebDriverURL: + """ + Web driver download URL + """ + + def __init__( + self, + version: str, + driver_type: WebDriverType, + arch: WebDriverArch + ): + + self.version = version + self.driver_type = driver_type + self.arch = arch + self.file_name = str(WebDriverFileName(self.version, self.driver_type, self.arch)) + + + def __str__( + self + ) -> str: + + match self.driver_type: + case WebDriverType.CHROME: + return f"https://storage.googleapis.com/chrome-for-testing-public/"\ + f"{self.version}/"\ + f"{self.arch.value}/"\ + f"{self.file_name}" + case WebDriverType.FIREFOX: + return f"https://github.com/mozilla/geckodriver/releases/download/"\ + f"v{self.version}/"\ + f"{self.file_name}" + case WebDriverType.EDGE: + return f"https://msedgedriver.microsoft.com/"\ + f"{self.version}/"\ + f"{self.file_name}" + case _: + raise ValueError(f"不受支持的 web driver 类型 : {self.driver_type}") + + +class WebDriverDownloader: + """ + Base class for WebDriver downloaders + + Args: + driver_type (WebDriverType): Web driver type + version (str): WebDriver version + arch (WebDriverArch): WebDriver architecture + download_dir (str): Download directory + """ + + def __init__( + self, + driver_type: WebDriverType, + driver_version: str, + driver_arch: WebDriverArch, + download_dir: str + ): + + self.driver_type = driver_type + self.arch = driver_arch + self.version = driver_version + self.download_url = str(WebDriverURL(self.version, self.driver_type, self.arch)) + self.download_dir = Path(download_dir)/self.driver_type.value/self.version/self.arch.value + self.download_dir.mkdir(mode=0o0755, parents=True, exist_ok=True) + self.download_path = self.download_dir/str(WebDriverFileName(self.version, self.driver_type, self.arch)) + + + def download( + self, + progress_callback: Optional[Callable[[int, int, float, str], None]] = None + ) -> Optional[Path]: + + try: + # downlaod file : 0% - 98% + if not self._download(progress_callback): + return None + # verify file : 98% - 99% + if not self._verify(progress_callback): + return None + # extract file : 99% - 100% + driver_path = self._extract(progress_callback) + if not driver_path: + return None + return driver_path + except Exception: + return None + + + def _download( + self, + progress_callback: Optional[Callable[[int, int, float, str], None]] = None, + max_retries: int = 3 + ) -> bool: + + CHUNK_SIZE = 8192*8 # 64KB chunk + headers = { + 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36', + 'Accept-Encoding': 'gzip, deflate' + } + + for attempt in range(max_retries): + try: + # resume download if file exists + if self.download_path.exists(): + downloaded_size = self.download_path.stat().st_size + headers_ = headers.copy() + headers_['Range'] = f"bytes={downloaded_size}-" + mode = 'ab' + else: + downloaded_size = 0 + headers_ = headers + mode = 'wb' + # get response + response = requests.get(str(self.download_url), headers=headers_, stream=True, timeout=120) + if response.status_code not in [200, 206]: + if self.download_path.exists(): + self.download_path.unlink() + downloaded_size = 0 + mode = 'wb' + response = requests.get(str(self.download_url), headers=headers, stream=True) + response.raise_for_status() + # get total size + total_size = int(response.headers.get('Content-Length', 0)) + if response.status_code == 206: # Partial Content - server supports Range + total_size += downloaded_size + # download file with progress callback and speed calculation + start_time = time.time() + last_time = start_time + last_size = downloaded_size + last_progress = 0.0 + with open(self.download_path, mode) as f: + for chunk in response.iter_content(CHUNK_SIZE): + if not chunk: + continue + f.write(chunk) + downloaded_size += len(chunk) + if not progress_callback or total_size == 0: + continue + current_time = time.time() + current_progress = (downloaded_size/total_size)*98.0 + if current_progress - last_progress >= 1.0 or current_progress == 98.0: + elapsed = current_time - last_time + if elapsed > 0: + speed = (downloaded_size - last_size)/elapsed/1024.0 # KB/s + else: + speed = 0.0 + progress_callback(current_progress, 100, speed, "下载中...") + last_progress = current_progress + last_size = downloaded_size + last_time = current_time + if total_size > 0 and self.download_path.stat().st_size < total_size: + raise Exception(f"下载不完整 : {self.download_path.stat().st_size}/{total_size} 字节") + return True + except Exception as e: + if attempt < max_retries - 1: + progress_callback(0, 100, 0.0, "准备重试...") + time.sleep(1) + continue + raise e + + + def _verify( + self, + progress_callback: Optional[Callable[[int, int, float, str], None]] = None + ) -> bool: + + progress_callback(98, 100, 0.0, "验证完成") + return True + + + def _extract( + self, + progress_callback: Optional[Callable[[int, int, float, str], None]] = None + ) -> Optional[Path]: + + try: + progress_callback(98, 100, 0.0, "解压中...") + file_path_str = str(self.download_path) + if file_path_str.endswith('.tar.gz'): + with tarfile.open(self.download_path, 'r:gz') as tar_ref: + tar_ref.extractall(self.download_dir) + else: + with zipfile.ZipFile(self.download_path, 'r') as zip_ref: + zip_ref.extractall(self.download_dir) + driver_file = None + for root, _, files in os.walk(self.download_dir): + for file in files: + expected_name = str(WebDriverExecName(self.driver_type, self.arch)) + if file == str(expected_name): + src_path = Path(root, file) + dst_path = self.download_dir/file + src_path.rename(dst_path) + driver_file = dst_path + break + if driver_file: + break + if not driver_file: + raise FileNotFoundError(f"未找到 web driver 文件 : {expected_name}") + progress_callback(100, 100, 0.0, "解压完成") + self.download_path.unlink() + self._cleanup(driver_file) + return driver_file + except Exception: + return None + + + def _cleanup( + self, + driver_file: Path + ) -> None: + + for item in self.download_dir.iterdir(): + if item != driver_file: + if item.is_dir(): + shutil.rmtree(item) + else: + item.unlink() + + +class ChromeDriverDownloader(WebDriverDownloader): + """ + Chrome web driver downloader + + Only support version higher than 114 + """ + + def __init__( + self, + version: str, + arch: WebDriverArch, + download_dir: str + ): + + super().__init__(WebDriverType.CHROME, version, arch, download_dir) + + +class FirefoxDriverDownloader(WebDriverDownloader): + """ + Firefox web driver downloader + + This class do not resolve version mapping, + only support driver version higher than 0.17.0 + """ + + def __init__( + self, + version: str, + arch: WebDriverArch, + download_dir: str + ): + + super().__init__(WebDriverType.FIREFOX, version, arch, download_dir) + + +class EdgeDriverDownloader(WebDriverDownloader): + """ + Edge web driver downloader + """ + + def __init__( + self, + version: str, + arch: WebDriverArch, + download_dir: str + ): + + super().__init__(WebDriverType.EDGE, version, arch, download_dir) diff --git a/src/managers/driver/WebDriverManager.py b/src/managers/driver/WebDriverManager.py new file mode 100644 index 0000000..afd44fe --- /dev/null +++ b/src/managers/driver/WebDriverManager.py @@ -0,0 +1,403 @@ +# -*- coding: utf-8 -*- +""" +Copyright (c) 2026 KenanZhu. +All rights reserved. + +This software is provided "as is", without any warranty of any kind. +You may use, modify, and distribute this file under the terms of the MIT License. +See the LICENSE file for details. +""" +import os +import threading +import packaging.version as ver + +from enum import Enum +from pathlib import Path +from typing import Optional, Callable + +from managers.driver.WebBrowserDetector import ( + WebBrowserType, WebBrowserArch, WebBrowserInfo, WebBrowserDetector +) +from managers.driver.WebDriverDownloader import ( + WebDriverArch, WebDriverType, + ChromeDriverDownloader, FirefoxDriverDownloader, EdgeDriverDownloader +) + + +class DriverStatus(Enum): + """ + Web driver status. + """ + + NOT_INSTALLED = 0 + INSTALLED = 1 + DOWNLOADING = 2 + ERROR = 3 + + +class WebDriverInfo: + """ + Web driver information. + + Attributes: + browser_info (WebBrowserInfo): Web browser information + driver_type (WebDriverType): Web driver type + driver_version (str): Web driver version + driver_path (Optional[Path]): Web driver executable file path + driver_status (DriverStatus): Web driver status + """ + + def __init__( + self, + browser_info: WebBrowserInfo + ): + + self.browser_info = browser_info + self.driver_type = WebDriverType(browser_info.browser_type.value) + self.driver_version = "" + self.driver_path: Optional[Path] = None + self.driver_status = DriverStatus.NOT_INSTALLED + + +class WebDriverManager: + """ + Web Driver Manager Singleton Class + + Args: + driver_dir (str): The directory to store web drivers. + """ + + def __init__( + self, + driver_dir: str + ): + + self.__driver_dir = os.path.abspath(driver_dir) + self.__browser_detector = WebBrowserDetector() + self.__driver_infos: list[WebDriverInfo] = [] + self.__initialized = False + self.__lock = threading.Lock() + + self.initialize() + + + def initialize( + self + ): + + if self.__initialized: + return + os.makedirs(self.__driver_dir, exist_ok=True) + self._detectBrowsers() + self._checkDriverStatus() + self.__initialized = True + + + def _detectBrowsers( + self + ): + + with self.__lock: + browser_infos = self.__browser_detector.detect() + self.__driver_infos = [WebDriverInfo(info) for info in browser_infos] + + + def _checkDriverStatus( + self + ): + + with self.__lock: + for driver_info in self.__driver_infos: + driver_arch = self._mapWebBrowserArch( + driver_info.browser_info.browser_type, + driver_info.browser_info.browser_arch + ) + driver_path = self._getDriverPath( + driver_info.driver_type, + driver_arch + ) + if driver_path and driver_path.exists() and driver_path.is_file(): + driver_info.driver_path = driver_path + driver_info.driver_status = DriverStatus.INSTALLED + try: + driver_info.driver_version = self._getDriverVersion( + driver_info.driver_type, + driver_info.driver_info.browser_version + ) + except Exception: + driver_info.driver_status = DriverStatus.ERROR + + + def _mapWebBrowserArch( + self, + browser_type: WebBrowserType, + browser_arch: WebBrowserArch + ) -> WebDriverArch: + + if browser_type == WebBrowserType.CHROME: + if browser_arch == WebBrowserArch.WINX86_32: + return WebDriverArch.Chrome.WINX86_32 + elif browser_arch == WebBrowserArch.WINX86_64: + return WebDriverArch.Chrome.WINX86_64 + elif browser_arch == WebBrowserArch.WINARM: + raise ValueError("Chrome 不支持 Windows ARM 架构") + elif browser_arch == WebBrowserArch.LINUXX86_32: + raise ValueError("Chrome 不支持 Linux x86_32 架构") + elif browser_arch == WebBrowserArch.LINUXX86_64: + return WebDriverArch.Chrome.LINUXX86_64 + elif browser_arch == WebBrowserArch.LINUXARM: + raise ValueError("Chrome 不支持 Linux ARM 架构") + elif browser_arch == WebBrowserArch.MACX86_64: + return WebDriverArch.Chrome.MACX86_64 + elif browser_arch == WebBrowserArch.MACARM: + return WebDriverArch.Chrome.MACARM + else: + raise ValueError(f"不支持的 Chrome 浏览器架构 : {browser_arch}") + elif browser_type == WebBrowserType.FIREFOX: + if browser_arch == WebBrowserArch.WINX86_32: + return WebDriverArch.Firefox.WINX86_32 + elif browser_arch == WebBrowserArch.WINX86_64: + return WebDriverArch.Firefox.WINX86_64 + elif browser_arch == WebBrowserArch.WINARM: + return WebDriverArch.Firefox.WINARM + elif browser_arch == WebBrowserArch.LINUXX86_32: + return WebDriverArch.Firefox.LINUXX86_32 + elif browser_arch == WebBrowserArch.LINUXX86_64: + return WebDriverArch.Firefox.LINUXX86_64 + elif browser_arch == WebBrowserArch.LINUXARM: + return WebDriverArch.Firefox.LINUXARM + elif browser_arch == WebBrowserArch.MACX86_64: + return WebDriverArch.Firefox.MACX86_64 + elif browser_arch == WebBrowserArch.MACARM: + return WebDriverArch.Firefox.MACARM + else: + raise ValueError(f"不支持的 Firefox 浏览器架构 : {browser_arch}") + elif browser_type == WebBrowserType.EDGE: + if browser_arch == WebBrowserArch.WINX86_32: + return WebDriverArch.Edge.WINX86_32 + elif browser_arch == WebBrowserArch.WINX86_64: + return WebDriverArch.Edge.WINX86_64 + elif browser_arch == WebBrowserArch.WINARM: + return WebDriverArch.Edge.WINARM + elif browser_arch == WebBrowserArch.LINUXX86_32: + raise ValueError("Edge 不支持 Linux x86_32 架构") + elif browser_arch == WebBrowserArch.LINUXX86_64: + return WebDriverArch.Edge.LINUXX86_64 + elif browser_arch == WebBrowserArch.LINUXARM: + raise ValueError("Edge 不支持 Linux ARM 架构") + elif browser_arch == WebBrowserArch.MACX86_64: + return WebDriverArch.Edge.MACX86_64 + elif browser_arch == WebBrowserArch.MACARM: + return WebDriverArch.Edge.MACARM + else: + raise ValueError(f"不支持的 Edge 浏览器架构 : {browser_arch}") + else: + raise ValueError(f"不支持的 Web 浏览器类型 : {browser_type}") + + + def _mapFirefoxDriverVersion( + self, + version: str + ) -> str: + + version_mapping = [ + (ver.Version("128.0"), ver.Version("999.0"), "0.36.0"), + (ver.Version("115.0"), ver.Version("127.0"), "0.35.0"), + (ver.Version("91.0"), ver.Version("114.0"), "0.34.0"), + (ver.Version("91.0"), ver.Version("120.0"), "0.33.0"), + (ver.Version("91.0"), ver.Version("120.0"), "0.32.0"), + (ver.Version("91.0"), ver.Version("120.0"), "0.31.0"), + (ver.Version("78.0"), ver.Version("90.0"), "0.30.0"), + (ver.Version("60.0"), ver.Version("90.0"), "0.29.0"), + (ver.Version("60.0"), ver.Version("90.0"), "0.28.0"), + (ver.Version("60.0"), ver.Version("90.0"), "0.27.0"), + (ver.Version("57.0"), ver.Version("90.0"), "0.26.0"), + (ver.Version("55.0"), ver.Version("62.0"), "0.25.0"), + (ver.Version("55.0"), ver.Version("62.0"), "0.24.0"), + (ver.Version("57.0"), ver.Version("79.0"), "0.23.0"), + (ver.Version("57.0"), ver.Version("79.0"), "0.22.0"), + (ver.Version("57.0"), ver.Version("79.0"), "0.21.0"), + (ver.Version("55.0"), ver.Version("62.0"), "0.20.0"), + (ver.Version("55.0"), ver.Version("62.0"), "0.19.0"), + (ver.Version("53.0"), ver.Version("62.0"), "0.18.0"), + (ver.Version("52.0"), ver.Version("62.0"), "0.17.0"), + ] + + try: + firefox_version = ver.Version(version) + for min_ver, max_ver, gecko_ver in version_mapping: + if min_ver <= firefox_version <= max_ver: + return gecko_ver + raise ValueError( + f"不支持的 Firefox 版本 : {version}" + f"Firefox 版本 52 及以上受支持" + ) + except Exception as e: + raise ValueError(f"无效的 Firefox 版本格式 : {version}") from e + + + def _getDriverPath( + self, + driver_type: WebDriverType, + driver_arch: WebDriverArch + ) -> Optional[Path]: + + if driver_type == WebDriverType.CHROME: + driver_name = "chromedriver" + elif driver_type == WebDriverType.FIREFOX: + driver_name = "geckodriver" + elif driver_type == WebDriverType.EDGE: + driver_name = "msedgedriver" + else: + return None + is_win = driver_arch in [ + WebDriverArch.Chrome.WINX86_32, + WebDriverArch.Chrome.WINX86_64, + WebDriverArch.Firefox.WINX86_32, + WebDriverArch.Firefox.WINX86_64, + WebDriverArch.Edge.WINX86_32, + WebDriverArch.Edge.WINX86_64, + ] + exe_name = f"{driver_name}.exe" if is_win else driver_name + driver_dir = Path(self.__driver_dir) / driver_type.value / driver_arch.value + driver_path = driver_dir / exe_name + if driver_path.exists() and driver_path.is_file(): + return driver_path + return None + + + def _getDriverVersion( + self, + driver_type: WebDriverType, + browser_version: str + ) -> str: + + if driver_type == WebDriverType.FIREFOX: + return self._mapFirefoxDriverVersion(browser_version) + return browser_version + + + def refresh( + self + ): + + with self.__lock: + self._detectBrowsers() + self._checkDriverStatus() + + + def getDriverInfos( + self + ) -> list[WebDriverInfo]: + + with self.__lock: + return self.__driver_infos.copy() + + + def getDriverInfo( + self, + driver_type: WebDriverType + ) -> Optional[WebDriverInfo]: + + with self.__lock: + for driver_info in self.__driver_infos: + if driver_info.driver_type == driver_type: + return driver_info + return None + + + def getDriverPath( + self, + driver_type: WebDriverType + ) -> Optional[Path]: + + driver_info = self.getDriverInfo(driver_type) + if driver_info and driver_info.driver_status == DriverStatus.INSTALLED: + return driver_info.driver_path + return None + + + def installDriver( + self, + driver_type: WebDriverType, + progress_callback: Optional[Callable[[int, int, float, str], None]] = None + ) -> Optional[Path]: + + with self.__lock: + driver_info = self.getDriverInfo(driver_type) + if not driver_info: + raise ValueError(f"未找到类型为 {driver_type} 的浏览器") + if driver_info.driver_status == DriverStatus.DOWNLOADING: + raise ValueError(f"{driver_type} 驱动正在下载中") + driver_info.driver_status = DriverStatus.DOWNLOADING + try: + driver_arch = self._mapWebBrowserArch( + driver_info.browser_info.browser_type, + driver_info.browser_info.browser_arch + ) + browser_version = driver_info.browser_info.browser_version + driver_version = self._getDriverVersion(driver_type, browser_version) + downloader = None + if driver_type == WebDriverType.CHROME: + downloader = ChromeDriverDownloader( + version=driver_version, + arch=driver_arch, + download_dir=self.__driver_dir + ) + elif driver_type == WebDriverType.FIREFOX: + downloader = FirefoxDriverDownloader( + version=driver_version, + arch=driver_arch, + download_dir=self.__driver_dir + ) + elif driver_type == WebDriverType.EDGE: + downloader = EdgeDriverDownloader( + version=driver_version, + arch=driver_arch, + download_dir=self.__driver_dir + ) + if downloader is None: + raise ValueError(f"不支持的 Web Driver 类型 : {driver_type}") + + driver_path = downloader.download(progress_callback=progress_callback) + with self.__lock: + if driver_path: + driver_info.driver_path = driver_path + driver_info.driver_version = driver_version + driver_info.driver_status = DriverStatus.INSTALLED + else: + driver_info.driver_status = DriverStatus.ERROR + return driver_path + except Exception as e: + with self.__lock: + driver_info.driver_status = DriverStatus.ERROR + raise + + + def driverDir( + self + ) -> str: + + return self.__driver_dir + + +# WebDriverManager singleton instance. +_webdriver_manager_instance = None + +# Singleton instance lock. +_instance_lock = threading.Lock() + +def instance( + driver_dir: str = "" +) -> WebDriverManager: + + global _webdriver_manager_instance + with _instance_lock: + if _webdriver_manager_instance is None: + if not driver_dir: + raise ValueError("WebDriverManager 需要驱动目录参数") + _webdriver_manager_instance = WebDriverManager(driver_dir) + else: + if driver_dir and _webdriver_manager_instance.driverDir() != os.path.abspath(driver_dir): + raise ValueError("WebDriverManager 的实例已初始化,不能使用不同的驱动目录") + return _webdriver_manager_instance