mirror of
https://gitee.com/g1879/DrissionPage.git
synced 2024-12-10 04:00:23 +08:00
添加Browser类，调整架构，未完成
This commit is contained in:
g1879
parent
0ff5b47a4c
commit
de6691adc2
137
DrissionPage/browser.py
Normal file
137
DrissionPage/browser.py
Normal file
@ -0,0 +1,137 @@
|
||||
# -*- coding:utf-8 -*-
|
||||
from time import sleep
|
||||
|
||||
from .browser_download_manager import BrowserDownloadManager
|
||||
from .chromium_driver import BrowserDriver
|
||||
|
||||
|
||||
class Browser(object):
|
||||
BROWSERS = {}
|
||||
|
||||
def __new__(cls, browser_id, page):
|
||||
"""
|
||||
:param browser_id: BrowserDriver对象
|
||||
:param page: ChromiumPage对象
|
||||
"""
|
||||
if browser_id in cls.BROWSERS:
|
||||
return cls.BROWSERS[browser_id]
|
||||
return object.__new__(cls)
|
||||
|
||||
def __init__(self, browser_id, page):
|
||||
"""
|
||||
:param page: BrowserDriver对象
|
||||
:param page: ChromiumPage对象
|
||||
"""
|
||||
if hasattr(self, '_created'):
|
||||
return
|
||||
self._created = True
|
||||
Browser.BROWSERS[browser_id] = self
|
||||
|
||||
self.page = page
|
||||
self.address = page.address
|
||||
self._driver = BrowserDriver(browser_id, 'browser', page.address)
|
||||
self.id = browser_id
|
||||
self._frames = {}
|
||||
|
||||
self._process_id = None
|
||||
r = self.run_cdp('SystemInfo.getProcessInfo')
|
||||
for i in r.get('processInfo', []):
|
||||
if i['type'] == 'browser':
|
||||
self._process_id = i['id']
|
||||
break
|
||||
|
||||
self._dl_mgr = BrowserDownloadManager(self)
|
||||
|
||||
self.run_cdp('Target.setDiscoverTargets')
|
||||
self._driver.set_listener('Target.targetDestroyed', self._onTargetDestroyed)
|
||||
|
||||
def _onTargetDestroyed(self, **kwargs):
|
||||
"""标签页关闭时执行"""
|
||||
tab_id = kwargs['targetId']
|
||||
self._dl_mgr.clear_tab_info(tab_id)
|
||||
for k, i in self._frames.items():
|
||||
if i == tab_id:
|
||||
self._frames.pop(k)
|
||||
|
||||
def run_cdp(self, cmd, **cmd_args):
|
||||
"""执行Chrome DevTools Protocol语句
|
||||
:param cmd: 协议项目
|
||||
:param cmd_args: 参数
|
||||
:return: 执行的结果
|
||||
"""
|
||||
return self._driver.call_method(cmd, **cmd_args)
|
||||
|
||||
@property
|
||||
def driver(self):
|
||||
return self._driver
|
||||
|
||||
@property
|
||||
def tabs_count(self):
|
||||
"""返回标签页数量"""
|
||||
return len(self.tabs)
|
||||
|
||||
@property
|
||||
def tabs(self):
|
||||
"""返回所有标签页id组成的列表"""
|
||||
j = self._driver.get(f'http://{self.address}/json').json() # 不要改用cdp
|
||||
return [i['id'] for i in j if i['type'] == 'page']
|
||||
|
||||
@property
|
||||
def process_id(self):
|
||||
"""返回浏览器进程id"""
|
||||
return self._process_id
|
||||
|
||||
def find_tabs(self, title=None, url=None, tab_type=None, single=True):
|
||||
"""查找符合条件的tab，返回它们的id组成的列表
|
||||
:param title: 要匹配title的文本
|
||||
:param url: 要匹配url的文本
|
||||
:param tab_type: tab类型，可用列表输入多个
|
||||
:param single: 是否返回首个结果的id，为False返回所有信息
|
||||
:return: tab id或tab dict
|
||||
"""
|
||||
tabs = self._driver.get(f'http://{self.address}/json').json() # 不要改用cdp
|
||||
|
||||
if isinstance(tab_type, str):
|
||||
tab_type = {tab_type}
|
||||
elif isinstance(tab_type, (list, tuple, set)):
|
||||
tab_type = set(tab_type)
|
||||
elif tab_type is not None:
|
||||
raise TypeError('tab_type只能是set、list、tuple、str、None。')
|
||||
|
||||
r = [i for i in tabs if ((title is None or title in i['title']) and (url is None or url in i['url'])
|
||||
and (tab_type is None or i['type'] in tab_type))]
|
||||
return r[0]['id'] if r and single else r
|
||||
|
||||
def close_tab(self, tab_id):
|
||||
"""关闭标签页
|
||||
:param tab_id: 标签页id
|
||||
:return: None
|
||||
"""
|
||||
self._driver.get(f'http://{self.address}/json/close/{tab_id}')
|
||||
|
||||
def activate_tab(self, tab_id):
|
||||
"""使标签页变为活动状态
|
||||
:param tab_id: 标签页id
|
||||
:return: None
|
||||
"""
|
||||
self._driver.get(f'http://{self.address}/json/activate/{tab_id}')
|
||||
|
||||
def get_window_bounds(self):
|
||||
"""返回浏览器窗口位置和大小信息"""
|
||||
return self.run_cdp('Browser.getWindowForTarget', targetId=self.id)['bounds']
|
||||
|
||||
def quit(self):
|
||||
"""关闭浏览器"""
|
||||
self.run_cdp('Browser.close')
|
||||
self.driver.stop()
|
||||
|
||||
if self.process_id:
|
||||
from os import popen
|
||||
from platform import system
|
||||
txt = f'tasklist | findstr {self.process_id}' if system().lower() == 'windows' \
|
||||
else f'ps -ef | grep {self.process_id}'
|
||||
while True:
|
||||
p = popen(txt)
|
||||
if f' {self.process_id} ' not in p.read():
|
||||
break
|
||||
sleep(.2)
|
||||
48
DrissionPage/browser.pyi
Normal file
48
DrissionPage/browser.pyi
Normal file
@ -0,0 +1,48 @@
|
||||
# -*- coding:utf-8 -*-
|
||||
from typing import List, Optional, Union
|
||||
|
||||
from .browser_download_manager import BrowserDownloadManager
|
||||
from .chromium_page import ChromiumPage
|
||||
from .chromium_driver import BrowserDriver
|
||||
|
||||
|
||||
class Browser(object):
|
||||
BROWSERS: dict = ...
|
||||
page: ChromiumPage = ...
|
||||
_driver: BrowserDriver = ...
|
||||
id: str = ...
|
||||
address: str = ...
|
||||
_frames: dict = ...
|
||||
_process_id: Optional[int] = ...
|
||||
_dl_mgr: BrowserDownloadManager = ...
|
||||
|
||||
def __new__(cls, browser_id: str, page: ChromiumPage): ...
|
||||
|
||||
def __init__(self, browser_id: str, page: ChromiumPage): ...
|
||||
|
||||
def run_cdp(self, cmd, **cmd_args) -> dict: ...
|
||||
|
||||
@property
|
||||
def driver(self) -> BrowserDriver: ...
|
||||
|
||||
@property
|
||||
def tabs_count(self) -> int: ...
|
||||
|
||||
@property
|
||||
def tabs(self) -> List[str]: ...
|
||||
|
||||
@property
|
||||
def process_id(self) -> Optional[int]: ...
|
||||
|
||||
def find_tabs(self, title: str = None, url: str = None,
|
||||
tab_type: Union[str, list, tuple] = None, single: bool = True) -> Union[str, List[str]]: ...
|
||||
|
||||
def close_tab(self, tab_id: str) -> None: ...
|
||||
|
||||
def activate_tab(self, tab_id: str) -> None: ...
|
||||
|
||||
def get_window_bounds(self) -> dict: ...
|
||||
|
||||
def _onTargetDestroyed(self, **kwargs): ...
|
||||
|
||||
def quit(self) -> None: ...
|
||||
@ -8,42 +8,26 @@ from .commons.tools import get_usable_path
|
||||
|
||||
|
||||
class BrowserDownloadManager(object):
|
||||
BROWSERS = {}
|
||||
|
||||
def __new__(cls, page):
|
||||
def __init__(self, browser):
|
||||
"""
|
||||
:param page: ChromiumPage对象
|
||||
:param browser: Browser对象
|
||||
"""
|
||||
if page.browser_driver.id in cls.BROWSERS:
|
||||
return cls.BROWSERS[page.browser_driver.id]
|
||||
return object.__new__(cls)
|
||||
|
||||
def __init__(self, page):
|
||||
"""
|
||||
:param page: ChromiumPage对象
|
||||
"""
|
||||
if hasattr(self, '_created'):
|
||||
return
|
||||
self._created = True
|
||||
|
||||
self._page = page
|
||||
self._browser = browser
|
||||
self._page = browser.page
|
||||
self._when_download_file_exists = 'rename'
|
||||
|
||||
t = TabDownloadSettings(page.tab_id)
|
||||
t.path = page.download_path
|
||||
self._tabs_settings = {page.tab_id: t} # {tab_id: TabDownloadSettings}
|
||||
t = TabDownloadSettings(self._page.tab_id)
|
||||
t.path = self._page.download_path
|
||||
self._missions = {} # {guid: DownloadMission}
|
||||
self._tabs_settings = {self._page.tab_id: t} # {tab_id: TabDownloadSettings}
|
||||
self._tab_missions = {} # {tab_id: DownloadMission}
|
||||
self._guid_and_tab = {} # 记录guid在哪个tab
|
||||
self._flags = {} # {tab_id: [bool, DownloadMission]}
|
||||
|
||||
self._page.browser_driver.set_listener('Browser.downloadProgress', self._onDownloadProgress)
|
||||
self._page.browser_driver.set_listener('Browser.downloadWillBegin', self._onDownloadWillBegin)
|
||||
self._page.browser_driver.call_method('Browser.setDownloadBehavior',
|
||||
downloadPath=self._page.download_path,
|
||||
behavior='allowAndName', eventsEnabled=True)
|
||||
|
||||
BrowserDownloadManager.BROWSERS[page.browser_driver.id] = self
|
||||
self._browser.driver.set_listener('Browser.downloadProgress', self._onDownloadProgress)
|
||||
self._browser.driver.set_listener('Browser.downloadWillBegin', self._onDownloadWillBegin)
|
||||
self._browser.run_cdp('Browser.setDownloadBehavior', downloadPath=self._page.download_path,
|
||||
behavior='allowAndName', eventsEnabled=True)
|
||||
|
||||
@property
|
||||
def missions(self):
|
||||
@ -58,9 +42,8 @@ class BrowserDownloadManager(object):
|
||||
"""
|
||||
self._tabs_settings.setdefault(tab_id, TabDownloadSettings(tab_id)).path = str(Path(path).absolute())
|
||||
if tab_id == self._page.tab_id:
|
||||
self._page.browser_driver.call_method('Browser.setDownloadBehavior',
|
||||
downloadPath=str(Path(path).absolute()),
|
||||
behavior='allowAndName', eventsEnabled=True)
|
||||
self._browser.run_cdp('Browser.setDownloadBehavior', downloadPath=str(Path(path).absolute()),
|
||||
behavior='allowAndName', eventsEnabled=True)
|
||||
|
||||
def set_rename(self, tab_id, rename):
|
||||
"""设置某个tab的重命名文件名
|
||||
@ -100,14 +83,6 @@ class BrowserDownloadManager(object):
|
||||
"""
|
||||
return self._tab_missions.get(tab_id, [])
|
||||
|
||||
def set_mission(self, tab_id, guid):
|
||||
"""绑定tab和下载任务信息
|
||||
:param tab_id: tab id
|
||||
:param guid: 下载任务id
|
||||
:return: None
|
||||
"""
|
||||
self._guid_and_tab[guid] = tab_id
|
||||
|
||||
def set_done(self, mission, state, final_path=None):
|
||||
"""设置任务结束
|
||||
:param mission: 任务对象
|
||||
@ -121,6 +96,7 @@ class BrowserDownloadManager(object):
|
||||
if mission.tab_id in self._tab_missions and mission.id in self._tab_missions[mission.tab_id]:
|
||||
self._tab_missions[mission.tab_id].remove(mission.id)
|
||||
self._missions.pop(mission.id)
|
||||
mission._is_done = True
|
||||
|
||||
def cancel(self, mission):
|
||||
"""取消任务
|
||||
@ -128,7 +104,7 @@ class BrowserDownloadManager(object):
|
||||
:return: None
|
||||
"""
|
||||
mission.state = 'canceled'
|
||||
self._page.browser_driver.call_method('Browser.cancelDownload', guid=mission.id)
|
||||
self._browser.run_cdp('Browser.cancelDownload', guid=mission.id)
|
||||
if mission.final_path:
|
||||
Path(mission.final_path).unlink(True)
|
||||
|
||||
@ -138,12 +114,22 @@ class BrowserDownloadManager(object):
|
||||
:return: None
|
||||
"""
|
||||
mission.state = 'skipped'
|
||||
self._page.browser_driver.call_method('Browser.cancelDownload', guid=mission.id)
|
||||
self._browser.run_cdp('Browser.cancelDownload', guid=mission.id)
|
||||
|
||||
def clear_tab_info(self, tab_id):
|
||||
"""当tab关闭时清除有关信息
|
||||
:param tab_id: 标签页id
|
||||
:return: None
|
||||
"""
|
||||
self._tabs_settings.pop(tab_id)
|
||||
self._tab_missions.pop(tab_id)
|
||||
self._flags.pop(tab_id)
|
||||
TabDownloadSettings.TABS.pop(tab_id)
|
||||
|
||||
def _onDownloadWillBegin(self, **kwargs):
|
||||
"""用于获取弹出新标签页触发的下载任务"""
|
||||
guid = kwargs['guid']
|
||||
tab_id = self._page._frames.get(kwargs['frameId'], self._page.tab_id)
|
||||
tab_id = self._browser._frames.get(kwargs['frameId'], self._page.tab_id)
|
||||
|
||||
settings = TabDownloadSettings(tab_id)
|
||||
if settings.rename:
|
||||
@ -249,6 +235,7 @@ class DownloadMission(object):
|
||||
self.received_bytes = 0
|
||||
self.final_path = None
|
||||
self.save_path = save_path
|
||||
self._is_done = False
|
||||
|
||||
def __repr__(self):
|
||||
return f'<DownloadMission {id(self)} {self.rate}>'
|
||||
@ -261,7 +248,7 @@ class DownloadMission(object):
|
||||
@property
|
||||
def is_done(self):
|
||||
"""返回任务是否在运行中"""
|
||||
return self.state in ('completed', 'skipped', 'canceled')
|
||||
return self._is_done
|
||||
|
||||
def cancel(self):
|
||||
"""取消该任务，如任务已完成，删除已下载的文件"""
|
||||
|
||||
@ -1,21 +1,19 @@
|
||||
from pathlib import Path
|
||||
from typing import Dict, Optional, Union
|
||||
|
||||
from chromium_page import ChromiumPage
|
||||
from .browser import Browser
|
||||
from .chromium_page import ChromiumPage
|
||||
|
||||
|
||||
class BrowserDownloadManager(object):
|
||||
BROWSERS: Dict[str, BrowserDownloadManager] = ...
|
||||
_browser: Browser = ...
|
||||
_page: ChromiumPage = ...
|
||||
_missions: Dict[str, DownloadMission] = ...
|
||||
_tab_missions: dict = ...
|
||||
_tabs_settings: Dict[str, TabDownloadSettings] = ...
|
||||
_guid_and_tab: Dict[str, str] = ...
|
||||
_flags: dict = ...
|
||||
|
||||
def __new__(cls, page: ChromiumPage): ...
|
||||
|
||||
def __init__(self, page: ChromiumPage): ...
|
||||
def __init__(self, browser: Browser): ...
|
||||
|
||||
@property
|
||||
def missions(self) -> Dict[str, DownloadMission]: ...
|
||||
@ -32,14 +30,14 @@ class BrowserDownloadManager(object):
|
||||
|
||||
def get_tab_missions(self, tab_id: str) -> list: ...
|
||||
|
||||
def set_mission(self, tab_id: str, guid: str) -> None: ...
|
||||
|
||||
def set_done(self, mission: DownloadMission, state: str, final_path: str = None) -> None: ...
|
||||
|
||||
def cancel(self, mission: DownloadMission) -> None: ...
|
||||
|
||||
def skip(self, mission: DownloadMission) -> None: ...
|
||||
|
||||
def clear_tab_info(self, tab_id: str) -> None: ...
|
||||
|
||||
def _onDownloadWillBegin(self, **kwargs) -> None: ...
|
||||
|
||||
def _onDownloadProgress(self, **kwargs) -> None: ...
|
||||
@ -68,6 +66,7 @@ class DownloadMission(object):
|
||||
received_bytes: int = ...
|
||||
final_path: Optional[str] = ...
|
||||
save_path: str = ...
|
||||
_is_done: bool = ...
|
||||
|
||||
def __init__(self, mgr: BrowserDownloadManager, tab_id: str, _id: str, path: str, name: str, url: str,
|
||||
save_path: str): ...
|
||||
|
||||
@ -11,7 +11,7 @@ from re import findall
|
||||
from threading import Thread
|
||||
from time import perf_counter, sleep, time
|
||||
|
||||
from requests import Session
|
||||
from requests import get
|
||||
|
||||
from .action_chains import ActionChains
|
||||
from .base import BasePage
|
||||
@ -79,9 +79,8 @@ class ChromiumBase(BasePage):
|
||||
"""
|
||||
self._chromium_init()
|
||||
if not tab_id:
|
||||
u = f'http://{self.address}/json'
|
||||
json = self._control_session.get(u).json()
|
||||
self._control_session.get(u, headers={'Connection': 'close'})
|
||||
json = get(f'http://{self.address}/json', headers={'Connection': 'close'}).json()
|
||||
|
||||
tab_id = [i['id'] for i in json if i['type'] == 'page']
|
||||
if not tab_id:
|
||||
raise BrowserConnectError('浏览器连接失败，可能是浏览器版本原因。')
|
||||
@ -92,9 +91,6 @@ class ChromiumBase(BasePage):
|
||||
|
||||
def _chromium_init(self):
|
||||
"""浏览器初始设置"""
|
||||
self._control_session = Session()
|
||||
self._control_session.keep_alive = False
|
||||
self._control_session.proxies = {'http': None, 'https': None}
|
||||
self._first_run = True
|
||||
self._is_reading = False
|
||||
self._upload_list = None
|
||||
@ -277,9 +273,13 @@ class ChromiumBase(BasePage):
|
||||
return self.ele(loc_or_str, timeout)
|
||||
|
||||
@property
|
||||
def browser(self):
|
||||
def main(self):
|
||||
return self._page
|
||||
|
||||
@property
|
||||
def browser(self):
|
||||
return self._browser
|
||||
|
||||
@property
|
||||
def driver(self):
|
||||
"""返回用于控制浏览器的ChromiumDriver对象"""
|
||||
|
||||
@ -7,8 +7,8 @@ from pathlib import Path
|
||||
from typing import Union, Tuple, List, Any
|
||||
|
||||
from DataRecorder import Recorder
|
||||
from requests import Session
|
||||
|
||||
from .browser import Browser
|
||||
from .action_chains import ActionChains
|
||||
from .base import BasePage
|
||||
from .chromium_driver import ChromiumDriver
|
||||
@ -27,8 +27,8 @@ class ChromiumBase(BasePage):
|
||||
address: Union[str, int],
|
||||
tab_id: str = None,
|
||||
timeout: float = None):
|
||||
self._browser: Browser = ...
|
||||
self._page: ChromiumPage = ...
|
||||
self._control_session: Session = ...
|
||||
self.address: str = ...
|
||||
self._tab_obj: ChromiumDriver = ...
|
||||
self._is_reading: bool = ...
|
||||
@ -47,10 +47,6 @@ class ChromiumBase(BasePage):
|
||||
self._screencast: Screencast = ...
|
||||
self._actions: ActionChains = ...
|
||||
self._listener: NetworkListener = ...
|
||||
# self._wait_download_flag: bool = ...
|
||||
# self._download_rename: str = ...
|
||||
# self._when_download_file_exists: str = ...
|
||||
# self._download_missions: set = ...
|
||||
|
||||
def _connect_browser(self, tab_id: str = None) -> None: ...
|
||||
|
||||
@ -88,7 +84,10 @@ class ChromiumBase(BasePage):
|
||||
timeout: float = None) -> ChromiumElement: ...
|
||||
|
||||
@property
|
||||
def browser(self) -> ChromiumPage: ...
|
||||
def main(self) -> ChromiumPage: ...
|
||||
|
||||
@property
|
||||
def browser(self) -> Browser: ...
|
||||
|
||||
@property
|
||||
def title(self) -> str: ...
|
||||
|
||||
@ -7,6 +7,7 @@ from json import dumps, loads
|
||||
from queue import Queue, Empty
|
||||
from threading import Thread, Event
|
||||
|
||||
from requests import get
|
||||
from websocket import WebSocketTimeoutException, WebSocketException, WebSocketConnectionClosedException, \
|
||||
create_connection
|
||||
|
||||
@ -223,3 +224,6 @@ class BrowserDriver(ChromiumDriver):
|
||||
|
||||
def __repr__(self):
|
||||
return f"<BrowserDriver {self.id}>"
|
||||
|
||||
def get(self, url):
|
||||
return get(url, headers={'Connection': 'close'})
|
||||
|
||||
@ -8,6 +8,8 @@ from re import search
|
||||
from threading import Thread
|
||||
from time import sleep, perf_counter
|
||||
|
||||
from requests import get
|
||||
|
||||
from .chromium_base import ChromiumBase, ChromiumPageScroll
|
||||
from .chromium_element import ChromiumElement
|
||||
from .errors import ContextLossError
|
||||
@ -24,8 +26,10 @@ class ChromiumFrame(ChromiumBase):
|
||||
page_type = str(type(page))
|
||||
if 'ChromiumPage' in page_type or 'WebPage' in page_type:
|
||||
self._page = self._target_page = self.tab = page
|
||||
self._browser = page.browser
|
||||
else: # Tab、Frame
|
||||
self._page = page.page
|
||||
self._browser = self._page.browser
|
||||
self._target_page = page
|
||||
self.tab = page.tab if 'ChromiumFrame' in page_type else page
|
||||
|
||||
@ -87,9 +91,7 @@ class ChromiumFrame(ChromiumBase):
|
||||
try:
|
||||
super()._driver_init(tab_id)
|
||||
except:
|
||||
u = f'http://{self.address}/json'
|
||||
self._control_session.get(u)
|
||||
self._control_session.get(u, headers={'Connection': 'close'})
|
||||
get(f'http://{self.address}/json', headers={'Connection': 'close'})
|
||||
super()._driver_init(tab_id)
|
||||
|
||||
def _reload(self):
|
||||
|
||||
@ -6,9 +6,11 @@
|
||||
from pathlib import Path
|
||||
from time import perf_counter, sleep
|
||||
|
||||
from .browser_download_manager import BrowserDownloadManager
|
||||
from requests import get
|
||||
|
||||
from .browser import Browser
|
||||
from .chromium_base import ChromiumBase, Timeout
|
||||
from .chromium_driver import ChromiumDriver, BrowserDriver
|
||||
from .chromium_driver import ChromiumDriver
|
||||
from .chromium_tab import ChromiumTab
|
||||
from .commons.browser import connect_browser
|
||||
from .configs.chromium_options import ChromiumOptions
|
||||
@ -27,9 +29,7 @@ class ChromiumPage(ChromiumBase):
|
||||
:param timeout: 超时时间
|
||||
"""
|
||||
self._page = self
|
||||
self._frames = {}
|
||||
super().__init__(addr_driver_opts, tab_id)
|
||||
self._dl_mgr = BrowserDownloadManager(self)
|
||||
self.set.timeouts(implicit=timeout)
|
||||
|
||||
def _set_start_options(self, addr_driver_opts, none):
|
||||
@ -79,9 +79,7 @@ class ChromiumPage(ChromiumBase):
|
||||
if not self._tab_obj: # 不是传入driver的情况
|
||||
connect_browser(self._driver_options)
|
||||
if not tab_id:
|
||||
u = f'http://{self.address}/json'
|
||||
json = self._control_session.get(u).json()
|
||||
self._control_session.get(u, headers={'Connection': 'close'})
|
||||
json = get(f'http://{self.address}/json', headers={'Connection': 'close'}).json()
|
||||
tab_id = [i['id'] for i in json if i['type'] == 'page']
|
||||
if not tab_id:
|
||||
raise BrowserConnectError('浏览器连接失败，可能是浏览器版本原因。')
|
||||
@ -95,10 +93,8 @@ class ChromiumPage(ChromiumBase):
|
||||
|
||||
def _page_init(self):
|
||||
"""浏览器相关设置"""
|
||||
u = f'http://{self.address}/json/version'
|
||||
ws = self._control_session.get(u).json()['webSocketDebuggerUrl']
|
||||
self._control_session.get(u, headers={'Connection': 'close'})
|
||||
self._browser_driver = BrowserDriver(ws.split('/')[-1], 'browser', self.address)
|
||||
ws = get(f'http://{self.address}/json/version', headers={'Connection': 'close'}).json()['webSocketDebuggerUrl']
|
||||
self._browser = Browser(ws.split('/')[-1], self)
|
||||
|
||||
self._alert = Alert()
|
||||
self._tab_obj.set_listener('Page.javascriptDialogOpening', self._on_alert_open)
|
||||
@ -107,32 +103,20 @@ class ChromiumPage(ChromiumBase):
|
||||
self._rect = None
|
||||
self._main_tab = self.tab_id
|
||||
|
||||
self._process_id = None
|
||||
r = self.browser_driver.call_method('SystemInfo.getProcessInfo')
|
||||
if 'processInfo' not in r:
|
||||
return None
|
||||
for i in r['processInfo']:
|
||||
if i['type'] == 'browser':
|
||||
self._process_id = i['id']
|
||||
break
|
||||
|
||||
@property
|
||||
def browser_driver(self):
|
||||
def browser(self):
|
||||
"""返回用于控制浏览器cdp的driver"""
|
||||
return self._browser_driver
|
||||
return self._browser
|
||||
|
||||
@property
|
||||
def tabs_count(self):
|
||||
"""返回标签页数量"""
|
||||
return len(self.tabs)
|
||||
return self.browser.tabs_count
|
||||
|
||||
@property
|
||||
def tabs(self):
|
||||
"""返回所有标签页id组成的列表"""
|
||||
u = f'http://{self.address}/json'
|
||||
j = self._control_session.get(u).json() # 不要改用cdp
|
||||
self._control_session.get(u, headers={'Connection': 'close'})
|
||||
return [i['id'] for i in j if i['type'] == 'page']
|
||||
return self.browser.tabs
|
||||
|
||||
@property
|
||||
def main_tab(self):
|
||||
@ -146,7 +130,7 @@ class ChromiumPage(ChromiumBase):
|
||||
@property
|
||||
def process_id(self):
|
||||
"""返回浏览器进程id"""
|
||||
return self._process_id
|
||||
return self.browser.process_id
|
||||
|
||||
@property
|
||||
def set(self):
|
||||
@ -183,19 +167,7 @@ class ChromiumPage(ChromiumBase):
|
||||
:param single: 是否返回首个结果的id，为False返回所有信息
|
||||
:return: tab id或tab dict
|
||||
"""
|
||||
u = f'http://{self.address}/json'
|
||||
tabs = self._control_session.get(u).json() # 不要改用cdp
|
||||
self._control_session.get(u, headers={'Connection': 'close'})
|
||||
if isinstance(tab_type, str):
|
||||
tab_type = {tab_type}
|
||||
elif isinstance(tab_type, (list, tuple, set)):
|
||||
tab_type = set(tab_type)
|
||||
elif tab_type is not None:
|
||||
raise TypeError('tab_type只能是set、list、tuple、str、None。')
|
||||
|
||||
r = [i for i in tabs if ((title is None or title in i['title']) and (url is None or url in i['url'])
|
||||
and (tab_type is None or i['type'] in tab_type))]
|
||||
return r[0]['id'] if r and single else r
|
||||
return self._browser.find_tabs(title, url, tab_type, single)
|
||||
|
||||
def _new_tab(self, url=None, switch_to=False):
|
||||
"""新建一个标签页,该标签页在最后面
|
||||
@ -265,7 +237,7 @@ class ChromiumPage(ChromiumBase):
|
||||
tab_id = self.latest_tab
|
||||
|
||||
if activate:
|
||||
self._control_session.get(f'http://{self.address}/json/activate/{tab_id}')
|
||||
self.browser.activate_tab(tab_id)
|
||||
|
||||
if tab_id == self.tab_id:
|
||||
return
|
||||
@ -305,7 +277,7 @@ class ChromiumPage(ChromiumBase):
|
||||
self.driver.stop()
|
||||
|
||||
for tab in tabs:
|
||||
self._control_session.get(f'http://{self.address}/json/close/{tab}')
|
||||
self.browser.close_tab(tab)
|
||||
while len(self.tabs) != end_len:
|
||||
sleep(.1)
|
||||
|
||||
@ -345,19 +317,7 @@ class ChromiumPage(ChromiumBase):
|
||||
|
||||
def quit(self):
|
||||
"""关闭浏览器"""
|
||||
self._tab_obj.call_method('Browser.close')
|
||||
self._tab_obj.stop()
|
||||
|
||||
if self.process_id:
|
||||
from os import popen
|
||||
from platform import system
|
||||
txt = f'tasklist | findstr {self.process_id}' if system().lower() == 'windows' \
|
||||
else f'ps -ef | grep {self.process_id}'
|
||||
while True:
|
||||
p = popen(txt)
|
||||
if f' {self.process_id} ' not in p.read():
|
||||
break
|
||||
sleep(.2)
|
||||
self.browser.quit()
|
||||
|
||||
def _on_alert_close(self, **kwargs):
|
||||
"""alert关闭时触发的方法"""
|
||||
@ -448,7 +408,7 @@ class ChromiumTabRect(object):
|
||||
|
||||
def _get_browser_rect(self):
|
||||
"""获取浏览器范围信息"""
|
||||
return self._page.browser_driver.call_method('Browser.getWindowForTarget', targetId=self._page.tab_id)['bounds']
|
||||
return self._page.browser.get_window_bounds()
|
||||
|
||||
|
||||
class Alert(object):
|
||||
|
||||
@ -3,9 +3,9 @@
|
||||
@Author : g1879
|
||||
@Contact : g1879@qq.com
|
||||
"""
|
||||
from typing import Union, Tuple, List
|
||||
from typing import Union, Tuple, List, Optional
|
||||
|
||||
from .browser_download_manager import BrowserDownloadManager
|
||||
from .browser import Browser
|
||||
from .chromium_base import ChromiumBase
|
||||
from .chromium_driver import ChromiumDriver
|
||||
from .chromium_tab import ChromiumTab
|
||||
@ -21,13 +21,10 @@ class ChromiumPage(ChromiumBase):
|
||||
tab_id: str = None,
|
||||
timeout: float = None):
|
||||
self._driver_options: ChromiumOptions = ...
|
||||
self._process_id: str = ...
|
||||
self._dl_mgr: BrowserDownloadManager = ...
|
||||
self._main_tab: str = ...
|
||||
self._alert: Alert = ...
|
||||
self._browser_driver: ChromiumDriver = ...
|
||||
self._browser: Browser = ...
|
||||
self._rect: ChromiumTabRect = ...
|
||||
self._frames: dict = ...
|
||||
|
||||
def _connect_browser(self,
|
||||
addr_driver_opts: Union[str, ChromiumDriver] = None,
|
||||
@ -38,7 +35,7 @@ class ChromiumPage(ChromiumBase):
|
||||
def _page_init(self) -> None: ...
|
||||
|
||||
@property
|
||||
def browser_driver(self) -> ChromiumDriver: ...
|
||||
def browser(self) -> Browser: ...
|
||||
|
||||
@property
|
||||
def tabs_count(self) -> int: ...
|
||||
@ -59,7 +56,7 @@ class ChromiumPage(ChromiumBase):
|
||||
def latest_tab(self) -> str: ...
|
||||
|
||||
@property
|
||||
def process_id(self) -> Union[None, int]: ...
|
||||
def process_id(self) -> Optional[int]: ...
|
||||
|
||||
@property
|
||||
def set(self) -> ChromiumPageSetter: ...
|
||||
|
||||
@ -22,6 +22,7 @@ class ChromiumTab(ChromiumBase):
|
||||
:param tab_id: 要控制的标签页id，不指定默认为激活的
|
||||
"""
|
||||
self._page = page
|
||||
self._browser = page.browser
|
||||
super().__init__(page.address, tab_id, page.timeout)
|
||||
|
||||
def _set_runtime_settings(self):
|
||||
@ -68,6 +69,7 @@ class WebPageTab(SessionPage, ChromiumTab):
|
||||
:param tab_id: 要控制的标签页id
|
||||
"""
|
||||
self._page = page
|
||||
self._browser = page.browser
|
||||
self.address = page.address
|
||||
self._debug = page._debug
|
||||
self._debug_recorder = page._debug_recorder
|
||||
|
||||
@ -7,7 +7,7 @@ from typing import Union, Tuple, Any, List
|
||||
|
||||
from requests import Session, Response
|
||||
|
||||
from waiter import ChromiumTabWaiter
|
||||
from .browser import Browser
|
||||
from .chromium_base import ChromiumBase
|
||||
from .chromium_element import ChromiumElement
|
||||
from .chromium_frame import ChromiumFrame
|
||||
@ -16,6 +16,7 @@ from .session_element import SessionElement
|
||||
from .session_page import SessionPage
|
||||
from .setter import TabSetter
|
||||
from .setter import WebPageTabSetter
|
||||
from .waiter import ChromiumTabWaiter
|
||||
from .web_page import WebPage
|
||||
|
||||
|
||||
@ -23,6 +24,7 @@ class ChromiumTab(ChromiumBase):
|
||||
|
||||
def __init__(self, page: ChromiumPage, tab_id: str = None):
|
||||
self._page: ChromiumPage = ...
|
||||
self._browser: Browser = ...
|
||||
|
||||
def _set_runtime_settings(self) -> None: ...
|
||||
|
||||
@ -44,6 +46,7 @@ class ChromiumTab(ChromiumBase):
|
||||
class WebPageTab(SessionPage, ChromiumTab):
|
||||
def __init__(self, page: WebPage, tab_id: str):
|
||||
self._page: WebPage = ...
|
||||
self._browser: Browser = ...
|
||||
self._mode: str = ...
|
||||
self._has_driver = ...
|
||||
self._has_session = ...
|
||||
|
||||
@ -155,9 +155,7 @@ def test_connect(ip, port):
|
||||
end_time = perf_counter() + 30
|
||||
while perf_counter() < end_time:
|
||||
try:
|
||||
u = f'http://{ip}:{port}/json'
|
||||
tabs = requests_get(u, timeout=10, proxies={'http': None, 'https': None}).json()
|
||||
requests_get(u, headers={'Connection': 'close'}, proxies={'http': None, 'https': None})
|
||||
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
|
||||
|
||||
@ -175,7 +175,7 @@ class ChromiumPageSetter(TabSetter):
|
||||
tab_or_id = self._page.tab_id
|
||||
elif not isinstance(tab_or_id, str): # 传入Tab对象
|
||||
tab_or_id = tab_or_id.tab_id
|
||||
self._page._control_session.get(f'http://{self._page.address}/json/activate/{tab_or_id}')
|
||||
self._page.browser.activate_tab(tab_or_id)
|
||||
|
||||
|
||||
class SessionPageSetter(object):
|
||||
|
||||
@ -237,20 +237,20 @@ class ChromiumPageWaiter(ChromiumTabWaiter):
|
||||
:return: 是否等待成功
|
||||
"""
|
||||
if not timeout:
|
||||
while self._driver._dl_mgr._missions:
|
||||
while self._driver.browser._dl_mgr._missions:
|
||||
sleep(.5)
|
||||
return True
|
||||
|
||||
else:
|
||||
end_time = perf_counter() + timeout
|
||||
while end_time > perf_counter():
|
||||
if not self._driver._dl_mgr._missions:
|
||||
if not self._driver.browser._dl_mgr._missions:
|
||||
return True
|
||||
sleep(.5)
|
||||
|
||||
if self._driver._dl_mgr._missions:
|
||||
if self._driver.browser._dl_mgr._missions:
|
||||
if cancel_if_timeout:
|
||||
for m in list(self._driver._dl_mgr._missions.values()):
|
||||
for m in list(self._driver.browser._dl_mgr._missions.values()):
|
||||
m.cancel()
|
||||
return False
|
||||
else:
|
||||
|
||||
@ -6,7 +6,6 @@
|
||||
from requests import Session
|
||||
|
||||
from .base import BasePage
|
||||
from .browser_download_manager import BrowserDownloadManager
|
||||
from .chromium_base import ChromiumBase, Timeout
|
||||
from .chromium_driver import ChromiumDriver
|
||||
from .chromium_page import ChromiumPage
|
||||
@ -53,7 +52,6 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
|
||||
self._set_runtime_settings()
|
||||
self._connect_browser()
|
||||
self._create_session()
|
||||
self._dl_mgr = BrowserDownloadManager(self)
|
||||
self.set.timeouts(implicit=timeout)
|
||||
|
||||
def _set_start_options(self, dr_opt, se_opt):
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user