github/FlareSolverr
1
1
Fork 0
mirror of https://github.com/FlareSolverr/FlareSolverr.git synced 2025-06-12 06:17:14 +00:00
Code Issues Packages Projects Releases Wiki Activity

Update undetected_chromedriver version to 3.4.6 (#715)

Browse Source 
Co-authored-by: ilike2burnthing <59480337+ilike2burnthing@users.noreply.github.com>
This commit is contained in:
Artemiy Ryabinkov 2023-03-06 13:57:38 +00:00 committed by GitHub
parent 3a6e8e0f92
commit 96fcd21174
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 206 additions and 469 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

129
src/undetected_chromedriver/__init__.py
View File

@ -17,7 +17,7 @@ by UltrafunkAmsterdam (https://github.com/ultrafunkamsterdam)
from __future__ import annotations from __future__ import annotations
__version__ = "3.2.1" __version__ = "3.4.6"
import json import json
import logging import logging
@ -122,7 +122,7 @@ class Chrome(selenium.webdriver.chrome.webdriver.WebDriver):
suppress_welcome=True, suppress_welcome=True,
use_subprocess=False, use_subprocess=False,
debug=False, debug=False,
no_sandbox=True, no_sandbox=True,
windows_headless=False, windows_headless=False,
**kw, **kw,
): ):
@ -239,13 +239,13 @@ class Chrome(selenium.webdriver.chrome.webdriver.WebDriver):
finalize(self, self._ensure_close, self) finalize(self, self._ensure_close, self)
self.debug = debug self.debug = debug
patcher = Patcher( self.patcher = Patcher(
executable_path=driver_executable_path, executable_path=driver_executable_path,
force=patcher_force_close, force=patcher_force_close,
version_main=version_main, version_main=version_main,
) )
patcher.auto() self.patcher.auto()
self.patcher = patcher # self.patcher = patcher
if not options: if not options:
options = ChromeOptions() options = ChromeOptions()
@ -287,6 +287,11 @@ class Chrome(selenium.webdriver.chrome.webdriver.WebDriver):
# see if a custom user profile is specified in options # see if a custom user profile is specified in options
for arg in options.arguments: for arg in options.arguments:
if any([_ in arg for _ in ("--headless", "headless")]):
options.arguments.remove(arg)
options.headless = True
if "lang" in arg: if "lang" in arg:
m = re.search("(?:--)?lang(?:[ =])?(.*)", arg) m = re.search("(?:--)?lang(?:[ =])?(.*)", arg)
try: try:
@ -365,13 +370,18 @@ class Chrome(selenium.webdriver.chrome.webdriver.WebDriver):
options.arguments.extend(["--no-default-browser-check", "--no-first-run"]) options.arguments.extend(["--no-default-browser-check", "--no-first-run"])
if no_sandbox: if no_sandbox:
options.arguments.extend(["--no-sandbox", "--test-type"]) options.arguments.extend(["--no-sandbox", "--test-type"])
if headless or options.headless: if headless or options.headless:
options.headless = True if self.patcher.version_main < 108:
options.add_argument("--window-size=1920,1080") options.add_argument("--headless=chrome")
options.add_argument("--start-maximized") elif self.patcher.version_main >= 108:
options.add_argument("--no-sandbox") options.add_argument("--headless=new")
# fixes "could not connect to chrome" error when running
# on linux using privileged user like root (which i don't recommend) options.add_argument("--window-size=1920,1080")
options.add_argument("--start-maximized")
options.add_argument("--no-sandbox")
# fixes "could not connect to chrome" error when running
# on linux using privileged user like root (which i don't recommend)
options.add_argument( options.add_argument(
"--log-level=%d" % log_level "--log-level=%d" % log_level
@ -408,23 +418,23 @@ class Chrome(selenium.webdriver.chrome.webdriver.WebDriver):
self.browser_pid = start_detached( self.browser_pid = start_detached(
options.binary_location, *options.arguments options.binary_location, *options.arguments
) )
else: else:
startupinfo = subprocess.STARTUPINFO() startupinfo = subprocess.STARTUPINFO()
if os.name == 'nt' and windows_headless: if os.name == 'nt' and windows_headless:
startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW
browser = subprocess.Popen( browser = subprocess.Popen(
[options.binary_location, *options.arguments], [options.binary_location, *options.arguments],
stdin=subprocess.PIPE, stdin=subprocess.PIPE,
stdout=subprocess.PIPE, stdout=subprocess.PIPE,
stderr=subprocess.PIPE, stderr=subprocess.PIPE,
close_fds=IS_POSIX, close_fds=IS_POSIX,
startupinfo=startupinfo startupinfo=startupinfo
) )
self.browser_pid = browser.pid self.browser_pid = browser.pid
if service_creationflags: if service_creationflags:
service = selenium.webdriver.common.service.Service( service = selenium.webdriver.common.service.Service(
patcher.executable_path, port, service_args, service_log_path self.patcher.executable_path, port, service_args, service_log_path
) )
for attr_name in ("creationflags", "creation_flags"): for attr_name in ("creationflags", "creation_flags"):
if hasattr(service, attr_name): if hasattr(service, attr_name):
@ -434,7 +444,7 @@ class Chrome(selenium.webdriver.chrome.webdriver.WebDriver):
service = None service = None
super(Chrome, self).__init__( super(Chrome, self).__init__(
executable_path=patcher.executable_path, executable_path=self.patcher.executable_path,
port=port, port=port,
options=options, options=options,
service_args=service_args, service_args=service_args,
@ -475,18 +485,18 @@ class Chrome(selenium.webdriver.chrome.webdriver.WebDriver):
{ {
"source": """ "source": """
Object.defineProperty(window, 'navigator', { Object.defineProperty(window, "navigator", {
value: new Proxy(navigator, { Object.defineProperty(window, "navigator", {
has: (target, key) => (key === 'webdriver' ? false : key in target), value: new Proxy(navigator, {
get: (target, key) => has: (target, key) => (key === "webdriver" ? false : key in target),
key === 'webdriver' ? get: (target, key) =>
false : key === "webdriver"
typeof target[key] === 'function' ? ? false
target[key].bind(target) : : typeof target[key] === "function"
target[key] ? target[key].bind(target)
}) : target[key],
}); }),
});
""" """
}, },
) )
@ -605,37 +615,38 @@ class Chrome(selenium.webdriver.chrome.webdriver.WebDriver):
self.get = get_wrapped self.get = get_wrapped
def _get_cdc_props(self): # def _get_cdc_props(self):
return self.execute_script( # return self.execute_script(
""" # """
let objectToInspect = window, # let objectToInspect = window,
result = []; # result = [];
while(objectToInspect !== null) # while(objectToInspect !== null)
{ result = result.concat(Object.getOwnPropertyNames(objectToInspect)); # { result = result.concat(Object.getOwnPropertyNames(objectToInspect));
objectToInspect = Object.getPrototypeOf(objectToInspect); } # objectToInspect = Object.getPrototypeOf(objectToInspect); }
return result.filter(i => i.match(/.+_.+_(Array|Promise|Symbol)/ig)) #
""" # return result.filter(i => i.match(/^([a-zA-Z]){27}(Array|Promise|Symbol)$/ig))
) # """
# )
def _hook_remove_cdc_props(self): #
self.execute_cdp_cmd( # def _hook_remove_cdc_props(self):
"Page.addScriptToEvaluateOnNewDocument", # self.execute_cdp_cmd(
{ # "Page.addScriptToEvaluateOnNewDocument",
"source": """ # {
let objectToInspect = window, # "source": """
result = []; # let objectToInspect = window,
while(objectToInspect !== null) # result = [];
{ result = result.concat(Object.getOwnPropertyNames(objectToInspect)); # while(objectToInspect !== null)
objectToInspect = Object.getPrototypeOf(objectToInspect); } # { result = result.concat(Object.getOwnPropertyNames(objectToInspect));
result.forEach(p => p.match(/.+_.+_(Array|Promise|Symbol)/ig) # objectToInspect = Object.getPrototypeOf(objectToInspect); }
&&delete window[p]&&console.log('removed',p)) # result.forEach(p => p.match(/^([a-zA-Z]){27}(Array|Promise|Symbol)$/ig)
""" # &&delete window[p]&&console.log('removed',p))
}, # """
) # },
# )
def get(self, url): def get(self, url):
if self._get_cdc_props(): # if self._get_cdc_props():
self._hook_remove_cdc_props() # self._hook_remove_cdc_props()
return super().get(url) return super().get(url)
def add_cdp_listener(self, event_name, callback): def add_cdp_listener(self, event_name, callback):
@ -702,7 +713,7 @@ class Chrome(selenium.webdriver.chrome.webdriver.WebDriver):
def quit(self): def quit(self):
try: try:
self.service.process.kill() self.service.process.kill()
self.service.process.wait(5) self.service.process.wait(5)
logger.debug("webdriver process ended") logger.debug("webdriver process ended")
except (AttributeError, RuntimeError, OSError): except (AttributeError, RuntimeError, OSError):

262
src/undetected_chromedriver/_compat.py
View File

@ -1,262 +0,0 @@
#!/usr/bin/env python3
# this module is part of undetected_chromedriver
"""
888 888 d8b
888 888 Y8P
888 888
.d8888b 88888b. 888d888 .d88b. 88888b.d88b. .d88b. .d88888 888d888 888 888 888 .d88b. 888d888
d88P" 888 "88b 888P" d88""88b 888 "888 "88b d8P Y8b d88" 888 888P" 888 888 888 d8P Y8b 888P"
888 888 888 888 888 888 888 888 888 88888888 888 888 888 888 Y88 88P 88888888 888
Y88b. 888 888 888 Y88..88P 888 888 888 Y8b. Y88b 888 888 888 Y8bd8P Y8b. 888
"Y8888P 888 888 888 "Y88P" 888 888 888 "Y8888 "Y88888 888 888 Y88P "Y8888 888 88888888
by UltrafunkAmsterdam (https://github.com/ultrafunkamsterdam)
"""
from distutils.version import LooseVersion
import io
import logging
import os
import random
import re
import string
import sys
from urllib.request import urlopen
from urllib.request import urlretrieve
import zipfile
from selenium.webdriver import Chrome as _Chrome
from selenium.webdriver import ChromeOptions as _ChromeOptions
TARGET_VERSION = 0
logger = logging.getLogger("uc")
class Chrome:
def __new__(cls, *args, emulate_touch=False, **kwargs):
if not ChromeDriverManager.installed:
ChromeDriverManager(*args, **kwargs).install()
if not ChromeDriverManager.selenium_patched:
ChromeDriverManager(*args, **kwargs).patch_selenium_webdriver()
if not kwargs.get("executable_path"):
kwargs["executable_path"] = "./{}".format(
ChromeDriverManager(*args, **kwargs).executable_path
)
if not kwargs.get("options"):
kwargs["options"] = ChromeOptions()
instance = object.__new__(_Chrome)
instance.__init__(*args, **kwargs)
instance._orig_get = instance.get
def _get_wrapped(*args, **kwargs):
if instance.execute_script("return navigator.webdriver"):
instance.execute_cdp_cmd(
"Page.addScriptToEvaluateOnNewDocument",
{
"source": """
Object.defineProperty(window, 'navigator', {
value: new Proxy(navigator, {
has: (target, key) => (key === 'webdriver' ? false : key in target),
get: (target, key) =>
key === 'webdriver'
? undefined
: typeof target[key] === 'function'
? target[key].bind(target)
: target[key]
})
});
"""
},
)
return instance._orig_get(*args, **kwargs)
instance.get = _get_wrapped
instance.get = _get_wrapped
instance.get = _get_wrapped
original_user_agent_string = instance.execute_script(
"return navigator.userAgent"
)
instance.execute_cdp_cmd(
"Network.setUserAgentOverride",
{
"userAgent": original_user_agent_string.replace("Headless", ""),
},
)
if emulate_touch:
instance.execute_cdp_cmd(
"Page.addScriptToEvaluateOnNewDocument",
{
"source": """
Object.defineProperty(navigator, 'maxTouchPoints', {
get: () => 1
})"""
},
)
logger.info(f"starting undetected_chromedriver.Chrome({args}, {kwargs})")
return instance
class ChromeOptions:
def __new__(cls, *args, **kwargs):
if not ChromeDriverManager.installed:
ChromeDriverManager(*args, **kwargs).install()
if not ChromeDriverManager.selenium_patched:
ChromeDriverManager(*args, **kwargs).patch_selenium_webdriver()
instance = object.__new__(_ChromeOptions)
instance.__init__()
instance.add_argument("start-maximized")
instance.add_experimental_option("excludeSwitches", ["enable-automation"])
instance.add_argument("--disable-blink-features=AutomationControlled")
return instance
class ChromeDriverManager(object):
installed = False
selenium_patched = False
target_version = None
DL_BASE = "https://chromedriver.storage.googleapis.com/"
def __init__(self, executable_path=None, target_version=None, *args, **kwargs):
_platform = sys.platform
if TARGET_VERSION:
# use global if set
self.target_version = TARGET_VERSION
if target_version:
# use explicitly passed target
self.target_version = target_version # user override
if not self.target_version:
# none of the above (default) and just get current version
self.target_version = self.get_release_version_number().version[
0
] # only major version int
self._base = base_ = "chromedriver{}"
exe_name = self._base
if _platform in ("win32",):
exe_name = base_.format(".exe")
if _platform in ("linux",):
_platform += "64"
exe_name = exe_name.format("")
if _platform in ("darwin",):
_platform = "mac64"
exe_name = exe_name.format("")
self.platform = _platform
self.executable_path = executable_path or exe_name
self._exe_name = exe_name
def patch_selenium_webdriver(self_):
"""
Patches selenium package Chrome, ChromeOptions classes for current session
:return:
"""
import selenium.webdriver.chrome.service
import selenium.webdriver
selenium.webdriver.Chrome = Chrome
selenium.webdriver.ChromeOptions = ChromeOptions
logger.info("Selenium patched. Safe to import Chrome / ChromeOptions")
self_.__class__.selenium_patched = True
def install(self, patch_selenium=True):
"""
Initialize the patch
This will:
download chromedriver if not present
patch the downloaded chromedriver
patch selenium package if <patch_selenium> is True (default)
:param patch_selenium: patch selenium webdriver classes for Chrome and ChromeDriver (for current python session)
:return:
"""
if not os.path.exists(self.executable_path):
self.fetch_chromedriver()
if not self.__class__.installed:
if self.patch_binary():
self.__class__.installed = True
if patch_selenium:
self.patch_selenium_webdriver()
def get_release_version_number(self):
"""
Gets the latest major version available, or the latest major version of self.target_version if set explicitly.
:return: version string
"""
path = (
"LATEST_RELEASE"
if not self.target_version
else f"LATEST_RELEASE_{self.target_version}"
)
return LooseVersion(urlopen(self.__class__.DL_BASE + path).read().decode())
def fetch_chromedriver(self):
"""
Downloads ChromeDriver from source and unpacks the executable
:return: on success, name of the unpacked executable
"""
base_ = self._base
zip_name = base_.format(".zip")
ver = self.get_release_version_number().vstring
if os.path.exists(self.executable_path):
return self.executable_path
urlretrieve(
f"{self.__class__.DL_BASE}{ver}/{base_.format(f'_{self.platform}')}.zip",
filename=zip_name,
)
with zipfile.ZipFile(zip_name) as zf:
zf.extract(self._exe_name)
os.remove(zip_name)
if sys.platform != "win32":
os.chmod(self._exe_name, 0o755)
return self._exe_name
@staticmethod
def random_cdc():
cdc = random.choices(string.ascii_lowercase, k=26)
cdc[-6:-4] = map(str.upper, cdc[-6:-4])
cdc[2] = cdc[0]
cdc[3] = "_"
return "".join(cdc).encode()
def patch_binary(self):
"""
Patches the ChromeDriver binary
:return: False on failure, binary name on success
"""
linect = 0
replacement = self.random_cdc()
with io.open(self.executable_path, "r+b") as fh:
for line in iter(lambda: fh.readline(), b""):
if b"cdc_" in line:
fh.seek(-len(line), 1)
newline = re.sub(b"cdc_.{22}", replacement, line)
fh.write(newline)
linect += 1
return linect
def install(executable_path=None, target_version=None, *args, **kwargs):
ChromeDriverManager(executable_path, target_version, *args, **kwargs).install()

1
src/undetected_chromedriver/dprocess.py
View File

@ -46,7 +46,6 @@ def start_detached(executable, *args):
def _start_detached(executable, *args, writer: multiprocessing.Pipe = None): def _start_detached(executable, *args, writer: multiprocessing.Pipe = None):
# configure launch # configure launch
kwargs = {} kwargs = {}
if platform.system() == "Windows": if platform.system() == "Windows":

1
src/undetected_chromedriver/options.py
View File

@ -56,7 +56,6 @@ class ChromeOptions(_ChromiumOptions):
def handle_prefs(self, user_data_dir): def handle_prefs(self, user_data_dir):
prefs = self.experimental_options.get("prefs") prefs = self.experimental_options.get("prefs")
if prefs: if prefs:
user_data_dir = user_data_dir or self._user_data_dir user_data_dir = user_data_dir or self._user_data_dir
default_path = os.path.join(user_data_dir, "Default") default_path = os.path.join(user_data_dir, "Default")
os.makedirs(default_path, exist_ok=True) os.makedirs(default_path, exist_ok=True)

74
src/undetected_chromedriver/patcher.py
View File

@ -7,7 +7,6 @@ import logging
import os import os
import random import random
import re import re
import secrets
import string import string
import sys import sys
import time import time
@ -41,7 +40,7 @@ class Patcher(object):
d = "~/appdata/roaming/undetected_chromedriver" d = "~/appdata/roaming/undetected_chromedriver"
elif "LAMBDA_TASK_ROOT" in os.environ: elif "LAMBDA_TASK_ROOT" in os.environ:
d = "/tmp/undetected_chromedriver" d = "/tmp/undetected_chromedriver"
elif platform.startswith(("linux","linux2")): elif platform.startswith(("linux", "linux2")):
d = "~/.local/share/undetected_chromedriver" d = "~/.local/share/undetected_chromedriver"
elif platform.endswith("darwin"): elif platform.endswith("darwin"):
d = "~/Library/Application Support/undetected_chromedriver" d = "~/Library/Application Support/undetected_chromedriver"
@ -51,7 +50,6 @@ class Patcher(object):
def __init__(self, executable_path=None, force=False, version_main: int = 0): def __init__(self, executable_path=None, force=False, version_main: int = 0):
""" """
Args: Args:
executable_path: None = automatic executable_path: None = automatic
a full file path to the chromedriver executable a full file path to the chromedriver executable
@ -60,10 +58,9 @@ class Patcher(object):
version_main: 0 = auto version_main: 0 = auto
specify main chrome version (rounded, ex: 82) specify main chrome version (rounded, ex: 82)
""" """
self.force = force self.force = force
self.executable_path = None self._custom_exe_path = False
prefix = secrets.token_hex(8) prefix = "undetected"
if not os.path.exists(self.data_path): if not os.path.exists(self.data_path):
os.makedirs(self.data_path, exist_ok=True) os.makedirs(self.data_path, exist_ok=True)
@ -85,8 +82,6 @@ class Patcher(object):
os.path.join(".", self.executable_path) os.path.join(".", self.executable_path)
) )
self._custom_exe_path = False
if executable_path: if executable_path:
self._custom_exe_path = True self._custom_exe_path = True
self.executable_path = executable_path self.executable_path = executable_path
@ -94,7 +89,6 @@ class Patcher(object):
self.version_full = None self.version_full = None
def auto(self, executable_path=None, force=False, version_main=None): def auto(self, executable_path=None, force=False, version_main=None):
""""""
if executable_path: if executable_path:
self.executable_path = executable_path self.executable_path = executable_path
self._custom_exe_path = True self._custom_exe_path = True
@ -206,43 +200,46 @@ class Patcher(object):
@staticmethod @staticmethod
def gen_random_cdc(): def gen_random_cdc():
cdc = random.choices(string.ascii_lowercase, k=26) cdc = random.choices(string.ascii_letters, k=27)
cdc[-6:-4] = map(str.upper, cdc[-6:-4])
cdc[2] = cdc[0]
cdc[3] = "_"
return "".join(cdc).encode() return "".join(cdc).encode()
def is_binary_patched(self, executable_path=None): def is_binary_patched(self, executable_path=None):
"""simple check if executable is patched.
:return: False if not patched, else True
"""
executable_path = executable_path or self.executable_path executable_path = executable_path or self.executable_path
with io.open(executable_path, "rb") as fh: try:
for line in iter(lambda: fh.readline(), b""): with io.open(executable_path, "rb") as fh:
if b"cdc_" in line: return fh.read().find(b"undetected chromedriver") != -1
return False except FileNotFoundError:
else: return False
return True
def patch_exe(self): def patch_exe(self):
""" start = time.perf_counter()
Patches the ChromeDriver binary
:return: False on failure, binary name on success
"""
logger.info("patching driver executable %s" % self.executable_path) logger.info("patching driver executable %s" % self.executable_path)
linect = 0
replacement = self.gen_random_cdc()
with io.open(self.executable_path, "r+b") as fh: with io.open(self.executable_path, "r+b") as fh:
for line in iter(lambda: fh.readline(), b""): content = fh.read()
if b"cdc_" in line: # match_injected_codeblock = re.search(rb"{window.*;}", content)
fh.seek(-len(line), 1) match_injected_codeblock = re.search(rb"\{window\.cdc.*?;\}", content)
newline = re.sub(b"cdc_.{22}", replacement, line) if match_injected_codeblock:
fh.write(newline) target_bytes = match_injected_codeblock[0]
linect += 1 new_target_bytes = (
return linect b'{console.log("undetected chromedriver 1337!")}'.ljust(
len(target_bytes), b" "
)
)
new_content = content.replace(target_bytes, new_target_bytes)
if new_content == content:
logger.warning(
"something went wrong patching the driver binary. could not find injection code block"
)
else:
logger.debug(
"found block:\n%s\nreplacing with:\n%s"
% (target_bytes, new_target_bytes)
)
fh.seek(0)
fh.write(new_content)
logger.debug(
"patching took us {:.2f} seconds".format(time.perf_counter() - start)
)
def __repr__(self): def __repr__(self):
return "{0:s}({1:s})".format( return "{0:s}({1:s})".format(
@ -251,7 +248,6 @@ class Patcher(object):
) )
def __del__(self): def __del__(self):
if self._custom_exe_path: if self._custom_exe_path:
# if the driver binary is specified by user # if the driver binary is specified by user
# we assume it is important enough to not delete it # we assume it is important enough to not delete it

201
src/undetected_chromedriver/reactor.py
View File

@ -1,102 +1,99 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
# this module is part of undetected_chromedriver # this module is part of undetected_chromedriver
import asyncio import asyncio
import json import json
import logging import logging
import threading import threading
logger = logging.getLogger(__name__)
logger = logging.getLogger(__name__)
class Reactor(threading.Thread):
def __init__(self, driver: "Chrome"): class Reactor(threading.Thread):
super().__init__() def __init__(self, driver: "Chrome"):
super().__init__()
self.driver = driver
self.loop = asyncio.new_event_loop() self.driver = driver
self.loop = asyncio.new_event_loop()
self.lock = threading.Lock()
self.event = threading.Event() self.lock = threading.Lock()
self.daemon = True self.event = threading.Event()
self.handlers = {} self.daemon = True
self.handlers = {}
def add_event_handler(self, method_name, callback: callable):
""" def add_event_handler(self, method_name, callback: callable):
"""
Parameters
---------- Parameters
event_name: str ----------
example "Network.responseReceived" event_name: str
example "Network.responseReceived"
callback: callable
callable which accepts 1 parameter: the message object dictionary callback: callable
callable which accepts 1 parameter: the message object dictionary
Returns
------- Returns
-------
"""
with self.lock: """
self.handlers[method_name.lower()] = callback with self.lock:
self.handlers[method_name.lower()] = callback
@property
def running(self): @property
return not self.event.is_set() def running(self):
return not self.event.is_set()
def run(self):
try: def run(self):
asyncio.set_event_loop(self.loop) try:
self.loop.run_until_complete(self.listen()) asyncio.set_event_loop(self.loop)
except Exception as e: self.loop.run_until_complete(self.listen())
logger.warning("Reactor.run() => %s", e) except Exception as e:
logger.warning("Reactor.run() => %s", e)
async def _wait_service_started(self):
while True: async def _wait_service_started(self):
with self.lock: while True:
if ( with self.lock:
getattr(self.driver, "service", None) if (
and getattr(self.driver.service, "process", None) getattr(self.driver, "service", None)
and self.driver.service.process.poll() and getattr(self.driver.service, "process", None)
): and self.driver.service.process.poll()
await asyncio.sleep(self.driver._delay or 0.25) ):
else: await asyncio.sleep(self.driver._delay or 0.25)
break else:
break
async def listen(self):
async def listen(self):
while self.running: while self.running:
await self._wait_service_started()
await self._wait_service_started() await asyncio.sleep(1)
await asyncio.sleep(1)
try:
try: with self.lock:
with self.lock: log_entries = self.driver.get_log("performance")
log_entries = self.driver.get_log("performance")
for entry in log_entries:
for entry in log_entries: try:
obj_serialized: str = entry.get("message")
try: obj = json.loads(obj_serialized)
message = obj.get("message")
obj_serialized: str = entry.get("message") method = message.get("method")
obj = json.loads(obj_serialized)
message = obj.get("message") if "*" in self.handlers:
method = message.get("method") await self.loop.run_in_executor(
None, self.handlers["*"], message
if "*" in self.handlers: )
await self.loop.run_in_executor( elif method.lower() in self.handlers:
None, self.handlers["*"], message await self.loop.run_in_executor(
) None, self.handlers[method.lower()], message
elif method.lower() in self.handlers: )
await self.loop.run_in_executor(
None, self.handlers[method.lower()], message # print(type(message), message)
) except Exception as e:
raise e from None
# print(type(message), message)
except Exception as e: except Exception as e:
raise e from None if "invalid session id" in str(e):
pass
except Exception as e: else:
if "invalid session id" in str(e): logging.debug("exception ignored :", e)
pass
else:
logging.debug("exception ignored :", e)

4
src/undetected_chromedriver/v2.py
View File

@ -1,4 +0,0 @@
# for backward compatibility
import sys
sys.modules[__name__] = sys.modules[__package__]

3
src/undetected_chromedriver/webelement.py
View File

@ -1,6 +1,7 @@
from typing import List
from selenium.webdriver.common.by import By from selenium.webdriver.common.by import By
import selenium.webdriver.remote.webelement import selenium.webdriver.remote.webelement
from typing import List
class WebElement(selenium.webdriver.remote.webelement.WebElement): class WebElement(selenium.webdriver.remote.webelement.WebElement):