KenanZhu/AutoLibrary
1
1
Fork 0
mirror of https://github.com/KenanZhu/AutoLibrary.git synced 2026-08-02 06:09:36 +08:00
Code Issues Projects Activity

refactor(driver): WebDriver 子系统命名规范化与代码整理

Browse Source
This commit is contained in:
KenanZhu
2026-06-25 10:19:32 +08:00
parent 588718f9c5
commit 133433f2d8
3 changed files with 158 additions and 173 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/gui/ALWebDriverDownloadDialog.py
+98 -96
View File
@@ -8,6 +8,7 @@ You may use, modify, and distribute this file under the terms of the MIT License
See the LICENSE file for details. See the LICENSE file for details.
""" """
import threading import threading
from pathlib import Path
from typing import Optional from typing import Optional
from PySide6.QtCore import ( from PySide6.QtCore import (
@@ -33,7 +34,7 @@ from PySide6.QtGui import QCloseEvent
from managers.driver.WebDriverManager import ( from managers.driver.WebDriverManager import (
instance as webdriverInstance, instance as webdriverInstance,
WebDriverManager, WebDriverManager,
WebDriverInfo, WebDriverCtx,
WebDriverType, WebDriverType,
WebDriverStatus WebDriverStatus
) )
@@ -54,12 +55,12 @@ class DownloadWorker(QThread):
def __init__( def __init__(
self, self,
driver_manager: WebDriverManager, driver_manager: WebDriverManager,
driver_info: WebDriverInfo driver_ctx: WebDriverCtx
): ):
super().__init__() super().__init__()
self.__driver_manager = driver_manager self.__driver_manager = driver_manager
self.__driver_info = driver_info self.__driver_ctx = driver_ctx
self.__driver_path = None self.__driver_path: Optional[Path] = None
self.__cancelled = False self.__cancelled = False
self.__cancel_event = threading.Event() self.__cancel_event = threading.Event()
@@ -81,7 +82,7 @@ class DownloadWorker(QThread):
self.downloadCancelled.emit() self.downloadCancelled.emit()
return return
self.__driver_path = self.__driver_manager.installDriver( self.__driver_path = self.__driver_manager.installDriver(
self.__driver_info, self.__driver_ctx,
progress_callback=self.onProgress, progress_callback=self.onProgress,
cancel_event=self.__cancel_event cancel_event=self.__cancel_event
) )
@@ -140,12 +141,11 @@ class ALWebDriverDownloadDialog(CenterOnParentMixin, QDialog):
""" """
super().__init__(parent) super().__init__(parent)
self.__driver_dir = driver_dir self.__driver_dir = driver_dir
self.__driver_manager: Optional[WebDriverManager] = None self.__driver_manager: Optional[WebDriverManager] = None
self.__confirmed = False self.__confirmed = False
self.__selected_driver_info: Optional[WebDriverInfo] = None self.__selected_driver_ctx: Optional[WebDriverCtx] = None
self.__driver_infos: list[WebDriverInfo] = [] self.__driver_ctxs: list[WebDriverCtx] = []
self.__download_thread: Optional[DownloadWorker] = None self.__download_thread: Optional[DownloadWorker] = None
self.setupUi() self.setupUi()
@@ -153,6 +153,27 @@ class ALWebDriverDownloadDialog(CenterOnParentMixin, QDialog):
self.initializeDriverManager() self.initializeDriverManager()
self.refreshDriverList() self.refreshDriverList()
def closeEvent(
self,
event: QCloseEvent
):
if self.__download_thread and self.__download_thread.isRunning():
reply = QMessageBox.question(
self,
"确认关闭 - AutoLibrary",
"驱动正在下载中, 确定要取消并关闭对话框吗 ?",
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No
)
if reply == QMessageBox.StandardButton.No:
event.ignore()
return
self.__download_thread.stop()
if not self.__confirmed:
self.__selected_driver_ctx = None
event.accept()
super().closeEvent(event)
def setupUi( def setupUi(
self self
): ):
@@ -246,89 +267,42 @@ class ALWebDriverDownloadDialog(CenterOnParentMixin, QDialog):
if not self.__driver_manager: if not self.__driver_manager:
return return
self.__driver_manager.refresh() self.__driver_manager.refresh()
self.__driver_infos = self.__driver_manager.getDriverInfos() self.__driver_ctxs = self.__driver_manager.getDriverCtxs()
self.DriverComboBox.clear() self.DriverComboBox.clear()
installed = 0 installed = 0
installed_idx = 0 installed_idx = 0
for i, driver_info in enumerate(self.__driver_infos): for i, driver_ctx in enumerate(self.__driver_ctxs):
display_text = f"{driver_info.driver_type.value} - {driver_info.browser_version}" display_text = f"{driver_ctx.driver_type.value} - {driver_ctx.browser_version}"
if driver_info.driver_status == WebDriverStatus.INSTALLED: if driver_ctx.driver_status == WebDriverStatus.INSTALLED:
installed += 1 installed += 1
installed_idx = i # get the installed driver index installed_idx = i # get the installed driver index
display_text += " : 已安装" display_text += " - 已安装"
self.DriverComboBox.addItem(display_text) self.DriverComboBox.addItem(display_text)
count = len(self.__driver_infos) count = len(self.__driver_ctxs)
self.BrowserCountLabel.setText(f"检测到 {count} 个可用浏览器，{installed} 个已安装驱动：") self.BrowserCountLabel.setText(f"检测到 {count} 个可用浏览器，{installed} 个已安装驱动：")
if self.__driver_infos: if self.__driver_ctxs:
self.DriverComboBox.setCurrentIndex(installed_idx) self.DriverComboBox.setCurrentIndex(installed_idx)
def onDriverComboBoxChanged(
self,
index: int
):
if not self.__driver_infos or index < 0 or index >= len(self.__driver_infos):
return
driver_info = self.__driver_infos[index]
self.updateDriverInfoDisplay(driver_info)
self.updateProgressBarStates(driver_info)
self.updateButtonStates(driver_info)
def closeEvent(
self,
event: QCloseEvent
):
if self.__download_thread and self.__download_thread.isRunning():
reply = QMessageBox.question(
self,
"确认关闭 - AutoLibrary",
"驱动正在下载中, 确定要取消并关闭对话框吗 ?",
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No
)
if reply == QMessageBox.StandardButton.No:
event.ignore()
return
self.__download_thread.stop()
if not self.__confirmed:
self.__selected_driver_info = None
event.accept()
super().closeEvent(event)
def onThreadFinished(
self
):
if self.__download_thread:
self.__download_thread.deleteLater()
self.__download_thread = None
def getSelectedDriverInfo(
self
) -> Optional[WebDriverInfo]:
return self.__selected_driver_info
def updateDriverInfoDisplay( def updateDriverInfoDisplay(
self, self,
driver_info: WebDriverInfo driver_ctx: WebDriverCtx
): ):
if driver_info.driver_type == WebDriverType.CHROME: if driver_ctx.driver_type == WebDriverType.CHROME:
driver_type = "Google Chrome" driver_type = "Google Chrome"
elif driver_info.driver_type == WebDriverType.FIREFOX: elif driver_ctx.driver_type == WebDriverType.FIREFOX:
driver_type = "Mozilla Firefox" driver_type = "Mozilla Firefox"
elif driver_info.driver_type == WebDriverType.EDGE: elif driver_ctx.driver_type == WebDriverType.EDGE:
driver_type = "Microsoft Edge" driver_type = "Microsoft Edge"
else: else:
driver_type = "未知" driver_type = "未知"
self.BrowserTypeLabel.setText(f"类型：{driver_type}") self.BrowserTypeLabel.setText(f"类型：{driver_type}")
self.VersionLabel.setText(f"版本：{driver_info.driver_version}") self.VersionLabel.setText(f"版本：{driver_ctx.driver_version}")
if driver_info.driver_path: if driver_ctx.driver_path:
self.PathLabel.setText(str(driver_info.driver_path)) self.PathLabel.setText(str(driver_ctx.driver_path))
else: else:
self.PathLabel.setText("未安装") self.PathLabel.setText("未安装")
match driver_info.driver_status: match driver_ctx.driver_status:
case WebDriverStatus.NOT_INSTALLED: case WebDriverStatus.NOT_INSTALLED:
self.StatusLabel.status = ALStatusLabel.Status.WAITING self.StatusLabel.status = ALStatusLabel.Status.WAITING
case WebDriverStatus.INSTALLED: case WebDriverStatus.INSTALLED:
@@ -340,51 +314,70 @@ class ALWebDriverDownloadDialog(CenterOnParentMixin, QDialog):
def updateProgressBarStates( def updateProgressBarStates(
self, self,
driver_info: WebDriverInfo driver_ctx: WebDriverCtx
): ):
if driver_info.driver_status == WebDriverStatus.NOT_INSTALLED: if driver_ctx.driver_status == WebDriverStatus.NOT_INSTALLED:
self.ProgressBar.setValue(0) self.ProgressBar.setValue(0)
self.ProgressText.setText("未安装") self.ProgressText.setText("未安装")
elif driver_info.driver_status == WebDriverStatus.INSTALLED: elif driver_ctx.driver_status == WebDriverStatus.INSTALLED:
self.ProgressBar.setValue(100) self.ProgressBar.setValue(100)
self.ProgressText.setText("已安装") self.ProgressText.setText("已安装")
elif driver_info.driver_status == WebDriverStatus.DOWNLOADING: elif driver_ctx.driver_status == WebDriverStatus.DOWNLOADING:
pass # update by worker thread pass # update by worker thread
elif driver_info.driver_status == WebDriverStatus.ERROR: elif driver_ctx.driver_status == WebDriverStatus.ERROR:
self.ProgressBar.setValue(0) self.ProgressBar.setValue(0)
self.ProgressText.setText("下载失败") self.ProgressText.setText("下载失败")
def updateButtonStates( def updateButtonStates(
self, self,
driver_info: WebDriverInfo driver_ctx: WebDriverCtx
): ):
if driver_info.driver_status == WebDriverStatus.NOT_INSTALLED: if driver_ctx.driver_status == WebDriverStatus.NOT_INSTALLED:
self.RefreshButton.setEnabled(True) self.RefreshButton.setEnabled(True)
self.DeleteButton.setEnabled(False) self.DeleteButton.setEnabled(False)
self.DownloadButton.setEnabled(True) self.DownloadButton.setEnabled(True)
self.CancelButton.setEnabled(True) self.CancelButton.setEnabled(True)
self.ConfirmButton.setEnabled(False) self.ConfirmButton.setEnabled(False)
elif driver_info.driver_status == WebDriverStatus.INSTALLED: elif driver_ctx.driver_status == WebDriverStatus.INSTALLED:
self.RefreshButton.setEnabled(True) self.RefreshButton.setEnabled(True)
self.DownloadButton.setEnabled(False) self.DownloadButton.setEnabled(False)
self.DeleteButton.setEnabled(True) self.DeleteButton.setEnabled(True)
self.CancelButton.setEnabled(True) self.CancelButton.setEnabled(True)
self.ConfirmButton.setEnabled(True) self.ConfirmButton.setEnabled(True)
elif driver_info.driver_status == WebDriverStatus.DOWNLOADING: elif driver_ctx.driver_status == WebDriverStatus.DOWNLOADING:
self.RefreshButton.setEnabled(False) self.RefreshButton.setEnabled(False)
self.DownloadButton.setEnabled(False) self.DownloadButton.setEnabled(False)
self.DeleteButton.setEnabled(False) self.DeleteButton.setEnabled(False)
self.CancelButton.setEnabled(True) self.CancelButton.setEnabled(True)
self.ConfirmButton.setEnabled(False) self.ConfirmButton.setEnabled(False)
elif driver_info.driver_status == WebDriverStatus.ERROR: elif driver_ctx.driver_status == WebDriverStatus.ERROR:
self.RefreshButton.setEnabled(True) self.RefreshButton.setEnabled(True)
self.DownloadButton.setEnabled(True) self.DownloadButton.setEnabled(True)
self.DeleteButton.setEnabled(False) self.DeleteButton.setEnabled(False)
self.CancelButton.setEnabled(True) self.CancelButton.setEnabled(True)
self.ConfirmButton.setEnabled(False) self.ConfirmButton.setEnabled(False)
def getSelectedDriverInfo(
self
) -> Optional[WebDriverCtx]:
return self.__selected_driver_ctx
@Slot(int)
def onDriverComboBoxChanged(
self,
index: int
):
if not self.__driver_ctxs or index < 0 or index >= len(self.__driver_ctxs):
return
driver_ctx = self.__driver_ctxs[index]
self.updateDriverInfoDisplay(driver_ctx)
self.updateProgressBarStates(driver_ctx)
self.updateButtonStates(driver_ctx)
@Slot() @Slot()
def onRefreshButtonClicked( def onRefreshButtonClicked(
self self
@@ -398,9 +391,9 @@ class ALWebDriverDownloadDialog(CenterOnParentMixin, QDialog):
): ):
index = self.DriverComboBox.currentIndex() index = self.DriverComboBox.currentIndex()
if index < 0 or index >= len(self.__driver_infos): if index < 0 or index >= len(self.__driver_ctxs):
return return
driver_info = self.__driver_infos[index] driver_info = self.__driver_ctxs[index]
if driver_info.driver_status.name != "INSTALLED": if driver_info.driver_status.name != "INSTALLED":
QMessageBox.information(self, "提示 - AutoLibrary", "该驱动未安装, 无需删除") QMessageBox.information(self, "提示 - AutoLibrary", "该驱动未安装, 无需删除")
return return
@@ -426,9 +419,9 @@ class ALWebDriverDownloadDialog(CenterOnParentMixin, QDialog):
self.DriverComboBox.setEnabled(False) self.DriverComboBox.setEnabled(False)
index = self.DriverComboBox.currentIndex() index = self.DriverComboBox.currentIndex()
if index < 0 or index >= len(self.__driver_infos): if index < 0 or index >= len(self.__driver_ctxs):
return return
driver_info = self.__driver_infos[index] driver_info = self.__driver_ctxs[index]
if driver_info.driver_status == WebDriverStatus.INSTALLED: if driver_info.driver_status == WebDriverStatus.INSTALLED:
return return
driver_info.driver_status = WebDriverStatus.DOWNLOADING # we set this only to update driver_info.driver_status = WebDriverStatus.DOWNLOADING # we set this only to update
@@ -444,7 +437,7 @@ class ALWebDriverDownloadDialog(CenterOnParentMixin, QDialog):
self.__download_thread.downloadFinished.connect(self.onDownloadFinished) self.__download_thread.downloadFinished.connect(self.onDownloadFinished)
self.__download_thread.downloadError.connect(self.onDownloadError) self.__download_thread.downloadError.connect(self.onDownloadError)
self.__download_thread.downloadCancelled.connect(self.onDownloadCancelled) self.__download_thread.downloadCancelled.connect(self.onDownloadCancelled)
self.__download_thread.finished.connect(self.onThreadFinished) self.__download_thread.finished.connect(self.onDownloadThreadFinished)
self.__download_thread.start() self.__download_thread.start()
@Slot() @Slot()
@@ -472,8 +465,8 @@ class ALWebDriverDownloadDialog(CenterOnParentMixin, QDialog):
self.DriverComboBox.setEnabled(True) self.DriverComboBox.setEnabled(True)
index = self.DriverComboBox.currentIndex() index = self.DriverComboBox.currentIndex()
if 0 <= index < len(self.__driver_infos): if 0 <= index < len(self.__driver_ctxs):
driver_info = self.__driver_infos[index] driver_info = self.__driver_ctxs[index]
driver_info.driver_status = WebDriverStatus.INSTALLED driver_info.driver_status = WebDriverStatus.INSTALLED
self.updateDriverInfoDisplay(driver_info) self.updateDriverInfoDisplay(driver_info)
self.updateProgressBarStates(driver_info) self.updateProgressBarStates(driver_info)
@@ -487,8 +480,8 @@ class ALWebDriverDownloadDialog(CenterOnParentMixin, QDialog):
self.DriverComboBox.setEnabled(True) self.DriverComboBox.setEnabled(True)
index = self.DriverComboBox.currentIndex() index = self.DriverComboBox.currentIndex()
if 0 <= index < len(self.__driver_infos): if 0 <= index < len(self.__driver_ctxs):
driver_info = self.__driver_infos[index] driver_info = self.__driver_ctxs[index]
driver_info.driver_status = WebDriverStatus.ERROR driver_info.driver_status = WebDriverStatus.ERROR
self.updateDriverInfoDisplay(driver_info) self.updateDriverInfoDisplay(driver_info)
self.updateProgressBarStates(driver_info) self.updateProgressBarStates(driver_info)
@@ -502,8 +495,8 @@ class ALWebDriverDownloadDialog(CenterOnParentMixin, QDialog):
self.DriverComboBox.setEnabled(True) self.DriverComboBox.setEnabled(True)
index = self.DriverComboBox.currentIndex() index = self.DriverComboBox.currentIndex()
if 0 <= index < len(self.__driver_infos): if 0 <= index < len(self.__driver_ctxs):
driver_info = self.__driver_infos[index] driver_info = self.__driver_ctxs[index]
self.__driver_manager.cancelDriverDownload(driver_info) self.__driver_manager.cancelDriverDownload(driver_info)
driver_info.driver_status = WebDriverStatus.NOT_INSTALLED driver_info.driver_status = WebDriverStatus.NOT_INSTALLED
self.updateDriverInfoDisplay(driver_info) self.updateDriverInfoDisplay(driver_info)
@@ -511,18 +504,27 @@ class ALWebDriverDownloadDialog(CenterOnParentMixin, QDialog):
self.updateButtonStates(driver_info) self.updateButtonStates(driver_info)
self.ProgressText.setText("下载已取消") self.ProgressText.setText("下载已取消")
@Slot()
def onDownloadThreadFinished(
self
):
if self.__download_thread:
self.__download_thread.deleteLater()
self.__download_thread = None
@Slot() @Slot()
def onConfirmButtonClicked( def onConfirmButtonClicked(
self self
): ):
index = self.DriverComboBox.currentIndex() index = self.DriverComboBox.currentIndex()
if index < 0 or index >= len(self.__driver_infos): if index < 0 or index >= len(self.__driver_ctxs):
return return
driver_info = self.__driver_infos[index] driver_ctx = self.__driver_ctxs[index]
if driver_info.driver_status != WebDriverStatus.INSTALLED: if driver_ctx.driver_status != WebDriverStatus.INSTALLED:
return return
self.__selected_driver_info = driver_info self.__selected_driver_ctx = driver_ctx
self.__confirmed = True self.__confirmed = True
self.accept() self.accept()
@@ -542,5 +544,5 @@ class ALWebDriverDownloadDialog(CenterOnParentMixin, QDialog):
self.__download_thread.cancel() self.__download_thread.cancel()
else: else:
self.__confirmed = False self.__confirmed = False
self.__selected_driver_info = None self.__selected_driver_ctx = None
self.reject() self.reject()
src/managers/driver/WebDriverDownloader.py
+2 -2
View File
@@ -181,7 +181,7 @@ class WebDriverFileName:
raise ValueError(f"不受支持的 web driver 类型 : {self.driver_type}") raise ValueError(f"不受支持的 web driver 类型 : {self.driver_type}")
class WebDriverURL: class WebDriverDownloadURL:
""" """
Web driver download URL Web driver download URL
""" """
@@ -242,7 +242,7 @@ class WebDriverDownloader:
self.driver_type = driver_type self.driver_type = driver_type
self.arch = driver_arch self.arch = driver_arch
self.version = driver_version self.version = driver_version
self.download_url = str(WebDriverURL(self.version, self.driver_type, self.arch)) self.download_url = str(WebDriverDownloadURL(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 = 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_dir.mkdir(mode=0o0755, parents=True, exist_ok=True)
self.download_path = self.download_dir/str(WebDriverFileName(self.version, self.driver_type, self.arch)) self.download_path = self.download_dir/str(WebDriverFileName(self.version, self.driver_type, self.arch))
src/managers/driver/WebDriverManager.py
+58 -75
View File
@@ -8,6 +8,7 @@ You may use, modify, and distribute this file under the terms of the MIT License
See the LICENSE file for details. See the LICENSE file for details.
""" """
import os import os
import shutil
import threading import threading
import packaging.version as ver import packaging.version as ver
@@ -19,7 +20,7 @@ from managers.driver.WebBrowserDetector import (
WebBrowserType, WebBrowserArch, WebBrowserInfo, WebBrowserDetector WebBrowserType, WebBrowserArch, WebBrowserInfo, WebBrowserDetector
) )
from managers.driver.WebDriverDownloader import ( from managers.driver.WebDriverDownloader import (
WebDriverArch, WebDriverType, WebDriverArch, WebDriverType, WebDriverExecName,
ChromeDriverDownloader, FirefoxDriverDownloader, EdgeDriverDownloader ChromeDriverDownloader, FirefoxDriverDownloader, EdgeDriverDownloader
) )
@@ -35,9 +36,9 @@ class WebDriverStatus(Enum):
ERROR = 3 ERROR = 3
class WebDriverInfo: class WebDriverCtx:
""" """
Web driver information. Web browser and driver context.
Attributes: Attributes:
driver_type (WebDriverType): Web driver type driver_type (WebDriverType): Web driver type
@@ -52,8 +53,8 @@ class WebDriverInfo:
self self
): ):
self.driver_type = None self.driver_type : Optional[WebDriverType] = None
self.driver_arch = None self.driver_arch : Optional[WebDriverArch] = None
self.driver_version = "" self.driver_version = ""
self.browser_version = "" self.browser_version = ""
self.driver_path: Optional[Path] = None self.driver_path: Optional[Path] = None
@@ -75,7 +76,7 @@ class WebDriverManager:
self.__driver_dir = os.path.abspath(driver_dir) self.__driver_dir = os.path.abspath(driver_dir)
self.__browser_detector = WebBrowserDetector() self.__browser_detector = WebBrowserDetector()
self.__driver_infos: list[WebDriverInfo] = [] self.__driver_ctxs: list[WebDriverCtx] = []
self.__initialized = False self.__initialized = False
self.__lock = threading.Lock() self.__lock = threading.Lock()
@@ -98,7 +99,7 @@ class WebDriverManager:
with self.__lock: with self.__lock:
browser_infos = self.__browser_detector.detect() browser_infos = self.__browser_detector.detect()
self.__driver_infos = [ self.__driver_ctxs = [
self._getDriverInfo(info) self._getDriverInfo(info)
for info in browser_infos for info in browser_infos
] ]
@@ -108,7 +109,7 @@ class WebDriverManager:
): ):
with self.__lock: with self.__lock:
for driver_info in self.__driver_infos: for driver_info in self.__driver_ctxs:
driver_path = self._getDriverPath(driver_info) driver_path = self._getDriverPath(driver_info)
if driver_path and driver_path.exists() and driver_path.is_file(): if driver_path and driver_path.exists() and driver_path.is_file():
# Repair missing execute permission on Unix # Repair missing execute permission on Unix
@@ -198,7 +199,7 @@ class WebDriverManager:
else: else:
raise ValueError(f"不支持的 Web 浏览器类型 : {browser_type}") raise ValueError(f"不支持的 Web 浏览器类型 : {browser_type}")
def _mapFirefoxDriverVersion( def _mapFirefoxBrowserVersionToDriver(
self, self,
version: str version: str
) -> str: ) -> str:
@@ -241,43 +242,27 @@ class WebDriverManager:
def _getDriverInfo( def _getDriverInfo(
self, self,
browser_info: WebBrowserInfo browser_info: WebBrowserInfo
) -> WebDriverInfo: ) -> WebDriverCtx:
driver_info = WebDriverInfo() driver_ctx = WebDriverCtx()
driver_info.driver_type = self._mapWebBrowserTypeToDriver(browser_info.browser_type) driver_ctx.driver_type = self._mapWebBrowserTypeToDriver(browser_info.browser_type)
driver_info.driver_arch = self._mapWebBrowserArchToDriver(browser_info.browser_type, browser_info.browser_arch) driver_ctx.driver_arch = self._mapWebBrowserArchToDriver(browser_info.browser_type, browser_info.browser_arch)
if browser_info.browser_type == WebBrowserType.FIREFOX: if browser_info.browser_type == WebBrowserType.FIREFOX:
driver_info.driver_version = self._mapFirefoxDriverVersion(browser_info.browser_version) driver_ctx.driver_version = self._mapFirefoxBrowserVersionToDriver(browser_info.browser_version)
else: else:
driver_info.driver_version = browser_info.browser_version driver_ctx.driver_version = browser_info.browser_version
driver_info.browser_version = browser_info.browser_version driver_ctx.browser_version = browser_info.browser_version
return driver_info return driver_ctx
def _getDriverPath( def _getDriverPath(
self, self,
driver_info: WebDriverInfo driver_ctx: WebDriverCtx
) -> Optional[Path]: ) -> Optional[Path]:
driver_type = driver_info.driver_type driver_type = driver_ctx.driver_type
driver_arch = driver_info.driver_arch driver_arch = driver_ctx.driver_arch
driver_version = driver_info.driver_version driver_version = driver_ctx.driver_version
if driver_type == WebDriverType.CHROME: exe_name = str(WebDriverExecName(driver_type, driver_arch))
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_version/driver_arch.value driver_dir = Path(self.__driver_dir)/driver_type.value/driver_version/driver_arch.value
driver_path = driver_dir/exe_name driver_path = driver_dir/exe_name
return driver_path return driver_path
@@ -289,28 +274,28 @@ class WebDriverManager:
self._detectBrowsers() self._detectBrowsers()
self._checkDriverStatus() self._checkDriverStatus()
def getDriverInfos( def getDriverCtxs(
self self
) -> list[WebDriverInfo]: ) -> list[WebDriverCtx]:
with self.__lock: with self.__lock:
return self.__driver_infos.copy() return self.__driver_ctxs.copy()
def getDriverInfo( def getDriverCtx(
self, self,
driver_type: WebDriverType driver_type: WebDriverType
) -> list[WebDriverInfo]: ) -> list[WebDriverCtx]:
with self.__lock: with self.__lock:
return [ return [
info info
for info in self.__driver_infos for info in self.__driver_ctxs
if info.driver_type == driver_type if info.driver_type == driver_type
] ]
def getDriverPath( def getDriverPath(
self, self,
driver_info: WebDriverInfo driver_info: WebDriverCtx
) -> Optional[Path]: ) -> Optional[Path]:
if driver_info and driver_info.driver_status == WebDriverStatus.INSTALLED: if driver_info and driver_info.driver_status == WebDriverStatus.INSTALLED:
@@ -319,28 +304,28 @@ class WebDriverManager:
def installDriver( def installDriver(
self, self,
driver_info: WebDriverInfo, driver_ctx: WebDriverCtx,
progress_callback: Optional[Callable[[float, int, float, str], None]] = None, progress_callback: Optional[Callable[[float, int, float, str], None]] = None,
cancel_event: Optional[threading.Event] = None cancel_event: Optional[threading.Event] = None
) -> Optional[Path]: ) -> Optional[Path]:
with self.__lock: with self.__lock:
if not driver_info: if not driver_ctx:
if progress_callback: if progress_callback:
progress_callback(0, 0, 0, "未找到浏览器信息") progress_callback(0, 0, 0, "未找到浏览器信息")
else: else:
raise ValueError("未找到浏览器信息") raise ValueError("未找到浏览器信息")
if driver_info and driver_info.driver_status == WebDriverStatus.DOWNLOADING: if driver_ctx and driver_ctx.driver_status == WebDriverStatus.DOWNLOADING:
if progress_callback: if progress_callback:
progress_callback(0, 0, 0, f"{driver_info.driver_type} 驱动正在下载中") progress_callback(0, 0, 0, f"{driver_ctx.driver_type} 驱动正在下载中")
else: else:
raise ValueError(f"{driver_info.driver_type} 驱动正在下载中") raise ValueError(f"{driver_ctx.driver_type} 驱动正在下载中")
try: try:
if not driver_info: if not driver_ctx:
raise ValueError("未找到浏览器信息") raise ValueError("未找到浏览器信息")
driver_arch = driver_info.driver_arch driver_arch = driver_ctx.driver_arch
driver_type = driver_info.driver_type driver_type = driver_ctx.driver_type
driver_version = driver_info.driver_version driver_version = driver_ctx.driver_version
downloader = None downloader = None
if driver_type == WebDriverType.CHROME: if driver_type == WebDriverType.CHROME:
downloader = ChromeDriverDownloader( downloader = ChromeDriverDownloader(
@@ -366,68 +351,66 @@ class WebDriverManager:
else: else:
raise ValueError(f"不支持的 Web Driver 类型") raise ValueError(f"不支持的 Web Driver 类型")
with self.__lock: with self.__lock:
driver_info.driver_status = WebDriverStatus.DOWNLOADING driver_ctx.driver_status = WebDriverStatus.DOWNLOADING
driver_path = downloader.download(progress_callback=progress_callback, cancel_event=cancel_event) driver_path = downloader.download(progress_callback=progress_callback, cancel_event=cancel_event)
with self.__lock: with self.__lock:
if driver_path: if driver_path:
driver_info.driver_path = driver_path driver_ctx.driver_path = driver_path
driver_info.driver_version = driver_version driver_ctx.driver_version = driver_version
driver_info.driver_status = WebDriverStatus.INSTALLED driver_ctx.driver_status = WebDriverStatus.INSTALLED
else: else:
driver_info.driver_status = WebDriverStatus.ERROR driver_ctx.driver_status = WebDriverStatus.ERROR
return driver_path return driver_path
except Exception as e: except Exception as e:
with self.__lock: with self.__lock:
driver_info.driver_status = WebDriverStatus.ERROR driver_ctx.driver_status = WebDriverStatus.ERROR
raise e raise e
def cancelDriverDownload( def cancelDriverDownload(
self, self,
driver_info: WebDriverInfo driver_ctx: WebDriverCtx
) -> bool: ) -> bool:
import shutil
try: try:
driver_path = self._getDriverPath(driver_info) driver_path = self._getDriverPath(driver_ctx)
if driver_path: if driver_path:
download_dir = driver_path.parent download_dir = driver_path.parent
if download_dir.exists(): if download_dir.exists():
shutil.rmtree(download_dir, ignore_errors=True) shutil.rmtree(download_dir, ignore_errors=True)
with self.__lock: with self.__lock:
driver_info.driver_path = None driver_ctx.driver_path = None
driver_info.driver_status = WebDriverStatus.NOT_INSTALLED driver_ctx.driver_status = WebDriverStatus.NOT_INSTALLED
return True return True
except Exception: except Exception:
return False return False
def uninstallDriver( def uninstallDriver(
self, self,
driver_info: WebDriverInfo, driver_ctx: WebDriverCtx,
progress_callback: Optional[Callable[[int, int, float, str], None]] = None progress_callback: Optional[Callable[[int, int, float, str], None]] = None
) -> bool: ) -> bool:
with self.__lock: with self.__lock:
if not driver_info: if not driver_ctx:
if progress_callback: if progress_callback:
progress_callback(0, 0, 0, "未找到浏览器信息") progress_callback(0, 0, 0, "未找到浏览器信息")
else: else:
raise ValueError("未找到浏览器信息") raise ValueError("未找到浏览器信息")
if driver_info.driver_status != WebDriverStatus.INSTALLED: if driver_ctx.driver_status != WebDriverStatus.INSTALLED:
if progress_callback: if progress_callback:
progress_callback(0, 0, 0, f"{driver_info.driver_type} 驱动未安装") progress_callback(0, 0, 0, f"{driver_ctx.driver_type} 驱动未安装")
else: else:
raise ValueError(f"{driver_info.driver_type} 驱动未安装") raise ValueError(f"{driver_ctx.driver_type} 驱动未安装")
try: try:
driver_path = driver_info.driver_path driver_path = driver_ctx.driver_path
driver_path.unlink() driver_path.unlink()
with self.__lock: with self.__lock:
driver_info.driver_path = None driver_ctx.driver_path = None
driver_info.driver_status = WebDriverStatus.NOT_INSTALLED driver_ctx.driver_status = WebDriverStatus.NOT_INSTALLED
return True return True
except Exception: except Exception:
with self.__lock: with self.__lock:
driver_info.driver_status = WebDriverStatus.ERROR driver_ctx.driver_status = WebDriverStatus.ERROR
raise raise
def driverDir( def driverDir(