mirror of
https://gitee.com/g1879/DrissionPage.git
synced 2024-12-10 04:00:23 +08:00
Compare commits
2 Commits
ab1f85d192
...
341591edc6
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
341591edc6 | ||
|
|
cd1369e33a |
@ -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('请先安装cv2,pip 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信息的类"""
|
||||
|
||||
|
||||
@ -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):
|
||||
|
||||
@ -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'、ChromiumOptions、ChromiumDriver
|
||||
:param addr_or_opts: 'ip:port'、ChromiumOptions、ChromiumDriver
|
||||
: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()
|
||||
|
||||
|
||||
@ -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: ...
|
||||
|
||||
|
||||
@ -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):
|
||||
|
||||
@ -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 = ...
|
||||
|
||||
@ -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:
|
||||
|
||||
172
DrissionPage/_units/screencast.py
Normal file
172
DrissionPage/_units/screencast.py
Normal 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('请先安装cv2,pip 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'
|
||||
46
DrissionPage/_units/screencast.pyi
Normal file
46
DrissionPage/_units/screencast.pyi
Normal 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: ...
|
||||
2
setup.py
2
setup.py
@ -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.",
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user