ChromiumOptions增加existing_only()和is_existing_only

This commit is contained in:
g1879 2023-10-21 10:42:26 +08:00
parent 8e8394a889
commit 116bfe7e2f
10 changed files with 60 additions and 32 deletions

View File

@ -6,7 +6,7 @@
from time import sleep
from .chromium_driver import BrowserDriver
from .._units.browser_download_manager import BrowserDownloadManager
from .._units.download_manager import BrowserDownloadManager
class Browser(object):

View File

@ -7,7 +7,7 @@ from typing import List, Optional, Union
from .chromium_driver import BrowserDriver
from .._pages.chromium_page import ChromiumPage
from .._units.browser_download_manager import BrowserDownloadManager
from .._units.download_manager import BrowserDownloadManager
class Browser(object):

View File

@ -19,24 +19,19 @@ from .tools import port_is_using
def connect_browser(option):
"""连接或启动浏览器
:param option: ChromiumOptions对象
:return: chrome 路径和进程对象组成的元组
:return: None
"""
debugger_address = option.debugger_address.replace('localhost', '127.0.0.1').lstrip('http://').lstrip('https://')
chrome_path = option.browser_path
ip, port = debugger_address.split(':')
if ip != '127.0.0.1':
if ip != '127.0.0.1' or port_is_using(ip, port) or option.is_existing_only:
test_connect(ip, port)
return None, None
if port_is_using(ip, port):
test_connect(ip, port)
return None, None
args = get_launch_args(option)
set_prefs(option)
return
# ----------创建浏览器进程----------
args = get_launch_args(option)
set_prefs(option)
try:
debugger = _run_browser(port, chrome_path, args)
@ -63,7 +58,7 @@ def get_launch_args(opt):
result = set()
has_user_path = False
remote_allow = False
headless = False
headless = None
for i in opt.arguments:
if i.startswith(('--load-extension=', '--remote-debugging-port=')):
continue
@ -74,7 +69,14 @@ def get_launch_args(opt):
elif i.startswith('--remote-allow-origins='):
remote_allow = True
elif i.startswith('--headless'):
headless = True
if i == '--headless=false':
headless = False
continue
elif i == '--headless':
i = '--headless=new'
headless = True
else:
headless = True
result.add(i)
@ -87,7 +89,7 @@ def get_launch_args(opt):
if not remote_allow:
result.add('--remote-allow-origins=*')
if not headless and system().lower() == 'linux':
if headless is not None and system().lower() == 'linux':
from os import popen
r = popen('systemctl list-units | grep graphical.target')
if 'graphical.target' not in r.read():
@ -146,27 +148,29 @@ def set_prefs(opt):
dump(prefs_dict, f)
def test_connect(ip, port):
def test_connect(ip, port, timeout=30):
"""测试浏览器是否可用
:param ip: 浏览器ip
:param port: 浏览器端口
:param timeout: 超时时间
:return: None
"""
end_time = perf_counter() + 30
end_time = perf_counter() + timeout
while perf_counter() < end_time:
try:
tabs = requests_get(f'http://{ip}:{port}/json', timeout=10, headers={'Connection': 'close'}, proxies={'http': None, 'https': None}).json()
tabs = requests_get(f'http://{ip}:{port}/json', timeout=10, headers={'Connection': 'close'},
proxies={'http': None, 'https': None}).json()
for tab in tabs:
if tab['type'] == 'page':
return
except Exception:
sleep(.2)
if ip in ('127.0.0.1', 'localhost'):
raise BrowserConnectError(f'\n连接浏览器失败,可能原因:\n1、浏览器未启动\n2、{port}端口不是Chromium内核浏览器\n'
f'3、该浏览器未允许控制\n4、和已打开的浏览器冲突\n'
f'请尝试用ChromiumOptions指定别的端口和指定浏览器路径')
raise BrowserConnectError(f'{ip}:{port}浏览器无法链接')
raise BrowserConnectError(f'\n{ip}:{port}浏览器无法链接。\n请确认:\n1、该端口为浏览器\n'
f'2、已添加--remote-allow-origins=*和--remote-debugging-port={port}启动项\n'
f'3、用户文件夹没有和已打开的浏览器冲突\n'
f'4、如为无界面系统请使用headless模式\n'
f'可使用ChromiumOptions设置端口和用户文件夹路径')
def _run_browser(port, path: str, args) -> Popen:

View File

@ -3,13 +3,18 @@
@Author : g1879
@Contact : g1879@qq.com
"""
from typing import Union
from .._configs.chromium_options import ChromiumOptions
def connect_browser(option: ChromiumOptions) -> tuple: ...
def connect_browser(option: ChromiumOptions) -> None: ...
def get_launch_args(opt: ChromiumOptions) -> list: ...
def set_prefs(opt: ChromiumOptions) -> None: ...
def test_connect(ip: str, port: Union[int, str], timeout: float = 30) -> None: ...

View File

@ -35,6 +35,7 @@ class ChromiumOptions(object):
self._page_load_strategy = options.get('page_load_strategy', 'normal')
self._proxy = om.proxies.get('http', None)
self._system_user_path = options.get('system_user_path', False)
self._existing_only = options.get('existing_only', False)
user_path = user = False
for arg in self._arguments:
@ -71,6 +72,7 @@ class ChromiumOptions(object):
self._proxy = None
self._auto_port = False
self._system_user_path = False
self._existing_only = False
@property
def download_path(self):
@ -138,6 +140,11 @@ class ChromiumOptions(object):
"""返回是否使用系统安装的浏览器所使用的用户数据文件夹"""
return self._system_user_path
@property
def is_existing_only(self):
"""返回是否只接管现有浏览器方式"""
return self._existing_only
def set_argument(self, arg, value=None):
"""设置浏览器配置的argument属性
:param arg: 属性名
@ -242,7 +249,7 @@ class ChromiumOptions(object):
:param on_off: 开或关
:return: 当前对象
"""
on_off = 'new' if on_off else False
on_off = 'new' if on_off else 'false'
return self.set_argument('--headless', on_off)
def set_no_imgs(self, on_off=True):
@ -354,6 +361,14 @@ class ChromiumOptions(object):
self._auto_port = False
return self
def existing_only(self, on_off=True):
"""设置只接管已有浏览器,不自动启动新的
:param on_off: 是否开启自动获取端口号
:return: 当前对象
"""
self._existing_only = on_off
return self
def save(self, path=None):
"""保存设置到文件
:param path: ini文件的路径 None 保存到当前读取的配置文件传入 'default' 保存到默认ini文件
@ -380,7 +395,7 @@ class ChromiumOptions(object):
# 设置chrome_options
attrs = ('debugger_address', 'binary_location', 'arguments', 'extensions', 'user', 'page_load_strategy',
'auto_port', 'system_user_path')
'auto_port', 'system_user_path', 'is_existing_only')
for i in attrs:
om.set_item('chrome_options', i, self.__getattribute__(f'_{i}'))
# 设置代理

View File

@ -25,6 +25,7 @@ class ChromiumOptions(object):
self._prefs_to_del: list = ...
self._auto_port: bool = ...
self._system_user_path: bool = ...
self._existing_only: bool = ...
@property
def download_path(self) -> str: ...
@ -53,9 +54,6 @@ class ChromiumOptions(object):
@property
def arguments(self) -> list: ...
@debugger_address.setter
def debugger_address(self, address: str): ...
@property
def extensions(self) -> list: ...
@ -65,6 +63,9 @@ class ChromiumOptions(object):
@property
def system_user_path(self) -> bool: ...
@property
def is_existing_only(self) -> bool: ...
def set_argument(self, arg: str, value: Union[str, None, bool] = None) -> ChromiumOptions: ...
def remove_argument(self, value: str) -> ChromiumOptions: ...
@ -106,6 +107,8 @@ class ChromiumOptions(object):
def auto_port(self, on_off: bool = True) -> ChromiumOptions: ...
def existing_only(self, on_off: bool = True) -> ChromiumOptions: ...
def save(self, path: Union[str, Path] = None) -> str: ...
def save_to_default(self) -> str: ...

View File

@ -1,5 +1,5 @@
[paths]
download_path =
download_path =
[chrome_options]
debugger_address = 127.0.0.1:9222
@ -11,6 +11,7 @@ page_load_strategy = normal
user = Default
auto_port = False
system_user_path = False
is_existing_only = False
[session_options]
headers = {'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_12_6) AppleWebKit/603.3.8 (KHTML, like Gecko) Version/10.1.2 Safari/603.3.8', 'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8', 'connection': 'keep-alive', 'accept-charset': 'GB2312,utf-8;q=0.7,*;q=0.7'}
@ -21,6 +22,6 @@ page_load = 30
script = 30
[proxies]
http =
http =
https =

View File

@ -5,7 +5,7 @@
"""
from typing import Union
from .browser_download_manager import DownloadMission
from .download_manager import DownloadMission
from .._elements.chromium_element import ChromiumElement
from .._pages.chromium_base import ChromiumBase
from .._pages.chromium_frame import ChromiumFrame