1
1
mirror of https://github.com/KenanZhu/AutoLibrary.git synced 2026-06-17 23:13:03 +08:00

feat(WebDriverManager): 新增浏览器管理类 WebDriverManager

- 新增浏览器管理类,支持下载和管理浏览器驱动
This commit is contained in:
2026-03-20 19:20:43 +08:00
parent 571af554d2
commit 95aa2bb518
3 changed files with 1012 additions and 0 deletions
+166
View File
@@ -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
+443
View File
@@ -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)
+403
View File
@@ -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