浏览器简单抽象

browser/IDriver.py

@@ -0,0 +1,2 @@
+class IDriver():
+    ...

browser/__init__.py

@@ -0,0 +1,3 @@
+from browser.edge import EdgeDriver
+
+driver = EdgeDriver()

browser/chrome.py

@@ -0,0 +1,59 @@
+from selenium import webdriver
+from selenium.webdriver import Proxy, DesiredCapabilities
+from selenium.webdriver.common.proxy import ProxyType
+
+from config.helper import config
+from browser.IDriver import IDriver
+from selenium.webdriver.chrome.options import Options
+
+
+class ChromeDriver(IDriver):
+    def __init__(self):
+        super(ChromeDriver, self).__init__()
+        options = Options()
+        if config()['webdriver']['headless']:
+            options.add_argument("--headless")
+        options.add_argument('--proxy-server=%s:%s' % (config()['mitm']['host'], config()['mitm']['port']))
+        options.add_argument('-ignore-certificate-errors')
+        options.add_argument('-ignore -ssl-errors')
+        options.add_argument('--incognito')
+        proxy = Proxy()
+        proxy.proxy_type = ProxyType.MANUAL
+        proxy.http_proxy = "%s:%s" % (config()['mitm']['host'], config()['mitm']['port'])
+        proxy.ssl_proxy = "%s:%s" % (config()['mitm']['host'], config()['mitm']['port'])
+        capabilities = DesiredCapabilities.EDGE
+        proxy.add_to_capabilities(capabilities)
+
+        self.browser = webdriver.Chrome(options=options,
+                                        desired_capabilities=capabilities,
+                                        executable_path=config()['webdriver']['chrome']['bin']
+                                        )
+
+    def new_tab(self) -> str:
+        current_window_handles = self.browser.window_handles
+        self.browser.execute_script("window.open('')")
+        new_window_handles = self.browser.window_handles
+        for _handle in new_window_handles:
+            if _handle not in current_window_handles:
+                return _handle
+        return ""
+
+    def change_tab(self, tab_handler: str):
+        if tab_handler not in self.browser.window_handles:
+            return
+        if tab_handler == self.browser.current_window_handle:
+            return
+        self.browser.switch_to_window(tab_handler)
+
+    def open_url(self, url: str, tab_handler: str = ""):
+        if tab_handler != "":
+            self.change_tab(tab_handler)
+        self.browser.get(url)
+
+    def screenshot(self, tab_handler: str = "") -> str:
+        cur_handle = self.browser.current_window_handle
+        if tab_handler != "":
+            self.change_tab(tab_handler)
+        b64 = self.browser.get_screenshot_as_base64()
+        self.change_tab(cur_handle)
+        return b64

browser/edge.py

@@ -0,0 +1,59 @@
+from selenium import webdriver
+from selenium.webdriver import Proxy, DesiredCapabilities
+from selenium.webdriver.common.proxy import ProxyType
+
+from config.helper import config
+from browser.IDriver import IDriver
+from selenium.webdriver.edge.options import Options
+
+
+class EdgeDriver(IDriver):
+    def __init__(self):
+        super(EdgeDriver, self).__init__()
+        options = Options()
+        if config()['webdriver']['headless']:
+            options.add_argument("--headless")
+        options.add_argument('--proxy-server=%s:%s' % (config()['mitm']['host'], config()['mitm']['port']))
+        options.add_argument('-ignore-certificate-errors')
+        options.add_argument('-ignore -ssl-errors')
+        options.add_argument('--incognito')
+        proxy = Proxy()
+        proxy.proxy_type = ProxyType.MANUAL
+        proxy.http_proxy = "%s:%s" % (config()['mitm']['host'], config()['mitm']['port'])
+        proxy.ssl_proxy = "%s:%s" % (config()['mitm']['host'], config()['mitm']['port'])
+        capabilities = DesiredCapabilities.EDGE
+        proxy.add_to_capabilities(capabilities)
+
+        self.browser = webdriver.Chrome(options=options,
+                                        desired_capabilities=capabilities,
+                                        executable_path=config()['webdriver']['edge']['bin']
+                                        )
+
+    def new_tab(self) -> str:
+        current_window_handles = self.browser.window_handles
+        self.browser.execute_script("window.open('')")
+        new_window_handles = self.browser.window_handles
+        for _handle in new_window_handles:
+            if _handle not in current_window_handles:
+                return _handle
+        return ""
+
+    def change_tab(self, tab_handler: str):
+        if tab_handler not in self.browser.window_handles:
+            return
+        if tab_handler == self.browser.current_window_handle:
+            return
+        self.browser.switch_to_window(tab_handler)
+
+    def open_url(self, url: str, tab_handler: str = ""):
+        if tab_handler != "":
+            self.change_tab(tab_handler)
+        self.browser.get(url)
+
+    def screenshot(self, tab_handler: str = "") -> str:
+        cur_handle = self.browser.current_window_handle
+        if tab_handler != "":
+            self.change_tab(tab_handler)
+        b64 = self.browser.get_screenshot_as_base64()
+        self.change_tab(cur_handle)
+        return b64

scripts/webdriver.py

@@ -1,49 +0,0 @@
-import requests
-import json
-
-from selenium import webdriver
-from selenium.webdriver.edge.options import Options
-from selenium.webdriver.common.desired_capabilities import DesiredCapabilities
-from selenium.webdriver.common.proxy import Proxy, ProxyType
-from selenium.webdriver.common.by import By
-from selenium.webdriver.support.ui import WebDriverWait
-from selenium.webdriver.support.expected_conditions import presence_of_element_located
-
-from config.helper import config
-
-def go(url):
-    chrome_options = Options()
-    chrome_options.add_argument('--proxy-server=%s:%s' % (config()['mitm']['host'], config()['mitm']['port']))
-
-    # 2022-04-09 添加一个忽略证书
-    chrome_options.add_argument('-ignore-certificate-errors')
-    chrome_options.add_argument('-ignore -ssl-errors')
-    chrome_options.add_argument('--incognito')
-
-    proxy = Proxy()
-    proxy.proxy_type = ProxyType.MANUAL
-    proxy.http_proxy = "%s:%s" % (config()['mitm']['host'], config()['mitm']['port'])
-    proxy.ssl_proxy = "%s:%s" % (config()['mitm']['host'], config()['mitm']['port'])
-
-    capabilities = DesiredCapabilities.EDGE
-    proxy.add_to_capabilities(capabilities)
-
-    with webdriver.Chrome(options=chrome_options,
-                            desired_capabilities=capabilities,
-                            executable_path=config()['webdriver']['bin']
-                            ) as driver:
-        wait = WebDriverWait(driver, 10)
-
-        driver.implicitly_wait(24 * 60 * 60)
-
-        driver.get(url)
-
-        first_result = wait.until(presence_of_element_located((By.ID, "RENDER_DATA")))
-        json_str = requests.utils.unquote(first_result.get_attribute("textContent"))
-        json_obj = json.loads(json_str)
-
-        roomInfo = json_obj['initialState']['roomStore']['roomInfo']
-        print(roomInfo)
-
-        wait.until(presence_of_element_located((By.CLASS_NAME, "oSu9Aw19")))
-