Compare commits

...

2 Commits

Author SHA1 Message Date
g1879
341591edc6 不再使用传递ChromiumDriver方式创建ChromiumPageWebPage 2023-10-26 23:15:21 +08:00
g1879
cd1369e33a 修复监听器小bug 2023-10-26 21:40:06 +08:00
10 changed files with 258 additions and 247 deletions

View File

@ -3,13 +3,11 @@
@Author : g1879
@Contact : g1879@qq.com
"""
from base64 import b64decode
from json import loads, JSONDecodeError
from os.path import sep
from pathlib import Path
from re import findall
from threading import Thread
from time import perf_counter, sleep, time
from time import perf_counter, sleep
from requests import get
@ -17,12 +15,13 @@ from .._base.base import BasePage
from .._base.chromium_driver import ChromiumDriver
from .._commons.constants import ERROR, NoneElement
from .._commons.locator import get_loc
from .._commons.tools import get_usable_path, clean_folder
from .._commons.tools import get_usable_path
from .._commons.web import location_in_viewport
from .._elements.chromium_element import ChromiumScroll, ChromiumElement, run_js, make_chromium_ele
from .._elements.session_element import make_session_ele
from .._units.action_chains import ActionChains
from .._units.network_listener import NetworkListener
from .._units.screencast import Screencast
from .._units.setter import ChromiumBaseSetter
from .._units.waiter import ChromiumBaseWaiter
from ..errors import ContextLossError, ElementLossError, AlertExistsError, CDPError, TabClosedError, \
@ -94,15 +93,12 @@ class ChromiumBase(BasePage):
self._get_document()
self._first_run = False
def _driver_init(self, tab_id, is_init=True):
def _driver_init(self, tab_id):
"""新建页面、页面刷新、切换标签页后要进行的cdp参数初始化
:param tab_id: 要跳转到的标签页id
:param is_init: 是否初始化时执行本方法用于判断是否to_tab()调用
:return: None
"""
self._is_loading = True
if is_init and hasattr(self, '_driver'):
return # ChromiumPage接收ChromiumDriver方式启动时
self._driver = ChromiumDriver(tab_id=tab_id, tab_type='page', address=self.address)
self._alert = Alert()
self._driver.set_listener('Page.javascriptDialogOpening', self._on_alert_open)
@ -1058,166 +1054,6 @@ class Timeout(object):
return str({'implicit': self.implicit, 'page_load': self.page_load, 'script': self.script})
class Screencast(object):
def __init__(self, page):
self._page = page
self._path = None
self._running = False
self._enable = False
self._mode = 'video'
@property
def set_mode(self):
"""返回用于设置录屏幕式的对象"""
return ScreencastMode(self)
def start(self, save_path=None):
"""开始录屏
:param save_path: 录屏保存位置
:return: None
"""
self.set_save_path(save_path)
if self._path is None:
raise ValueError('save_path必须设置。')
clean_folder(self._path)
if self._mode.startswith('frugal'):
self._page.driver.set_listener('Page.screencastFrame', self._onScreencastFrame)
self._page.run_cdp('Page.startScreencast', everyNthFrame=1, quality=100)
elif not self._mode.startswith('js'):
self._running = True
self._enable = True
Thread(target=self._run).start()
else:
js = '''
async function () {
stream = await navigator.mediaDevices.getDisplayMedia({video: true, audio: true})
mime = MediaRecorder.isTypeSupported("video/webm; codecs=vp9")
? "video/webm; codecs=vp9"
: "video/webm"
mediaRecorder = new MediaRecorder(stream, {mimeType: mime})
DrissionPage_Screencast_chunks = []
mediaRecorder.addEventListener('dataavailable', function(e) {
DrissionPage_Screencast_blob_ok = false;
DrissionPage_Screencast_chunks.push(e.data);
DrissionPage_Screencast_blob_ok = true;
})
mediaRecorder.start()
mediaRecorder.addEventListener('stop', function(){
while(DrissionPage_Screencast_blob_ok==false){}
DrissionPage_Screencast_blob = new Blob(DrissionPage_Screencast_chunks,
{type: DrissionPage_Screencast_chunks[0].type});
})
}
'''
print('请手动选择要录制的目标。')
self._page.run_js('var DrissionPage_Screencast_blob;var DrissionPage_Screencast_blob_ok=false;')
self._page.run_js(js)
def stop(self, video_name=None):
"""停止录屏
:param video_name: 视频文件名为None时以当前时间名命
:return: 文件路径
"""
if video_name and not video_name.endswith('mp4'):
video_name = f'{video_name}.mp4'
name = f'{time()}.mp4' if not video_name else video_name
path = f'{self._path}{sep}{name}'
if self._mode.startswith('js'):
self._page.run_js('mediaRecorder.stop();', as_expr=True)
while not self._page.run_js('return DrissionPage_Screencast_blob_ok;'):
sleep(.1)
blob = self._page.run_js('return DrissionPage_Screencast_blob;')
uuid = self._page.run_cdp('IO.resolveBlob', objectId=blob['result']['objectId'])['uuid']
data = self._page.run_cdp('IO.read', handle=f'blob:{uuid}')['data']
with open(path, 'wb') as f:
f.write(b64decode(data))
return path
if self._mode.startswith('frugal'):
self._page.driver.set_listener('Page.screencastFrame', None)
self._page.run_cdp('Page.stopScreencast')
else:
self._enable = False
while self._running:
sleep(.1)
if self._mode.endswith('imgs'):
return str(Path(self._path).absolute())
if not str(video_name).isascii() or not str(self._path).isascii():
raise TypeError('转换成视频仅支持英文路径和文件名。')
try:
from cv2 import VideoWriter, imread, VideoWriter_fourcc
from numpy import fromfile, uint8
except ModuleNotFoundError:
raise ModuleNotFoundError('请先安装cv2pip install opencv-python')
pic_list = Path(self._path).glob('*.jpg')
img = imread(str(next(pic_list)))
imgInfo = img.shape
size = (imgInfo[1], imgInfo[0])
videoWrite = VideoWriter(path, VideoWriter_fourcc(*"mp4v"), 5, size)
for i in pic_list:
img = imread(str(i))
videoWrite.write(img)
clean_folder(self._path, ignore=(name,))
return f'{self._path}{sep}{name}'
def set_save_path(self, save_path=None):
"""设置保存路径
:param save_path: 保存路径
:return: None
"""
if save_path:
save_path = Path(save_path)
if save_path.exists() and save_path.is_file():
raise TypeError('save_path必须指定文件夹。')
save_path.mkdir(parents=True, exist_ok=True)
self._path = save_path
def _run(self):
"""非节俭模式运行方法"""
self._running = True
while self._enable:
self._page.get_screenshot(path=self._path, name=f'{time()}.jpg')
sleep(.04)
self._running = False
def _onScreencastFrame(self, **kwargs):
"""节俭模式运行方法"""
with open(f'{self._path}\\{kwargs["metadata"]["timestamp"]}.jpg', 'wb') as f:
f.write(b64decode(kwargs['data']))
self._page.run_cdp('Page.screencastFrameAck', sessionId=kwargs['sessionId'])
class ScreencastMode(object):
def __init__(self, screencast):
self._screencast = screencast
def video_mode(self):
self._screencast._mode = 'video'
def frugal_video_mode(self):
self._screencast._mode = 'frugal_video'
def js_video_mode(self):
self._screencast._mode = 'js_video'
def frugal_imgs_mode(self):
self._screencast._mode = 'frugal_imgs'
def imgs_mode(self):
self._screencast._mode = 'imgs'
class Alert(object):
"""用于保存alert信息的类"""

View File

@ -18,6 +18,7 @@ from .._pages.chromium_frame import ChromiumFrame
from .._pages.chromium_page import ChromiumPage
from .._units.action_chains import ActionChains
from .._units.network_listener import NetworkListener
from .._units.screencast import Screencast
from .._units.setter import ChromiumBaseSetter
from .._units.waiter import ChromiumBaseWaiter
@ -52,7 +53,7 @@ class ChromiumBase(BasePage):
def _connect_browser(self, tab_id: str = None) -> None: ...
def _driver_init(self, tab_id: str, is_init: bool = True) -> None: ...
def _driver_init(self, tab_id: str) -> None: ...
def _get_document(self) -> None: ...
@ -262,43 +263,6 @@ class Timeout(object):
self.script: float = ...
class Screencast(object):
def __init__(self, page: ChromiumBase):
self._page: ChromiumBase = ...
self._path: Path = ...
self._running: bool = ...
self._enable: bool = ...
self._mode: str = ...
@property
def set_mode(self) -> ScreencastMode: ...
def start(self, save_path: Union[str, Path] = None) -> None: ...
def stop(self, video_name: str = None) -> str: ...
def set_save_path(self, save_path: Union[str, Path] = None) -> None: ...
def _run(self) -> None: ...
def _onScreencastFrame(self, **kwargs) -> None: ...
class ScreencastMode(object):
def __init__(self, screencast: Screencast):
self._screencast: Screencast = ...
def video_mode(self) -> None: ...
def frugal_video_mode(self) -> None: ...
def js_video_mode(self) -> None: ...
def frugal_imgs_mode(self) -> None: ...
def imgs_mode(self) -> None: ...
class Alert(object):
def __init__(self):

View File

@ -9,7 +9,6 @@ from time import sleep
from requests import get
from .._base.browser import Browser
from .._base.chromium_driver import ChromiumDriver
from .._commons.browser import connect_browser
from .._configs.chromium_options import ChromiumOptions
from .._pages.chromium_base import ChromiumBase, Timeout
@ -23,42 +22,39 @@ from ..errors import BrowserConnectError
class ChromiumPage(ChromiumBase):
"""用于管理浏览器的类"""
def __init__(self, addr_driver_opts=None, tab_id=None, timeout=None):
def __init__(self, addr_or_opts=None, tab_id=None, timeout=None, addr_driver_opts=None):
"""
:param addr_driver_opts: 浏览器地址:端口或ChromiumOptions对象
:param addr_or_opts: 浏览器地址:端口或ChromiumOptions对象
:param tab_id: 要控制的标签页id不指定默认为激活的
:param timeout: 超时时间
"""
if not addr_or_opts and addr_driver_opts:
addr_or_opts = addr_driver_opts
self._page = self
address = self._handle_options(addr_driver_opts)
address = self._handle_options(addr_or_opts)
self._run_browser()
super().__init__(address, tab_id)
self.set.timeouts(implicit=timeout)
self._page_init()
def _handle_options(self, addr_driver_opts):
def _handle_options(self, addr_or_opts):
"""设置浏览器启动属性
:param addr_driver_opts: 'ip:port'ChromiumOptionsChromiumDriver
:param addr_or_opts: 'ip:port'ChromiumOptionsChromiumDriver
:return: 返回浏览器地址
"""
if not addr_driver_opts:
self._driver_options = ChromiumOptions(addr_driver_opts)
if not addr_or_opts:
self._driver_options = ChromiumOptions(addr_or_opts)
elif isinstance(addr_driver_opts, ChromiumOptions):
self._driver_options = addr_driver_opts
elif isinstance(addr_or_opts, ChromiumOptions):
self._driver_options = addr_or_opts
# 接收浏览器地址和端口
elif isinstance(addr_driver_opts, str):
elif isinstance(addr_or_opts, str):
self._driver_options = ChromiumOptions()
self._driver_options.set_debugger_address(addr_driver_opts)
elif isinstance(addr_driver_opts, ChromiumDriver):
self._driver_options = ChromiumOptions(False)
self._driver_options.set_debugger_address(addr_driver_opts.address)
self._driver = addr_driver_opts
self._driver_options.set_debugger_address(addr_or_opts)
else:
raise TypeError('只能接收ip:port格式、ChromiumOptions或ChromiumDriver类型参数。')
raise TypeError('只能接收ip:port格式或ChromiumOptions类型参数。')
return self._driver_options.debugger_address
@ -229,7 +225,7 @@ class ChromiumPage(ChromiumBase):
return
self.driver.stop()
self._driver_init(tab_id, False)
self._driver_init(tab_id)
if read_doc and self.ready_state in ('complete', None):
self._get_document()

View File

@ -6,7 +6,6 @@
from typing import Union, Tuple, List, Optional
from .._base.browser import Browser
from .._base.chromium_driver import ChromiumDriver
from .._configs.chromium_options import ChromiumOptions
from .._pages.chromium_base import ChromiumBase
from .._pages.chromium_tab import ChromiumTab
@ -18,7 +17,7 @@ from .._units.waiter import ChromiumPageWaiter
class ChromiumPage(ChromiumBase):
def __init__(self,
addr_driver_opts: Union[str, int, ChromiumOptions, ChromiumDriver] = None,
addr_or_opts: Union[str, int, ChromiumOptions] = None,
tab_id: str = None,
timeout: float = None):
self._driver_options: ChromiumOptions = ...
@ -26,7 +25,7 @@ class ChromiumPage(ChromiumBase):
self._browser: Browser = ...
self._rect: Optional[ChromiumTabRect] = ...
def _handle_options(self, addr_driver_opts: Union[str, ChromiumDriver, ChromiumOptions]) -> str: ...
def _handle_options(self, addr_or_opts: Union[str, ChromiumOptions]) -> str: ...
def _run_browser(self) -> None: ...

View File

@ -15,13 +15,15 @@ from .._units.setter import WebPageSetter
class WebPage(SessionPage, ChromiumPage, BasePage):
"""整合浏览器和request的页面类"""
def __init__(self, mode='d', timeout=None, driver_or_options=None, session_or_options=None):
def __init__(self, mode='d', timeout=None, driver_options=None, session_or_options=None, driver_or_options=None):
"""初始化函数
:param mode: 'd' 's'即driver模式和session模式
:param timeout: 超时时间d模式时为寻找元素时间s模式时为连接时间默认10秒
:param driver_or_options: ChromiumDriver对象只使用s模式时应传入False
:param driver_options: ChromiumDriver对象只使用s模式时应传入False
:param session_or_options: Session对象或SessionOptions对象只使用d模式时应传入False
"""
if not driver_options and driver_or_options:
driver_options = driver_or_options
self._mode = mode.lower()
if self._mode not in ('s', 'd'):
raise ValueError('mode参数只能是s或d。')
@ -29,10 +31,10 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
self._has_session = True
super().__init__(session_or_options=session_or_options)
if not driver_or_options:
driver_or_options = ChromiumOptions(read_file=driver_or_options)
driver_or_options.set_timeouts(implicit=self._timeout).set_paths(download_path=self.download_path)
super(SessionPage, self).__init__(addr_driver_opts=driver_or_options, timeout=timeout)
if not driver_options:
driver_options = ChromiumOptions(read_file=driver_options)
driver_options.set_timeouts(implicit=self._timeout).set_paths(download_path=self.download_path)
super(SessionPage, self).__init__(addr_or_opts=driver_options, timeout=timeout)
self.change_mode(self._mode, go=False, copy_cookies=False)
def __call__(self, loc_or_str, timeout=None):

View File

@ -25,7 +25,7 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
def __init__(self,
mode: str = 'd',
timeout: float = None,
driver_or_options: Union[ChromiumDriver, ChromiumOptions, bool] = None,
driver_options: Union[ChromiumOptions, bool] = None,
session_or_options: Union[Session, SessionOptions, bool] = None) -> None:
self._mode: str = ...
self._has_driver: bool = ...

View File

@ -12,7 +12,6 @@ from time import perf_counter, sleep
from requests.structures import CaseInsensitiveDict
from .._base.chromium_driver import ChromiumDriver
from ..errors import CDPError
class NetworkListener(object):
@ -218,16 +217,13 @@ class NetworkListener(object):
request_id = kwargs['requestId']
dp = self._request_ids.get(request_id)
if dp:
try:
r = self._driver.call_method('Network.getResponseBody', requestId=request_id)
body = r['body']
is_base64 = r['base64Encoded']
except CDPError:
body = ''
is_base64 = False
dp._raw_body = body
dp._base64_body = is_base64
r = self._driver.call_method('Network.getResponseBody', requestId=request_id)
if 'body' in r:
dp._raw_body = r['body']
dp._base64_body = r['base64Encoded']
else:
dp._raw_body = ''
dp._base64_body = False
self._caught.put(dp)
try:

View File

@ -0,0 +1,172 @@
# -*- coding:utf-8 -*-
"""
@Author : g1879
@Contact : g1879@qq.com
"""
from base64 import b64decode
from os.path import sep
from pathlib import Path
from threading import Thread
from time import sleep, time
from .._commons.tools import clean_folder
class Screencast(object):
def __init__(self, page):
self._page = page
self._path = None
self._running = False
self._enable = False
self._mode = 'video'
@property
def set_mode(self):
"""返回用于设置录屏幕式的对象"""
return ScreencastMode(self)
def start(self, save_path=None):
"""开始录屏
:param save_path: 录屏保存位置
:return: None
"""
self.set_save_path(save_path)
if self._path is None:
raise ValueError('save_path必须设置。')
clean_folder(self._path)
if self._mode.startswith('frugal'):
self._page.driver.set_listener('Page.screencastFrame', self._onScreencastFrame)
self._page.run_cdp('Page.startScreencast', everyNthFrame=1, quality=100)
elif not self._mode.startswith('js'):
self._running = True
self._enable = True
Thread(target=self._run).start()
else:
js = '''
async function () {
stream = await navigator.mediaDevices.getDisplayMedia({video: true, audio: true})
mime = MediaRecorder.isTypeSupported("video/webm; codecs=vp9")
? "video/webm; codecs=vp9"
: "video/webm"
mediaRecorder = new MediaRecorder(stream, {mimeType: mime})
DrissionPage_Screencast_chunks = []
mediaRecorder.addEventListener('dataavailable', function(e) {
DrissionPage_Screencast_blob_ok = false;
DrissionPage_Screencast_chunks.push(e.data);
DrissionPage_Screencast_blob_ok = true;
})
mediaRecorder.start()
mediaRecorder.addEventListener('stop', function(){
while(DrissionPage_Screencast_blob_ok==false){}
DrissionPage_Screencast_blob = new Blob(DrissionPage_Screencast_chunks,
{type: DrissionPage_Screencast_chunks[0].type});
})
}
'''
print('请手动选择要录制的目标。')
self._page.run_js('var DrissionPage_Screencast_blob;var DrissionPage_Screencast_blob_ok=false;')
self._page.run_js(js)
def stop(self, video_name=None):
"""停止录屏
:param video_name: 视频文件名为None时以当前时间名命
:return: 文件路径
"""
if video_name and not video_name.endswith('mp4'):
video_name = f'{video_name}.mp4'
name = f'{time()}.mp4' if not video_name else video_name
path = f'{self._path}{sep}{name}'
if self._mode.startswith('js'):
self._page.run_js('mediaRecorder.stop();', as_expr=True)
while not self._page.run_js('return DrissionPage_Screencast_blob_ok;'):
sleep(.1)
blob = self._page.run_js('return DrissionPage_Screencast_blob;')
uuid = self._page.run_cdp('IO.resolveBlob', objectId=blob['result']['objectId'])['uuid']
data = self._page.run_cdp('IO.read', handle=f'blob:{uuid}')['data']
with open(path, 'wb') as f:
f.write(b64decode(data))
return path
if self._mode.startswith('frugal'):
self._page.driver.set_listener('Page.screencastFrame', None)
self._page.run_cdp('Page.stopScreencast')
else:
self._enable = False
while self._running:
sleep(.1)
if self._mode.endswith('imgs'):
return str(Path(self._path).absolute())
if not str(video_name).isascii() or not str(self._path).isascii():
raise TypeError('转换成视频仅支持英文路径和文件名。')
try:
from cv2 import VideoWriter, imread, VideoWriter_fourcc
from numpy import fromfile, uint8
except ModuleNotFoundError:
raise ModuleNotFoundError('请先安装cv2pip install opencv-python')
pic_list = Path(self._path).glob('*.jpg')
img = imread(str(next(pic_list)))
imgInfo = img.shape
size = (imgInfo[1], imgInfo[0])
videoWrite = VideoWriter(path, VideoWriter_fourcc(*"mp4v"), 5, size)
for i in pic_list:
img = imread(str(i))
videoWrite.write(img)
clean_folder(self._path, ignore=(name,))
return f'{self._path}{sep}{name}'
def set_save_path(self, save_path=None):
"""设置保存路径
:param save_path: 保存路径
:return: None
"""
if save_path:
save_path = Path(save_path)
if save_path.exists() and save_path.is_file():
raise TypeError('save_path必须指定文件夹。')
save_path.mkdir(parents=True, exist_ok=True)
self._path = save_path
def _run(self):
"""非节俭模式运行方法"""
self._running = True
while self._enable:
self._page.get_screenshot(path=self._path, name=f'{time()}.jpg')
sleep(.04)
self._running = False
def _onScreencastFrame(self, **kwargs):
"""节俭模式运行方法"""
with open(f'{self._path}\\{kwargs["metadata"]["timestamp"]}.jpg', 'wb') as f:
f.write(b64decode(kwargs['data']))
self._page.run_cdp('Page.screencastFrameAck', sessionId=kwargs['sessionId'])
class ScreencastMode(object):
def __init__(self, screencast):
self._screencast = screencast
def video_mode(self):
self._screencast._mode = 'video'
def frugal_video_mode(self):
self._screencast._mode = 'frugal_video'
def js_video_mode(self):
self._screencast._mode = 'js_video'
def frugal_imgs_mode(self):
self._screencast._mode = 'frugal_imgs'
def imgs_mode(self):
self._screencast._mode = 'imgs'

View File

@ -0,0 +1,46 @@
# -*- coding:utf-8 -*-
"""
@Author : g1879
@Contact : g1879@qq.com
"""
from pathlib import Path
from typing import Union
from .._pages.chromium_base import ChromiumBase
class Screencast(object):
def __init__(self, page: ChromiumBase):
self._page: ChromiumBase = ...
self._path: Path = ...
self._running: bool = ...
self._enable: bool = ...
self._mode: str = ...
@property
def set_mode(self) -> ScreencastMode: ...
def start(self, save_path: Union[str, Path] = None) -> None: ...
def stop(self, video_name: str = None) -> str: ...
def set_save_path(self, save_path: Union[str, Path] = None) -> None: ...
def _run(self) -> None: ...
def _onScreencastFrame(self, **kwargs) -> None: ...
class ScreencastMode(object):
def __init__(self, screencast: Screencast):
self._screencast: Screencast = ...
def video_mode(self) -> None: ...
def frugal_video_mode(self) -> None: ...
def js_video_mode(self) -> None: ...
def frugal_imgs_mode(self) -> None: ...
def imgs_mode(self) -> None: ...

View File

@ -6,7 +6,7 @@ with open("README.md", "r", encoding='utf-8') as fh:
setup(
name="DrissionPage",
version="3.3.0",
version="4.0.0b0",
author="g1879",
author_email="g1879@qq.com",
description="Python based web automation tool. It can control the browser and send and receive data packets.",