Optimized some methods
This commit is contained in:
parent
06d53f913a
commit
35499a5784
367
Galactic.py
367
Galactic.py
|
@ -35,7 +35,7 @@ from PyQt5.QtNetwork import QNetworkProxyFactory
|
|||
from PyQt5.QtWidgets import QApplication, QMainWindow, QMessageBox, QSystemTrayIcon, QMenu, QAction, QDesktopWidget
|
||||
from PyQt5.QtCore import Qt, QCoreApplication, QTimer, QUrl
|
||||
from PyQt5.QtGui import QIcon
|
||||
from typing import List
|
||||
from typing import List, Tuple, Any
|
||||
from fastapi import FastAPI, Response, UploadFile, File, HTTPException
|
||||
from pydantic import BaseModel
|
||||
from starlette.responses import JSONResponse, FileResponse
|
||||
|
@ -43,9 +43,9 @@ from winotify import Notification, audio
|
|||
from func_timeout import func_set_timeout, FunctionTimedOut
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
sys.path.append(os.path.join(os.path.dirname(__file__), 'Packages'))
|
||||
|
||||
|
||||
def _fd(f):
|
||||
return f.fileno() if hasattr(f, 'fileno') else f
|
||||
|
||||
|
@ -164,18 +164,21 @@ try:
|
|||
except ValueError:
|
||||
raise SystemError('Not supported driver classes.')
|
||||
|
||||
|
||||
def notification_send(app_id=None, title='', message='', reminder=None):
|
||||
app_id = app_id() if callable(app_id) else app_id
|
||||
notify = Notification(app_id=app_id, title=title, msg=message, icon=os.path.join(os.path.dirname(__file__), 'favicon.ico'))
|
||||
notify.set_audio(audio.Reminder, bool(reminder))
|
||||
notify.show()
|
||||
|
||||
|
||||
def import_module(file: str):
|
||||
spec = importlib.util.spec_from_file_location(os.path.splitext(os.path.basename(file))[0], file)
|
||||
module_from_spec = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(module_from_spec)
|
||||
return module_from_spec
|
||||
|
||||
|
||||
class BrowserMobileEmulation(dict):
|
||||
"""
|
||||
Mobile emulation parameters.
|
||||
|
@ -489,6 +492,22 @@ class ColorUtils:
|
|||
return r, g, b
|
||||
|
||||
|
||||
class FetchResult:
|
||||
def __init__(self, data):
|
||||
self.status = 0
|
||||
self.header = {}
|
||||
self.result = ''
|
||||
data and self._resolve_data(data)
|
||||
|
||||
def _resolve_data(self, data: dict):
|
||||
self.status = data['status']
|
||||
self.header = data['header']
|
||||
self.result = data['result']
|
||||
|
||||
def __str__(self):
|
||||
return '%s: %s\n%s: %s\n%s: %s' % ('STATUS', self.status, 'HEADER', self.header, 'RESULT', self.result)
|
||||
|
||||
|
||||
class Browser(browser_webdriver):
|
||||
"""
|
||||
Browser web driver.
|
||||
|
@ -655,20 +674,38 @@ class Browser(browser_webdriver):
|
|||
|
||||
@property
|
||||
def title(self) -> str:
|
||||
self.update_tab_auto_switch()
|
||||
if (self.current_alert() is None) == 1:
|
||||
"""
|
||||
Return current title.
|
||||
"""
|
||||
self._update_tab_auto_switch()
|
||||
if (self.current_alert is None) == 1:
|
||||
return super().title
|
||||
else:
|
||||
return ''
|
||||
|
||||
@property
|
||||
def current_url(self) -> str:
|
||||
self.update_tab_auto_switch()
|
||||
if (self.current_alert() is None) == 1:
|
||||
"""
|
||||
Return current url.
|
||||
"""
|
||||
self._update_tab_auto_switch()
|
||||
if (self.current_alert is None) == 1:
|
||||
return super().current_url
|
||||
else:
|
||||
return ''
|
||||
|
||||
@property
|
||||
def current_alert(self):
|
||||
"""
|
||||
Return current alert object.
|
||||
"""
|
||||
try:
|
||||
alert = Alert(self)
|
||||
self.execute(Command.W3C_GET_ALERT_TEXT)
|
||||
return alert
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
@property
|
||||
def current_alert_text(self) -> str:
|
||||
"""
|
||||
|
@ -679,6 +716,28 @@ class Browser(browser_webdriver):
|
|||
except Exception:
|
||||
return ''
|
||||
|
||||
@property
|
||||
def window_inner_size(self) -> tuple:
|
||||
"""
|
||||
Return the page window inner size.
|
||||
"""
|
||||
size = self.execute_script('return [window.innerWidth, window.innerHeight];')
|
||||
return size[0], size[1]
|
||||
|
||||
def open(self, url=None):
|
||||
"""
|
||||
Open the URL, simulate into the URL in the address bar and jump, the new page has no Referrer.
|
||||
"""
|
||||
self._update_tab_auto_switch()
|
||||
self._update_cdp_command()
|
||||
return self.get(url)
|
||||
|
||||
def turn(self, url=None):
|
||||
"""
|
||||
Simulation "window.location.href" jumps, the new page has Referrer.
|
||||
"""
|
||||
return self.execute_script('window.location.href=%s;' % json.dumps(url, indent=None, ensure_ascii=True), None)
|
||||
|
||||
def wait(self, secs: int | float = 1):
|
||||
"""
|
||||
Will sleep waiting.
|
||||
|
@ -695,25 +754,14 @@ class Browser(browser_webdriver):
|
|||
self.quit()
|
||||
|
||||
def quit(self):
|
||||
"""
|
||||
Exit the browser.
|
||||
"""
|
||||
try:
|
||||
super().quit()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def open(self, url=None):
|
||||
"""
|
||||
Open the URL, simulate into the URL in the address bar and jump, the new page has no Referrer.
|
||||
"""
|
||||
self.update_tab_auto_switch()
|
||||
self.update_cdp_command()
|
||||
return self.get(url)
|
||||
|
||||
def turn(self, url=None):
|
||||
"""
|
||||
Simulation "window.location.href" jumps, the new page has Referrer.
|
||||
"""
|
||||
return self.execute_script('window.location.href=%s;' % json.dumps(url, indent=None, ensure_ascii=True), None)
|
||||
|
||||
def find(self, path, wait_for=False, timeout: float = 5.0, freq: float = 0.5, delay: float = 0.0) -> WebElement:
|
||||
"""
|
||||
Use XPath to find an element.
|
||||
|
@ -721,7 +769,7 @@ class Browser(browser_webdriver):
|
|||
element = self.webdriver_wait(timeout, freq).until(EC.presence_of_element_located((By.XPATH, path))) if wait_for else self.find_element(By.XPATH, path)
|
||||
delay and self.wait(delay)
|
||||
element = self.find_element(By.XPATH, path) if delay else element
|
||||
self.element_appear(element, '#F8BE5F')
|
||||
self._element_highlight(element, '#F8BE5F')
|
||||
return element
|
||||
|
||||
def find_mult(self, path) -> list:
|
||||
|
@ -729,16 +777,16 @@ class Browser(browser_webdriver):
|
|||
Use XPath to find elements.
|
||||
"""
|
||||
element = self.find_elements(By.XPATH, path)
|
||||
for this_element in element: self.element_appear(this_element, '#F8BE5F')
|
||||
for this_element in element: self._element_highlight(this_element, '#F8BE5F')
|
||||
return element
|
||||
|
||||
def find_mult_random_choice(self, path) -> WebElement:
|
||||
"""
|
||||
Use XPath to find elements then random_choice one.
|
||||
Use XPath to find elements then random choice one.
|
||||
"""
|
||||
element = self.find_elements(By.XPATH, path)
|
||||
element = random.choice(element)
|
||||
self.element_appear(element, '#F8BE5F')
|
||||
self._element_highlight(element, '#F8BE5F')
|
||||
return element
|
||||
|
||||
def find_element_by(self, sentence):
|
||||
|
@ -746,21 +794,21 @@ class Browser(browser_webdriver):
|
|||
Custom find element, pass into a tuple or list.
|
||||
"""
|
||||
element = self.find_element(*sentence)
|
||||
self.element_appear(element, '#F8BE5F')
|
||||
self._element_highlight(element, '#F8BE5F')
|
||||
return element
|
||||
|
||||
def click(self, element):
|
||||
"""
|
||||
Click element.
|
||||
"""
|
||||
self.element_appear(element, '#FF0000')
|
||||
self._element_highlight(element, '#FF0000')
|
||||
element.click()
|
||||
|
||||
def click_simu(self, element):
|
||||
def click_simulate(self, element):
|
||||
"""
|
||||
Click element for simulate.
|
||||
"""
|
||||
self.element_effect(element)
|
||||
self._element_click_effect(element)
|
||||
self.action_chains().reset_actions()
|
||||
self.action_chains().click(element).perform()
|
||||
self.wait(0.1)
|
||||
|
@ -777,7 +825,7 @@ class Browser(browser_webdriver):
|
|||
"""
|
||||
Enter the content to the element.
|
||||
"""
|
||||
self.element_appear(element, '#00B6F1')
|
||||
self._element_highlight(element, '#00B6F1')
|
||||
self.action_chains().reset_actions()
|
||||
self.action_chains().send_keys_to_element(element, content).perform()
|
||||
self.wait(0.1)
|
||||
|
@ -786,80 +834,13 @@ class Browser(browser_webdriver):
|
|||
"""
|
||||
Park the mouse here.
|
||||
"""
|
||||
self.element_appear(element, '#49DC07')
|
||||
self._element_highlight(element, '#49DC07')
|
||||
self.action_chains().reset_actions()
|
||||
self.action_chains().move_to_element(element).perform()
|
||||
|
||||
def tab_create(self, url=None):
|
||||
"""
|
||||
Create a new tab and open the URL.
|
||||
"""
|
||||
self.switch_to.new_window('tab')
|
||||
self.update_cdp_command()
|
||||
url and self.open(url)
|
||||
|
||||
def tab_switch(self, tab: int | str):
|
||||
"""
|
||||
Switch the browser tab page
|
||||
"""
|
||||
handles = self.window_handles
|
||||
lengths = len(handles)
|
||||
current = handles.index(self.current_window_handle)
|
||||
if isinstance(tab, int):
|
||||
handle = tab
|
||||
elif tab == PositionTab.Prev:
|
||||
handle = (current - 1)
|
||||
elif tab == PositionTab.Next:
|
||||
handle = (current + 1) % lengths
|
||||
else:
|
||||
handle = None
|
||||
self.switch_to.window(handles[handle])
|
||||
self.wait(0.2)
|
||||
self.update_cdp_command()
|
||||
|
||||
def tab_switch_prev(self):
|
||||
self.tab_switch(PositionTab.Prev)
|
||||
|
||||
def tab_switch_next(self):
|
||||
self.tab_switch(PositionTab.Next)
|
||||
|
||||
def tab_cancel(self):
|
||||
"""
|
||||
Close the current browser tab page.
|
||||
"""
|
||||
handles = self.window_handles
|
||||
if len(handles):
|
||||
current = handles.index(self.current_window_handle)
|
||||
self.close()
|
||||
current > 0 and self.switch_to.window(handles[current - 1])
|
||||
self.wait(0.2)
|
||||
|
||||
def tab_cancel_all(self):
|
||||
"""
|
||||
Close all the browser tab page.
|
||||
"""
|
||||
handles = self.window_handles
|
||||
for i in handles:
|
||||
self.tab_cancel()
|
||||
|
||||
def frame_switch_to(self, element_of_frame):
|
||||
"""
|
||||
Switch frame to the specified frame element.
|
||||
"""
|
||||
self.switch_to.frame(element_of_frame)
|
||||
self.wait(0.2)
|
||||
|
||||
def frame_switch_to_default(self):
|
||||
"""
|
||||
Switch to the default frame.
|
||||
"""
|
||||
self.switch_to.default_content()
|
||||
self.wait(0.2)
|
||||
|
||||
def scroll(self):
|
||||
"""
|
||||
Scroll page.
|
||||
:return:
|
||||
"""
|
||||
self.action_chains().reset_actions()
|
||||
self.action_chains().scroll_by_amount(0, self.execute_script('return document.documentElement.clientHeight;')).perform()
|
||||
|
@ -887,7 +868,108 @@ class Browser(browser_webdriver):
|
|||
self.action_chains().scroll_to_element(element).perform()
|
||||
self.wait(0.8)
|
||||
|
||||
def element_force_display(self, element):
|
||||
def fetch(self, url: str, options: dict):
|
||||
"""
|
||||
Sending http requests using fetch.
|
||||
"""
|
||||
return FetchResult(self.execute_async_script('''
|
||||
var _callback = arguments[arguments.length - 1];
|
||||
var _url = arguments[0];
|
||||
var _options = arguments[1];
|
||||
var _data = {};
|
||||
fetch(_url, _options)
|
||||
.then((response) => {
|
||||
_data.status = response.status;
|
||||
let headers = {};
|
||||
response.headers.forEach((value, name) => {
|
||||
headers[name] = value;
|
||||
});
|
||||
_data.header = headers;
|
||||
return response.text();
|
||||
})
|
||||
.then((result) => {
|
||||
_data.result = result;
|
||||
_callback(_data);
|
||||
})
|
||||
.catch((error) => {
|
||||
console.error(error);
|
||||
_callback(null);
|
||||
});
|
||||
''', url, options or {}))
|
||||
|
||||
def frame_switch_to(self, element_of_frame):
|
||||
"""
|
||||
Switch frame to the specified frame element.
|
||||
"""
|
||||
self.switch_to.frame(element_of_frame)
|
||||
self.wait(0.2)
|
||||
|
||||
def frame_switch_to_default(self):
|
||||
"""
|
||||
Switch to the default frame.
|
||||
"""
|
||||
self.switch_to.default_content()
|
||||
self.wait(0.2)
|
||||
|
||||
def tab_create(self, url=None):
|
||||
"""
|
||||
Create a new tab and open the URL.
|
||||
"""
|
||||
self.switch_to.new_window('tab')
|
||||
self._update_cdp_command()
|
||||
url and self.open(url)
|
||||
|
||||
def tab_switch(self, tab: int | str):
|
||||
"""
|
||||
Switch the browser tab page.
|
||||
"""
|
||||
handles = self.window_handles
|
||||
lengths = len(handles)
|
||||
current = handles.index(self.current_window_handle)
|
||||
if isinstance(tab, int):
|
||||
handle = tab
|
||||
elif tab == PositionTab.Prev:
|
||||
handle = (current - 1)
|
||||
elif tab == PositionTab.Next:
|
||||
handle = (current + 1) % lengths
|
||||
else:
|
||||
handle = None
|
||||
self.switch_to.window(handles[handle])
|
||||
self.wait(0.2)
|
||||
self._update_cdp_command()
|
||||
|
||||
def tab_switch_prev(self):
|
||||
"""
|
||||
Switch to the previous tab.
|
||||
"""
|
||||
self.tab_switch(PositionTab.Prev)
|
||||
|
||||
def tab_switch_next(self):
|
||||
"""
|
||||
Switch to next tab.
|
||||
"""
|
||||
self.tab_switch(PositionTab.Next)
|
||||
|
||||
def tab_cancel(self):
|
||||
"""
|
||||
Close the current browser tab page.
|
||||
"""
|
||||
handles = self.window_handles
|
||||
if len(handles):
|
||||
current = handles.index(self.current_window_handle)
|
||||
self.close()
|
||||
current > 0 and self.switch_to.window(handles[current - 1])
|
||||
self.wait(0.2)
|
||||
|
||||
def tab_cancel_all(self):
|
||||
"""
|
||||
Close all the browser tab page.
|
||||
"""
|
||||
handles = self.window_handles
|
||||
for i in handles:
|
||||
self.tab_cancel()
|
||||
|
||||
def force_display_element(self, element):
|
||||
"""
|
||||
Make hidden element visible and interactive.
|
||||
"""
|
||||
|
@ -895,7 +977,30 @@ class Browser(browser_webdriver):
|
|||
'let e=arguments[0];e.style.display="inline-block";e.style.visibility="visible";e.setAttribute("hidden","false");', element
|
||||
)
|
||||
|
||||
def element_appear(self, element=None, color='#ff0000', dura=2500):
|
||||
def screenshot(self) -> bytes:
|
||||
"""
|
||||
Screenshot as bytes.
|
||||
"""
|
||||
return self.get_screenshot_as_png()
|
||||
|
||||
def action_chains(self) -> ActionChains:
|
||||
"""
|
||||
Return ActionChains object.
|
||||
"""
|
||||
return ActionChains(self)
|
||||
|
||||
def webdriver_wait(self, timeout: float, poll_frequency: float = 0.5, ignored_exceptions=None):
|
||||
"""
|
||||
Return WebDriverWait object.
|
||||
"""
|
||||
return WebDriverWait(
|
||||
driver=self,
|
||||
timeout=timeout,
|
||||
poll_frequency=poll_frequency,
|
||||
ignored_exceptions=ignored_exceptions
|
||||
)
|
||||
|
||||
def _element_highlight(self, element=None, color='#ff0000', dura=2500):
|
||||
"""
|
||||
Make the element highlight.
|
||||
"""
|
||||
|
@ -918,7 +1023,7 @@ class Browser(browser_webdriver):
|
|||
''' % (color, r, g, b, dura), element
|
||||
)
|
||||
|
||||
def element_effect(self, element=None, x: int = 0, y: int = 0):
|
||||
def _element_click_effect(self, element=None, x: int = 0, y: int = 0):
|
||||
"""
|
||||
Make a coordinate click effect.
|
||||
"""
|
||||
|
@ -961,53 +1066,12 @@ class Browser(browser_webdriver):
|
|||
''' % (0, 0, 0, 0, 30, 30), element, x, y
|
||||
)
|
||||
|
||||
def webdriver_wait(self, timeout: float, poll_frequency: float = 0.5, ignored_exceptions=None):
|
||||
"""
|
||||
Return WebDriverWait object.
|
||||
"""
|
||||
return WebDriverWait(
|
||||
driver=self,
|
||||
timeout=timeout,
|
||||
poll_frequency=poll_frequency,
|
||||
ignored_exceptions=ignored_exceptions
|
||||
)
|
||||
|
||||
def current_alert(self):
|
||||
"""
|
||||
Return current alert object.
|
||||
"""
|
||||
try:
|
||||
alert = Alert(self)
|
||||
self.execute(Command.W3C_GET_ALERT_TEXT)
|
||||
return alert
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
def window_inner_size(self):
|
||||
"""
|
||||
Get the page window inner size.
|
||||
"""
|
||||
size = self.execute_script('return [window.innerWidth, window.innerHeight];')
|
||||
return {'w': size[0] or 0, 'h': size[1] or 0}
|
||||
|
||||
def action_chains(self):
|
||||
"""
|
||||
Return ActionChains object.
|
||||
"""
|
||||
return ActionChains(self)
|
||||
|
||||
def screenshot(self) -> bytes:
|
||||
"""
|
||||
Screenshot as bytes.
|
||||
"""
|
||||
return self.get_screenshot_as_png()
|
||||
|
||||
def update_cdp_command(self):
|
||||
def _update_cdp_command(self):
|
||||
for cmd in self.cdplist:
|
||||
self.execute_cdp_cmd(*cmd)
|
||||
|
||||
def update_tab_auto_switch(self):
|
||||
if (self.current_alert() is None) == 1:
|
||||
def _update_tab_auto_switch(self):
|
||||
if (self.current_alert is None) == 1:
|
||||
try:
|
||||
self.execute(Command.GET_TITLE)
|
||||
except Exception:
|
||||
|
@ -1426,7 +1490,8 @@ class BrowserManager:
|
|||
for module_path in module_list:
|
||||
print('Load plugins from \"%s\"' % (module_path,), file=sys.stderr)
|
||||
plugins_modules = import_module(module_path)
|
||||
plugins_classes = [type(name, (cls, BrowserPluginParent), {}) for name, cls in plugins_modules.__dict__.items() if name.startswith('BrowserPlugin') and inspect.isclass(cls)]
|
||||
plugins_classes = [type(name, (cls, BrowserPluginParent), {}) for name, cls in plugins_modules.__dict__.items() if
|
||||
name.startswith('BrowserPlugin') and inspect.isclass(cls)]
|
||||
plugins_classes_site.append(plugins_classes)
|
||||
self.load_plugins([element for sublist in plugins_classes_site for element in sublist], is_external=1)
|
||||
except Exception as e:
|
||||
|
@ -1643,7 +1708,7 @@ class BrowserManager:
|
|||
driver.switch_to.window(tab)
|
||||
for i in range(5):
|
||||
try:
|
||||
driver.current_alert().dismiss()
|
||||
driver.current_alert.dismiss()
|
||||
except Exception:
|
||||
break
|
||||
driver.close()
|
||||
|
@ -2016,7 +2081,8 @@ class MainRunner:
|
|||
self.application.setAttribute(Qt.AA_UseHighDpiPixmaps, True)
|
||||
self.application.setHighDpiScaleFactorRoundingPolicy(Qt.HighDpiScaleFactorRoundingPolicy.PassThrough)
|
||||
self.application_scale_rate = self.application.screens()[0].logicalDotsPerInch() / 96
|
||||
self.window = MainWindow(runner=self, app_name=self.app_name, app_version=self.app_version, scale_rate=self.application_scale_rate, web_listen_host=self.web_server_host, web_listen_port=self.web_server_port)
|
||||
self.window = MainWindow(runner=self, app_name=self.app_name, app_version=self.app_version, scale_rate=self.application_scale_rate, web_listen_host=self.web_server_host,
|
||||
web_listen_port=self.web_server_port)
|
||||
sys.exit(self.application.exec_())
|
||||
|
||||
def build(self):
|
||||
|
@ -2063,9 +2129,12 @@ class MainRunner:
|
|||
print('The template file \"%s\" does not exist.' % (compile_template,), file=sys.stderr)
|
||||
return None
|
||||
if (os.path.exists(compiler)) != 1:
|
||||
print('The compiler \"%s\" does not exist. Please check if Inno Setup is installed. You can download it at https://www.innosetup.com/' % (compiler,), file=sys.stderr)
|
||||
print('The compiler \"%s\" does not exist. Please check if Inno Setup is installed. You can download it at https://www.innosetup.com/' % (compiler,),
|
||||
file=sys.stderr)
|
||||
return None
|
||||
Path(compile_file).write_text(Path(compile_template).read_text().replace('%APPNAME%', self.app_name).replace('%APPEXEC%', os.path.splitext(os.path.basename(__file__))[0]).replace('%APPVERSION%', self.app_version))
|
||||
Path(compile_file).write_text(
|
||||
Path(compile_template).read_text().replace('%APPNAME%', self.app_name).replace('%APPEXEC%', os.path.splitext(os.path.basename(__file__))[0]).replace('%APPVERSION%',
|
||||
self.app_version))
|
||||
subprocess.run([compiler, compile_file])
|
||||
|
||||
def _handle_interrupt(self, _signal, _frame):
|
||||
|
|
File diff suppressed because one or more lines are too long
Loading…
Reference in New Issue