From 61dce186c615b1966ef60d4680bec85669ea4018 Mon Sep 17 00:00:00 2001 From: g1879 Date: Fri, 28 Jun 2024 23:34:30 +0800 Subject: [PATCH] =?UTF-8?q?ChromiumOptions=E5=92=8CBrowser=E5=A2=9E?= =?UTF-8?q?=E5=8A=A0is=5Fheadless=E5=B1=9E=E6=80=A7=EF=BC=9B=E6=8E=A5?= =?UTF-8?q?=E7=AE=A1=E6=B5=8F=E8=A7=88=E5=99=A8=E6=97=B6=E5=A6=82=E6=97=A0?= =?UTF-8?q?=E5=A4=B4=E7=8A=B6=E6=80=81=E5=92=8C=E8=AE=BE=E7=BD=AE=E4=B8=8D?= =?UTF-8?q?=E4=B8=80=E8=87=B4=EF=BC=8C=E4=BC=9A=E6=8C=89=E8=AE=BE=E7=BD=AE?= =?UTF-8?q?=E9=87=8D=E5=90=AF=E6=B5=8F=E8=A7=88=E5=99=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- DrissionPage/_base/browser.py | 56 ++++++++++++++-------- DrissionPage/_base/browser.pyi | 3 +- DrissionPage/_configs/chromium_options.py | 25 ++++++++-- DrissionPage/_configs/chromium_options.pyi | 52 ++++++++++---------- DrissionPage/_functions/browser.py | 24 ---------- 5 files changed, 87 insertions(+), 73 deletions(-) diff --git a/DrissionPage/_base/browser.py b/DrissionPage/_base/browser.py index efd965d..96f8c96 100644 --- a/DrissionPage/_base/browser.py +++ b/DrissionPage/_base/browser.py @@ -40,13 +40,12 @@ class Browser(object): :param session_options: 使用双模Tab时使用的默认Session配置，为True使用ini文件配置 """ opt = handle_options(addr_or_opts) - is_exist, browser_id = run_browser(opt) + is_headless, browser_id = run_browser(opt) if browser_id in cls._BROWSERS: - r = cls._BROWSERS[browser_id] - return r + return cls._BROWSERS[browser_id] r = object.__new__(cls) r._chromium_options = opt - r._is_exist = is_exist + r.is_headless = is_headless r.id = browser_id r.address = opt.address cls._BROWSERS[browser_id] = r @@ -63,7 +62,6 @@ class Browser(object): self._type = 'Browser' self._driver = BrowserDriver(self.id, 'browser', self.address, self) - self.version = self.run_cdp('Browser.getVersion')['product'] self._frames = {} self._drivers = {} @@ -82,9 +80,25 @@ class Browser(object): self.retry_times = self._chromium_options.retry_times self.retry_interval = self._chromium_options.retry_interval + if self.is_headless != self._chromium_options.is_headless: + self.quit(3, True) + connect_browser(self._chromium_options) + s = Session() + s.trust_env = False + ws = s.get(f'http://{self._chromium_options.address}/json/version', headers={'Connection': 'close'}) + self.id = ws.json()['webSocketDebuggerUrl'].split('/')[-1] + self._driver = BrowserDriver(self.id, 'browser', self.address, self) + ws.close() + s.close() + self._frames = {} + self._drivers = {} + self._all_drivers = {} + + self.version = self._run_cdp('Browser.getVersion')['product'] + self._process_id = None try: - r = self.run_cdp('SystemInfo.getProcessInfo') + r = self._run_cdp('SystemInfo.getProcessInfo') for i in r.get('processInfo', []): if i['type'] == 'browser': self._process_id = i['id'] @@ -92,7 +106,7 @@ class Browser(object): except: pass - self.run_cdp('Target.setDiscoverTargets', discover=True) + self._run_cdp('Target.setDiscoverTargets', discover=True) self._driver.set_callback('Target.targetDestroyed', self._onTargetDestroyed) self._driver.set_callback('Target.targetCreated', self._onTargetCreated) self._dl_mgr = DownloadManager(self) @@ -137,7 +151,7 @@ class Browser(object): self._drivers.pop(tab_id, None) self._all_drivers.pop(tab_id, None) - def run_cdp(self, cmd, **cmd_args): + def _run_cdp(self, cmd, **cmd_args): """执行Chrome DevTools Protocol语句 :param cmd: 协议项目 :param cmd_args: 参数 @@ -183,7 +197,7 @@ class Browser(object): @property def tabs_count(self): """返回标签页数量""" - j = self.run_cdp('Target.getTargets')['targetInfos'] # 不要改用get，避免卡死 + j = self._run_cdp('Target.getTargets')['targetInfos'] # 不要改用get，避免卡死 return len([i for i in j if i['type'] in ('page', 'webview') and not i['url'].startswith('devtools://')]) @property @@ -235,7 +249,7 @@ class Browser(object): """ tab = None if new_context: - tab = self.run_cdp('Target.createBrowserContext')['browserContextId'] + tab = self._run_cdp('Target.createBrowserContext')['browserContextId'] kwargs = {'url': ''} if new_window: @@ -245,7 +259,7 @@ class Browser(object): if tab: kwargs['browserContextId'] = tab - tab = self.run_cdp('Target.createTarget', **kwargs)['targetId'] + tab = self._run_cdp('Target.createTarget', **kwargs)['targetId'] while tab not in self._drivers: sleep(.1) tab = obj(self, tab) @@ -401,14 +415,14 @@ class Browser(object): :param tab_id: 标签页id :return: None """ - self.run_cdp('Target.activateTarget', targetId=tab_id) + self._run_cdp('Target.activateTarget', targetId=tab_id) def reconnect(self): """断开重连""" self._driver.stop() BrowserDriver.BROWSERS.pop(self.id) self._driver = BrowserDriver(self.id, 'browser', self.address, self) - self.run_cdp('Target.setDiscoverTargets', discover=True) + self._run_cdp('Target.setDiscoverTargets', discover=True) self._driver.set_callback('Target.targetDestroyed', self._onTargetDestroyed) self._driver.set_callback('Target.targetCreated', self._onTargetCreated) @@ -419,7 +433,7 @@ class Browser(object): :return: None """ try: - self.run_cdp('Browser.close') + self._run_cdp('Browser.close') except PageDisconnectedError: pass self._driver.stop() @@ -433,7 +447,7 @@ class Browser(object): return try: - pids = [pid['id'] for pid in self.run_cdp('SystemInfo.getProcessInfo')['processInfo']] + pids = [pid['id'] for pid in self._run_cdp('SystemInfo.getProcessInfo')['processInfo']] except: return @@ -516,18 +530,20 @@ def handle_options(addr_or_opts): def run_browser(chromium_options): """连接浏览器""" - is_exist = connect_browser(chromium_options) + connect_browser(chromium_options) try: s = Session() s.trust_env = False ws = s.get(f'http://{chromium_options.address}/json/version', headers={'Connection': 'close'}) if not ws: - raise BrowserConnectError('\n浏览器连接失败，如使用全局代理，须设置不代理127.0.0.1地址。') - browser_id = ws.json()['webSocketDebuggerUrl'].split('/')[-1] + raise BrowserConnectError('\n浏览器连接失败，请确认浏览器是否启动。') + json = ws.json() + browser_id = json['webSocketDebuggerUrl'].split('/')[-1] + is_headless = 'headless' in json['User-Agent'].lower() ws.close() s.close() except KeyError: raise BrowserConnectError('浏览器版本太旧或此浏览器不支持接管。') except: - raise BrowserConnectError('\n浏览器连接失败，如使用全局代理，须设置不代理127.0.0.1地址。') - return is_exist, browser_id + raise BrowserConnectError('\n浏览器连接失败，请确认浏览器是否启动。') + return is_headless, browser_id diff --git a/DrissionPage/_base/browser.pyi b/DrissionPage/_base/browser.pyi index 9973f82..bde7e8b 100644 --- a/DrissionPage/_base/browser.pyi +++ b/DrissionPage/_base/browser.pyi @@ -24,6 +24,7 @@ class Browser(object): version: str = ... retry_times: int = ... retry_interval: float = ... + is_headless:bool = ... _BROWSERS: dict = ... _chromium_options: ChromiumOptions = ... @@ -51,7 +52,7 @@ class Browser(object): def _get_driver(self, tab_id: str, owner=None) -> Driver: ... - def run_cdp(self, cmd, **cmd_args) -> dict: ... + def _run_cdp(self, cmd, **cmd_args) -> dict: ... @property def process_id(self) -> Optional[int]: ... diff --git a/DrissionPage/_configs/chromium_options.py b/DrissionPage/_configs/chromium_options.py index acb7295..03f1152 100644 --- a/DrissionPage/_configs/chromium_options.py +++ b/DrissionPage/_configs/chromium_options.py @@ -21,7 +21,7 @@ class ChromiumOptions(object): self._user = 'Default' self._prefs_to_del = [] self.clear_file_flags = False - self._headless = None + self._is_headless = False if read_file is False: ini_path = False @@ -47,6 +47,10 @@ class ChromiumOptions(object): self._load_mode = options.get('load_mode', 'normal') self._system_user_path = options.get('system_user_path', False) self._existing_only = options.get('existing_only', False) + for i in self._arguments: + if i.startswith('--headless'): + self._is_headless = True + break self._proxy = om.proxies.get('http', None) or om.proxies.get('https', None) @@ -164,6 +168,11 @@ class ChromiumOptions(object): """返回连接失败时的重试间隔（秒）""" return self._retry_interval + @property + def is_headless(self): + """返回是否无头模式""" + return self._is_headless + def set_retry(self, times=None, interval=None): """设置连接失败时的重试操作 :param times: 重试次数 @@ -184,11 +193,19 @@ class ChromiumOptions(object): """ self.remove_argument(arg) if value is not False: - if arg == '--headless' and value is None: - self._arguments.append('--headless=new') + if arg == '--headless': + if value == 'false': + self._is_headless = False + else: + if value is None: + value = 'new' + self._arguments.append(f'--headless={value}') + self._is_headless = True else: arg_str = arg if value is None else f'{arg}={value}' self._arguments.append(arg_str) + elif arg == '--headless': + self._is_headless = False return self def remove_argument(self, value): @@ -312,7 +329,7 @@ class ChromiumOptions(object): :param on_off: 开或关 :return: 当前对象 """ - on_off = 'new' if on_off else 'false' + on_off = 'new' if on_off else on_off return self.set_argument('--headless', on_off) def no_imgs(self, on_off=True): diff --git a/DrissionPage/_configs/chromium_options.pyi b/DrissionPage/_configs/chromium_options.pyi index c682a27..56fca1e 100644 --- a/DrissionPage/_configs/chromium_options.pyi +++ b/DrissionPage/_configs/chromium_options.pyi @@ -10,30 +10,31 @@ from typing import Union, Any, Literal, Optional, Tuple class ChromiumOptions(object): - def __init__(self, read_file: [bool, None] = True, ini_path: Union[str, Path] = None): - self.ini_path: str = ... - self._driver_path: str = ... - self._user_data_path: str = ... - self._download_path: str = ... - self._tmp_path: str = ... - self._arguments: list = ... - self._browser_path: str = ... - self._user: str = ... - self._load_mode: str = ... - self._timeouts: dict = ... - self._proxy: str = ... - self._address: str = ... - self._extensions: list = ... - self._prefs: dict = ... - self._flags: dict = ... - self._prefs_to_del: list = ... - self.clear_file_flags: bool = ... - self._auto_port: bool = ... - self._system_user_path: bool = ... - self._existing_only: bool = ... - self._headless: bool = ... - self._retry_times: int = ... - self._retry_interval: float = ... + ini_path: Optional[str] = ... + _driver_path: str = ... + _user_data_path: Optional[str] = ... + _download_path: str = ... + _tmp_path: str = ... + _arguments: list = ... + _browser_path: str = ... + _user: str = ... + _load_mode: str = ... + _timeouts: dict = ... + _proxy: str = ... + _address: str = ... + _extensions: list = ... + _prefs: dict = ... + _flags: dict = ... + _prefs_to_del: list = ... + clear_file_flags: bool = ... + _auto_port: bool = ... + _system_user_path: bool = ... + _existing_only: bool = ... + _retry_times: int = ... + _retry_interval: float = ... + _is_headless: bool = ... + + def __init__(self, read_file: [bool, None] = True, ini_path: Union[str, Path] = None): ... @property def download_path(self) -> str: ... @@ -89,6 +90,9 @@ class ChromiumOptions(object): @property def retry_interval(self) -> float: ... + @property + def is_headless(self) -> bool: ... + def set_retry(self, times: int = None, interval: float = None) -> ChromiumOptions: ... def set_argument(self, arg: str, value: Union[str, None, bool] = None) -> ChromiumOptions: ... diff --git a/DrissionPage/_functions/browser.py b/DrissionPage/_functions/browser.py index a8f79d4..4211748 100644 --- a/DrissionPage/_functions/browser.py +++ b/DrissionPage/_functions/browser.py @@ -30,11 +30,6 @@ def connect_browser(option): ip, port = address.split(':') if ip != '127.0.0.1' or port_is_using(ip, port) or option.is_existing_only: test_connect(ip, port) - option._headless = False - for i in option.arguments: - if i.startswith('--headless') and not i.endswith('=false'): - option._headless = True - break return True # ----------创建浏览器进程---------- @@ -65,7 +60,6 @@ def get_launch_args(opt): # ----------处理arguments----------- result = set() has_user_path = False - headless = None for i in opt.arguments: if i.startswith(('--load-extension=', '--remote-debugging-port=')): continue @@ -73,16 +67,6 @@ def get_launch_args(opt): result.add(f'--user-data-dir={Path(i[16:]).absolute()}') has_user_path = True continue - elif i.startswith('--headless'): - if i == '--headless=false': - headless = False - continue - elif i == '--headless': - i = '--headless=new' - headless = True - else: - headless = True - result.add(i) if not has_user_path and not opt.system_user_path: @@ -93,15 +77,7 @@ def get_launch_args(opt): opt.set_user_data_path(path) result.add(f'--user-data-dir={path}') - # if headless is None and system().lower() == 'linux': # 无界面Linux自动加入无头 - # from os import popen - # r = popen('systemctl list-units | grep graphical.target') - # if 'graphical.target' not in r.read(): - # headless = True - # result.add('--headless=new') - result = list(result) - opt._headless = headless # ----------处理插件extensions------------- ext = [str(Path(e).absolute()) for e in opt.extensions]