Fix examples, and evaluation on Chrome, handle corner cases; Initialize arm support

This commit is contained in:
Timothyxxx
2024-02-26 12:34:27 +08:00
parent 7f81a1ee5a
commit a66b36295a
13 changed files with 361 additions and 200 deletions

View File

@@ -2,26 +2,28 @@ import json
import logging import logging
import os import os
import platform import platform
import time
import sqlite3 import sqlite3
import time
from typing import Dict, Any, List from typing import Dict, Any, List
from pydrive.auth import GoogleAuth
from pydrive.drive import GoogleDrive, GoogleDriveFileList, GoogleDriveFile
from playwright.sync_api import sync_playwright
from urllib.parse import urlparse, parse_qs from urllib.parse import urlparse, parse_qs
import lxml.etree import lxml.etree
import requests
from lxml.cssselect import CSSSelector from lxml.cssselect import CSSSelector
from lxml.etree import _Element from lxml.etree import _Element
from playwright.sync_api import sync_playwright
from pydrive.auth import GoogleAuth
from pydrive.drive import GoogleDrive, GoogleDriveFileList, GoogleDriveFile
_accessibility_ns_map = {"st": "uri:deskat:state.at-spi.gnome.org" _accessibility_ns_map = {
, "attr": "uri:deskat:attributes.at-spi.gnome.org" "st": "uri:deskat:state.at-spi.gnome.org",
, "cp": "uri:deskat:component.at-spi.gnome.org" "attr": "uri:deskat:attributes.at-spi.gnome.org",
, "doc": "uri:deskat:document.at-spi.gnome.org" "cp": "uri:deskat:component.at-spi.gnome.org",
, "docattr": "uri:deskat:attributes.document.at-spi.gnome.org" "doc": "uri:deskat:document.at-spi.gnome.org",
, "txt": "uri:deskat:text.at-spi.gnome.org" "docattr": "uri:deskat:attributes.document.at-spi.gnome.org",
, "val": "uri:deskat:value.at-spi.gnome.org" "txt": "uri:deskat:text.at-spi.gnome.org",
, "act": "uri:deskat:action.at-spi.gnome.org" "val": "uri:deskat:value.at-spi.gnome.org",
"act": "uri:deskat:action.at-spi.gnome.org"
} }
logger = logging.getLogger("desktopenv.getters.chrome") logger = logging.getLogger("desktopenv.getters.chrome")
@@ -44,6 +46,9 @@ def get_default_search_engine(env, config: Dict[str, str]):
"import os; print(os.path.join(os.getenv('HOME'), 'Library/Application Support/Google/Chrome/Default/Preferences'))")[ "import os; print(os.path.join(os.getenv('HOME'), 'Library/Application Support/Google/Chrome/Default/Preferences'))")[
'output'].strip() 'output'].strip()
elif os_type == 'Linux': elif os_type == 'Linux':
if "arm" in platform.machine():
raise NotImplementedError
else:
preference_file_path = env.controller.execute_python_command( preference_file_path = env.controller.execute_python_command(
"import os; print(os.path.join(os.getenv('HOME'), '.config/google-chrome/Default/Preferences'))")[ "import os; print(os.path.join(os.getenv('HOME'), '.config/google-chrome/Default/Preferences'))")[
'output'].strip() 'output'].strip()
@@ -77,6 +82,9 @@ def get_cookie_data(env, config: Dict[str, str]):
"import os; print(os.path.join(os.getenv('HOME'), 'Library/Application Support/Google/Chrome/Default/Cookies'))")[ "import os; print(os.path.join(os.getenv('HOME'), 'Library/Application Support/Google/Chrome/Default/Cookies'))")[
'output'].strip() 'output'].strip()
elif os_type == 'Linux': elif os_type == 'Linux':
if "arm" in platform.machine():
raise NotImplementedError
else:
chrome_cookie_file_path = env.controller.execute_python_command( chrome_cookie_file_path = env.controller.execute_python_command(
"import os; print(os.path.join(os.getenv('HOME'), '.config/google-chrome/Default/Cookies'))")[ "import os; print(os.path.join(os.getenv('HOME'), '.config/google-chrome/Default/Cookies'))")[
'output'].strip() 'output'].strip()
@@ -113,6 +121,9 @@ def get_history(env, config: Dict[str, str]):
"""import os; print(os.path.join(os.getenv('HOME'), "Library", "Application Support", "Google", "Chrome", "Default", "History"))""")[ """import os; print(os.path.join(os.getenv('HOME'), "Library", "Application Support", "Google", "Chrome", "Default", "History"))""")[
'output'].strip() 'output'].strip()
elif os_type == 'Linux': elif os_type == 'Linux':
if "arm" in platform.machine():
raise NotImplementedError
else:
chrome_history_path = env.controller.execute_python_command( chrome_history_path = env.controller.execute_python_command(
"import os; print(os.path.join(os.getenv('HOME'), '.config', 'google-chrome', 'Default', 'History'))")[ "import os; print(os.path.join(os.getenv('HOME'), '.config', 'google-chrome', 'Default', 'History'))")[
'output'].strip() 'output'].strip()
@@ -149,6 +160,9 @@ def get_enabled_experiments(env, config: Dict[str, str]):
"import os; print(os.path.join(os.getenv('HOME'), 'Library/Application Support/Google/Chrome/Local State'))")[ "import os; print(os.path.join(os.getenv('HOME'), 'Library/Application Support/Google/Chrome/Local State'))")[
'output'].strip() 'output'].strip()
elif os_type == 'Linux': elif os_type == 'Linux':
if "arm" in platform.machine():
raise NotImplementedError
else:
preference_file_path = env.controller.execute_python_command( preference_file_path = env.controller.execute_python_command(
"import os; print(os.path.join(os.getenv('HOME'), '.config/google-chrome/Local State'))")[ "import os; print(os.path.join(os.getenv('HOME'), '.config/google-chrome/Local State'))")[
'output'].strip() 'output'].strip()
@@ -181,6 +195,9 @@ def get_profile_name(env, config: Dict[str, str]):
"import os; print(os.path.join(os.getenv('HOME'), 'Library/Application Support/Google/Chrome/Default/Preferences'))")[ "import os; print(os.path.join(os.getenv('HOME'), 'Library/Application Support/Google/Chrome/Default/Preferences'))")[
'output'].strip() 'output'].strip()
elif os_type == 'Linux': elif os_type == 'Linux':
if "arm" in platform.machine():
raise NotImplementedError
else:
preference_file_path = env.controller.execute_python_command( preference_file_path = env.controller.execute_python_command(
"import os; print(os.path.join(os.getenv('HOME'), '.config/google-chrome/Default/Preferences'))")[ "import os; print(os.path.join(os.getenv('HOME'), '.config/google-chrome/Default/Preferences'))")[
'output'].strip() 'output'].strip()
@@ -210,6 +227,9 @@ def get_chrome_language(env, config: Dict[str, str]):
"import os; print(os.path.join(os.getenv('HOME'), 'Library/Application Support/Google/Chrome/Local State'))")[ "import os; print(os.path.join(os.getenv('HOME'), 'Library/Application Support/Google/Chrome/Local State'))")[
'output'].strip() 'output'].strip()
elif os_type == 'Linux': elif os_type == 'Linux':
if "arm" in platform.machine():
raise NotImplementedError
else:
preference_file_path = env.controller.execute_python_command( preference_file_path = env.controller.execute_python_command(
"import os; print(os.path.join(os.getenv('HOME'), '.config/google-chrome/Local State'))")[ "import os; print(os.path.join(os.getenv('HOME'), '.config/google-chrome/Local State'))")[
'output'].strip() 'output'].strip()
@@ -239,6 +259,9 @@ def get_chrome_font_size(env, config: Dict[str, str]):
"import os; print(os.path.join(os.getenv('HOME'), 'Library/Application Support/Google/Chrome/Default/Preferences'))")[ "import os; print(os.path.join(os.getenv('HOME'), 'Library/Application Support/Google/Chrome/Default/Preferences'))")[
'output'].strip() 'output'].strip()
elif os_type == 'Linux': elif os_type == 'Linux':
if "arm" in platform.machine():
raise NotImplementedError
else:
preference_file_path = env.controller.execute_python_command( preference_file_path = env.controller.execute_python_command(
"import os; print(os.path.join(os.getenv('HOME'), '.config/google-chrome/Default/Preferences'))")[ "import os; print(os.path.join(os.getenv('HOME'), '.config/google-chrome/Default/Preferences'))")[
'output'].strip() 'output'].strip()
@@ -274,6 +297,9 @@ def get_bookmarks(env, config: Dict[str, str]):
"import os; print(os.path.join(os.getenv('HOME'), 'Library/Application Support/Google/Chrome/Default/Bookmarks'))")[ "import os; print(os.path.join(os.getenv('HOME'), 'Library/Application Support/Google/Chrome/Default/Bookmarks'))")[
'output'].strip() 'output'].strip()
elif os_type == 'Linux': elif os_type == 'Linux':
if "arm" in platform.machine():
raise NotImplementedError
else:
preference_file_path = env.controller.execute_python_command( preference_file_path = env.controller.execute_python_command(
"import os; print(os.path.join(os.getenv('HOME'), '.config/google-chrome/Default/Bookmarks'))")[ "import os; print(os.path.join(os.getenv('HOME'), '.config/google-chrome/Default/Bookmarks'))")[
'output'].strip() 'output'].strip()
@@ -301,6 +327,9 @@ def get_extensions_installed_from_shop(env, config: Dict[str, str]):
"""os.path.expanduser('~') + '/Library/Application Support/Google/Chrome/Default/Extensions/'""")[ """os.path.expanduser('~') + '/Library/Application Support/Google/Chrome/Default/Extensions/'""")[
'output'].strip() 'output'].strip()
elif os_type == 'Linux': elif os_type == 'Linux':
if "arm" in platform.machine():
raise NotImplementedError
else:
chrome_extension_dir = env.controller.execute_python_command( chrome_extension_dir = env.controller.execute_python_command(
"""os.path.expanduser('~') + '/.config/google-chrome/Default/Extensions/'""")['output'].strip() """os.path.expanduser('~') + '/.config/google-chrome/Default/Extensions/'""")['output'].strip()
else: else:
@@ -335,7 +364,28 @@ def get_page_info(env, config: Dict[str, str]):
remote_debugging_url = f"http://{host}:{port}" remote_debugging_url = f"http://{host}:{port}"
with sync_playwright() as p: with sync_playwright() as p:
# connect to remote Chrome instance # connect to remote Chrome instance
try:
browser = p.chromium.connect_over_cdp(remote_debugging_url) browser = p.chromium.connect_over_cdp(remote_debugging_url)
except Exception as e:
# If the connection fails, start a new browser instance
platform.machine()
if "arm" in platform.machine():
# start a new browser instance if the connection fails
payload = json.dumps({"command": [
"chromium",
"--remote-debugging-port=1337"
], "shell": False})
else:
payload = json.dumps({"command": [
"google-chrome",
"--remote-debugging-port=1337"
], "shell": False})
headers = {"Content-Type": "application/json"}
requests.post("http://" + host + ":5000/setup" + "/launch", headers=headers, data=payload)
time.sleep(5)
browser = p.chromium.connect_over_cdp(remote_debugging_url)
page = browser.contexts[0].new_page() page = browser.contexts[0].new_page()
page.goto(url) page.goto(url)
@@ -364,7 +414,30 @@ def get_open_tabs_info(env, config: Dict[str, str]):
remote_debugging_url = f"http://{host}:{port}" remote_debugging_url = f"http://{host}:{port}"
with sync_playwright() as p: with sync_playwright() as p:
# connect to remote Chrome instance # connect to remote Chrome instance
try:
browser = p.chromium.connect_over_cdp(remote_debugging_url) browser = p.chromium.connect_over_cdp(remote_debugging_url)
except Exception as e:
# If the connection fails, start a new browser instance
platform.machine()
if "arm" in platform.machine():
# start a new browser instance if the connection fails
payload = json.dumps({"command": [
"chromium",
"--remote-debugging-port=1337"
], "shell": False})
else:
payload = json.dumps({"command": [
"google-chrome",
"--remote-debugging-port=1337"
], "shell": False})
headers = {"Content-Type": "application/json"}
requests.post("http://" + host + ":5000/setup" + "/launch", headers=headers, data=payload)
time.sleep(5)
try:
browser = p.chromium.connect_over_cdp(remote_debugging_url)
except Exception as e:
return []
tabs_info = [] tabs_info = []
for context in browser.contexts: for context in browser.contexts:
@@ -414,9 +487,11 @@ def get_active_url_from_accessTree(env, config):
at: _Element = lxml.etree.fromstring(accessibility_tree) at: _Element = lxml.etree.fromstring(accessibility_tree)
arch = platform.machine() arch = platform.machine()
if "arm" in arch: if "arm" in arch:
selector = CSSSelector(", application[name=Chromium] entry[name=Address\\ and\\ search\\ bar]", namespaces=_accessibility_ns_map) selector = CSSSelector(", application[name=Chromium] entry[name=Address\\ and\\ search\\ bar]",
namespaces=_accessibility_ns_map)
else: else:
selector = CSSSelector(", application[name=Google\\ Chrome] entry[name=Address\\ and\\ search\\ bar]", namespaces=_accessibility_ns_map) selector = CSSSelector("application[name=Google\\ Chrome] entry[name=Address\\ and\\ search\\ bar]",
namespaces=_accessibility_ns_map)
elements: List[_Element] = selector(at) elements: List[_Element] = selector(at)
# if "xpath" in config: # if "xpath" in config:
# elements: List[_Element] = at.xpath(config["xpath"], namespaces=_accessibility_ns_map) # elements: List[_Element] = at.xpath(config["xpath"], namespaces=_accessibility_ns_map)
@@ -425,8 +500,9 @@ def get_active_url_from_accessTree(env, config):
# elements: List[_Element] = selector(at) # elements: List[_Element] = selector(at)
if len(elements) == 0: if len(elements) == 0:
print("no elements found") print("no elements found")
return 0. return None
active_tab_url = config["goto_prefix"]+elements[0].text if "goto_prefix" in config.keys() else "https://" + elements[0].text active_tab_url = config["goto_prefix"] + elements[0].text if "goto_prefix" in config.keys() else "https://" + \
elements[0].text
print("active tab url now: {}".format(active_tab_url)) print("active tab url now: {}".format(active_tab_url))
return active_tab_url return active_tab_url
@@ -448,8 +524,12 @@ def get_active_tab_info(env, config: Dict[str, str]):
remote_debugging_url = f"http://{host}:{port}" remote_debugging_url = f"http://{host}:{port}"
with sync_playwright() as p: with sync_playwright() as p:
# connect to remote Chrome instance # connect to remote Chrome instance, since it is supposed to be the active one, we won't start a new one if failed
try:
browser = p.chromium.connect_over_cdp(remote_debugging_url) browser = p.chromium.connect_over_cdp(remote_debugging_url)
except Exception as e:
return None
active_tab_info = {} active_tab_info = {}
# go to the target URL page # go to the target URL page
page = browser.new_page() page = browser.new_page()
@@ -481,7 +561,28 @@ def get_pdf_from_url(env, config: Dict[str, str]) -> str:
remote_debugging_url = f"http://{host}:{port}" remote_debugging_url = f"http://{host}:{port}"
with sync_playwright() as p: with sync_playwright() as p:
try:
browser = p.chromium.connect_over_cdp(remote_debugging_url) browser = p.chromium.connect_over_cdp(remote_debugging_url)
except Exception as e:
# If the connection fails, start a new browser instance
platform.machine()
if "arm" in platform.machine():
# start a new browser instance if the connection fails
payload = json.dumps({"command": [
"chromium",
"--remote-debugging-port=1337"
], "shell": False})
else:
payload = json.dumps({"command": [
"google-chrome",
"--remote-debugging-port=1337"
], "shell": False})
headers = {"Content-Type": "application/json"}
requests.post("http://" + host + ":5000/setup" + "/launch", headers=headers, data=payload)
time.sleep(5)
browser = p.chromium.connect_over_cdp(remote_debugging_url)
page = browser.new_page() page = browser.new_page()
page.goto(_url) page.goto(_url)
page.pdf(path=_path) page.pdf(path=_path)
@@ -498,6 +599,26 @@ def get_chrome_saved_address(env, config: Dict[str, str]):
remote_debugging_url = f"http://{host}:{port}" remote_debugging_url = f"http://{host}:{port}"
with sync_playwright() as p: with sync_playwright() as p:
# connect to remote Chrome instance # connect to remote Chrome instance
try:
browser = p.chromium.connect_over_cdp(remote_debugging_url)
except Exception as e:
# If the connection fails, start a new browser instance
platform.machine()
if "arm" in platform.machine():
# start a new browser instance if the connection fails
payload = json.dumps({"command": [
"chromium",
"--remote-debugging-port=1337"
], "shell": False})
else:
payload = json.dumps({"command": [
"google-chrome",
"--remote-debugging-port=1337"
], "shell": False})
headers = {"Content-Type": "application/json"}
requests.post("http://" + host + ":5000/setup" + "/launch", headers=headers, data=payload)
time.sleep(5)
browser = p.chromium.connect_over_cdp(remote_debugging_url) browser = p.chromium.connect_over_cdp(remote_debugging_url)
page = browser.new_page() page = browser.new_page()
@@ -556,6 +677,26 @@ def get_number_of_search_results(env, config: Dict[str, str]):
remote_debugging_url = f"http://{host}:{port}" remote_debugging_url = f"http://{host}:{port}"
with sync_playwright() as p: with sync_playwright() as p:
try:
browser = p.chromium.connect_over_cdp(remote_debugging_url)
except Exception as e:
# If the connection fails, start a new browser instance
platform.machine()
if "arm" in platform.machine():
# start a new browser instance if the connection fails
payload = json.dumps({"command": [
"chromium",
"--remote-debugging-port=1337"
], "shell": False})
else:
payload = json.dumps({"command": [
"google-chrome",
"--remote-debugging-port=1337"
], "shell": False})
headers = {"Content-Type": "application/json"}
requests.post("http://" + host + ":5000/setup" + "/launch", headers=headers, data=payload)
time.sleep(5)
browser = p.chromium.connect_over_cdp(remote_debugging_url) browser = p.chromium.connect_over_cdp(remote_debugging_url)
page = browser.new_page() page = browser.new_page()
page.goto(url) page.goto(url)
@@ -634,12 +775,15 @@ def get_enable_do_not_track(env, config: Dict[str, str]):
"import os; print(os.path.join(os.getenv('HOME'), 'Library/Application Support/Google/Chrome/Default/Preferences'))")[ "import os; print(os.path.join(os.getenv('HOME'), 'Library/Application Support/Google/Chrome/Default/Preferences'))")[
'output'].strip() 'output'].strip()
elif os_type == 'Linux': elif os_type == 'Linux':
if "arm" in platform.machine():
preference_file_path = env.controller.execute_python_command(
"import os; print(os.path.join(os.getenv('HOME'), 'snap/chromium/common/chromium/Default/Preferences'))")[
'output'].strip()
else:
preference_file_path = env.controller.execute_python_command( preference_file_path = env.controller.execute_python_command(
"import os; print(os.path.join(os.getenv('HOME'), '.config/google-chrome/Default/Preferences'))")[ "import os; print(os.path.join(os.getenv('HOME'), '.config/google-chrome/Default/Preferences'))")[
'output'].strip() 'output'].strip()
# preference_file_path = env.controller.execute_python_command(
# "import os; print(os.path.join(os.getenv('HOME'), 'snap/chromium/common/chromium/Default/Preferences'))")[
# 'output'].strip()
else: else:
raise Exception('Unsupported operating system') raise Exception('Unsupported operating system')
@@ -664,12 +808,15 @@ def get_enable_enhanced_safety_browsing(env, config: Dict[str, str]):
"import os; print(os.path.join(os.getenv('HOME'), 'Library/Application Support/Google/Chrome/Default/Preferences'))")[ "import os; print(os.path.join(os.getenv('HOME'), 'Library/Application Support/Google/Chrome/Default/Preferences'))")[
'output'].strip() 'output'].strip()
elif os_type == 'Linux': elif os_type == 'Linux':
if "arm" in platform.machine():
preference_file_path = env.controller.execute_python_command(
"import os; print(os.path.join(os.getenv('HOME'), 'snap/chromium/common/chromium/Default/Preferences'))")[
'output'].strip()
else:
preference_file_path = env.controller.execute_python_command( preference_file_path = env.controller.execute_python_command(
"import os; print(os.path.join(os.getenv('HOME'), '.config/google-chrome/Default/Preferences'))")[ "import os; print(os.path.join(os.getenv('HOME'), '.config/google-chrome/Default/Preferences'))")[
'output'].strip() 'output'].strip()
# preference_file_path = env.controller.execute_python_command(
# "import os; print(os.path.join(os.getenv('HOME'), 'snap/chromium/common/chromium/Default/Preferences'))")[
# 'output'].strip()
else: else:
raise Exception('Unsupported operating system') raise Exception('Unsupported operating system')
@@ -694,12 +841,15 @@ def get_new_startup_page(env, config: Dict[str, str]):
"import os; print(os.path.join(os.getenv('HOME'), 'Library/Application Support/Google/Chrome/Default/Preferences'))")[ "import os; print(os.path.join(os.getenv('HOME'), 'Library/Application Support/Google/Chrome/Default/Preferences'))")[
'output'].strip() 'output'].strip()
elif os_type == 'Linux': elif os_type == 'Linux':
if "arm" in platform.machine():
preference_file_path = env.controller.execute_python_command(
"import os; print(os.path.join(os.getenv('HOME'), 'snap/chromium/common/chromium/Default/Preferences'))")[
'output'].strip()
else:
preference_file_path = env.controller.execute_python_command( preference_file_path = env.controller.execute_python_command(
"import os; print(os.path.join(os.getenv('HOME'), '.config/google-chrome/Default/Preferences'))")[ "import os; print(os.path.join(os.getenv('HOME'), '.config/google-chrome/Default/Preferences'))")[
'output'].strip() 'output'].strip()
# preference_file_path = env.controller.execute_python_command(
# "import os; print(os.path.join(os.getenv('HOME'), 'snap/chromium/common/chromium/Default/Preferences'))")[
# 'output'].strip()
else: else:
raise Exception('Unsupported operating system') raise Exception('Unsupported operating system')
@@ -729,12 +879,15 @@ def get_find_unpacked_extension_path(env, config: Dict[str, str]):
"import os; print(os.path.join(os.getenv('HOME'), 'Library/Application Support/Google/Chrome/Default/Preferences'))")[ "import os; print(os.path.join(os.getenv('HOME'), 'Library/Application Support/Google/Chrome/Default/Preferences'))")[
'output'].strip() 'output'].strip()
elif os_type == 'Linux': elif os_type == 'Linux':
if "arm" in platform.machine():
preference_file_path = env.controller.execute_python_command(
"import os; print(os.path.join(os.getenv('HOME'), 'snap/chromium/common/chromium/Default/Preferences'))")[
'output'].strip()
else:
preference_file_path = env.controller.execute_python_command( preference_file_path = env.controller.execute_python_command(
"import os; print(os.path.join(os.getenv('HOME'), '.config/google-chrome/Default/Preferences'))")[ "import os; print(os.path.join(os.getenv('HOME'), '.config/google-chrome/Default/Preferences'))")[
'output'].strip() 'output'].strip()
# preference_file_path = env.controller.execute_python_command(
# "import os; print(os.path.join(os.getenv('HOME'), 'snap/chromium/common/chromium/Default/Preferences'))")[
# 'output'].strip()
else: else:
raise Exception('Unsupported operating system') raise Exception('Unsupported operating system')
@@ -766,12 +919,14 @@ def get_data_delete_automacally(env, config: Dict[str, str]):
"import os; print(os.path.join(os.getenv('HOME'), 'Library/Application Support/Google/Chrome/Default/Preferences'))")[ "import os; print(os.path.join(os.getenv('HOME'), 'Library/Application Support/Google/Chrome/Default/Preferences'))")[
'output'].strip() 'output'].strip()
elif os_type == 'Linux': elif os_type == 'Linux':
# preference_file_path = env.controller.execute_python_command( if "arm" in platform.machine():
# "import os; print(os.path.join(os.getenv('HOME'), '.config/google-chrome/Default/Preferences'))")[
# 'output'].strip()
preference_file_path = env.controller.execute_python_command( preference_file_path = env.controller.execute_python_command(
"import os; print(os.path.join(os.getenv('HOME'), 'snap/chromium/common/chromium/Default/Preferences'))")[ "import os; print(os.path.join(os.getenv('HOME'), 'snap/chromium/common/chromium/Default/Preferences'))")[
'output'].strip() 'output'].strip()
else:
preference_file_path = env.controller.execute_python_command(
"import os; print(os.path.join(os.getenv('HOME'), '.config/google-chrome/Default/Preferences'))")[
'output'].strip()
else: else:
raise Exception('Unsupported operating system') raise Exception('Unsupported operating system')
@@ -823,6 +978,26 @@ def get_active_tab_html_parse(env, config: Dict[str, Any]):
remote_debugging_url = f"http://{host}:{port}" remote_debugging_url = f"http://{host}:{port}"
with sync_playwright() as p: with sync_playwright() as p:
# connect to remote Chrome instance # connect to remote Chrome instance
try:
browser = p.chromium.connect_over_cdp(remote_debugging_url)
except Exception as e:
# If the connection fails, start a new browser instance
platform.machine()
if "arm" in platform.machine():
# start a new browser instance if the connection fails
payload = json.dumps({"command": [
"chromium",
"--remote-debugging-port=1337"
], "shell": False})
else:
payload = json.dumps({"command": [
"google-chrome",
"--remote-debugging-port=1337"
], "shell": False})
headers = {"Content-Type": "application/json"}
requests.post("http://" + host + ":5000/setup" + "/launch", headers=headers, data=payload)
time.sleep(5)
browser = p.chromium.connect_over_cdp(remote_debugging_url) browser = p.chromium.connect_over_cdp(remote_debugging_url)
target_page = None target_page = None
for context in browser.contexts: for context in browser.contexts:
@@ -831,6 +1006,8 @@ def get_active_tab_html_parse(env, config: Dict[str, Any]):
if page.url == active_tab_url: if page.url == active_tab_url:
target_page = page target_page = page
break break
if target_page is None:
return {}
return_json = {} return_json = {}
if config["category"] == "class": if config["category"] == "class":
# find the text of elements in html with specific class name # find the text of elements in html with specific class name
@@ -838,7 +1015,8 @@ def get_active_tab_html_parse(env, config: Dict[str, Any]):
for key in class_multiObject.keys(): for key in class_multiObject.keys():
object_dict = class_multiObject[key] object_dict = class_multiObject[key]
for order_key in object_dict.keys(): for order_key in object_dict.keys():
return_json[object_dict[order_key]] = target_page.query_selector_all("."+key)[int(order_key)].text_content().strip() return_json[object_dict[order_key]] = target_page.query_selector_all("." + key)[
int(order_key)].text_content().strip()
class_singleObject = config["class_singleObject"] class_singleObject = config["class_singleObject"]
for key in class_singleObject.keys(): for key in class_singleObject.keys():
return_json[class_singleObject[key]] = target_page.query_selector("." + key).text_content().strip() return_json[class_singleObject[key]] = target_page.query_selector("." + key).text_content().strip()
@@ -869,6 +1047,26 @@ def get_gotoRecreationPage_and_get_html_content(env, config: Dict[str, Any]):
remote_debugging_url = f"http://{host}:{port}" remote_debugging_url = f"http://{host}:{port}"
with sync_playwright() as p: with sync_playwright() as p:
try:
browser = p.chromium.connect_over_cdp(remote_debugging_url)
except Exception as e:
# If the connection fails, start a new browser instance
platform.machine()
if "arm" in platform.machine():
# start a new browser instance if the connection fails
payload = json.dumps({"command": [
"chromium",
"--remote-debugging-port=1337"
], "shell": False})
else:
payload = json.dumps({"command": [
"google-chrome",
"--remote-debugging-port=1337"
], "shell": False})
headers = {"Content-Type": "application/json"}
requests.post("http://" + host + ":5000/setup" + "/launch", headers=headers, data=payload)
time.sleep(5)
browser = p.chromium.connect_over_cdp(remote_debugging_url) browser = p.chromium.connect_over_cdp(remote_debugging_url)
page = browser.new_page() page = browser.new_page()
page.goto("https://www.recreation.gov/") page.goto("https://www.recreation.gov/")
@@ -888,14 +1086,14 @@ def get_gotoRecreationPage_and_get_html_content(env, config: Dict[str, Any]):
newpage.click("button.next-available") newpage.click("button.next-available")
print("after third click") print("after third click")
return_json = {} return_json = {}
return_json["expected"] = {} return_json["expected"] = {}
# find the text of elements in html with specific class name # find the text of elements in html with specific class name
if config["selector"] == "class": if config["selector"] == "class":
if "order" in config.keys(): if "order" in config.keys():
className = config["class"] className = config["class"]
return_json["expected"][className]=newpage.query_selector_all("."+className)[int(config["order"])].text_content().strip() return_json["expected"][className] = newpage.query_selector_all("." + className)[
int(config["order"])].text_content().strip()
else: else:
className = config["class"] className = config["class"]
return_json["expected"][className] = newpage.query_selector("." + className).text_content().strip() return_json["expected"][className] = newpage.query_selector("." + className).text_content().strip()
@@ -914,6 +1112,8 @@ def get_active_tab_url_parse(env, config:Dict[str, Any]):
( { "original key": "new key" } ) ( { "original key": "new key" } )
""" """
active_tab_url = get_active_url_from_accessTree(env, config) active_tab_url = get_active_url_from_accessTree(env, config)
if active_tab_url is None:
return None
# connect to remote Chrome instance # connect to remote Chrome instance
# parse in a hard-coded way to find the specific info about task # parse in a hard-coded way to find the specific info about task
@@ -944,6 +1144,8 @@ def get_url_dashPart(env, config: Dict[str, str]):
a string, used to indicate the return type, "string" or "json". a string, used to indicate the return type, "string" or "json".
""" """
active_tab_url = get_active_url_from_accessTree(env, config) active_tab_url = get_active_url_from_accessTree(env, config)
if active_tab_url is None:
return None
# extract the last dash-separated part of the URL, and delete all the characters after "id" # extract the last dash-separated part of the URL, and delete all the characters after "id"
dash_part = active_tab_url.split("/")[config["partIndex"]] dash_part = active_tab_url.split("/")[config["partIndex"]]

View File

@@ -17,11 +17,17 @@ def is_expected_active_tab(active_tab_info: Dict[str, str], rule: Dict[str, Any]
""" """
Checks if the expected active tab is open in Chrome. Checks if the expected active tab is open in Chrome.
""" """
if not active_tab_info:
return 0.
match_type = rule['type'] match_type = rule['type']
if match_type == "url": if match_type == "url":
expected_url = rule['url'] expected_url = rule['url']
if isinstance(active_tab_info, Dict):
actual_url = active_tab_info.get('url', None) actual_url = active_tab_info.get('url', None)
else:
actual_url = active_tab_info
print("expected_url: {}".format(expected_url)) print("expected_url: {}".format(expected_url))
print("actual_url: {}".format(actual_url)) print("actual_url: {}".format(actual_url))
return 1 if compare_urls(expected_url, actual_url) else 0 return 1 if compare_urls(expected_url, actual_url) else 0
@@ -29,12 +35,16 @@ def is_expected_active_tab(active_tab_info: Dict[str, str], rule: Dict[str, Any]
logger.error(f"Unknown type: {match_type}") logger.error(f"Unknown type: {match_type}")
return 0 return 0
# rules[expected] is a string-formatted regex # rules[expected] is a string-formatted regex
def is_expected_url_pattern_match(result, rules) -> float: def is_expected_url_pattern_match(result, rules) -> float:
""" """
This function is used to search the expected pattern in the url using regex. This function is used to search the expected pattern in the url using regex.
result is the return value of function "activte_tab_info" or return value of function "get_active_url_from_accessTree" result is the return value of function "activte_tab_info" or return value of function "get_active_url_from_accessTree"
""" """
if not result:
return 0.
if type(result) == dict: if type(result) == dict:
result_url = result["url"] result_url = result["url"]
print("result url: {}".format(result_url)) print("result url: {}".format(result_url))

View File

@@ -231,8 +231,11 @@ def check_direct_json_object(result, rules)->float:
print(result) print(result)
print("expected: ") print("expected: ")
print(rules["expected"]) print(rules["expected"])
if result is None:
return 0.
expected_json = rules["expected"] expected_json = rules["expected"]
for key in expected_json.keys(): for key in expected_json.keys():
if expected_json[key] != result[key]: expected_value = expected_json.get(key)
if expected_value != result.get(key):
return 0. return 0.
return 1.0 return 1.0

View File

@@ -55,7 +55,8 @@
"evaluator": { "evaluator": {
"func": "is_expected_active_tab", "func": "is_expected_active_tab",
"result": { "result": {
"type": "active_tab_info" "type": "active_url_from_accessTree",
"goto_prefix": "https://www."
}, },
"expected": { "expected": {
"type": "rule", "type": "rule",

View File

@@ -30,7 +30,8 @@
"evaluator": { "evaluator": {
"func": "is_expected_active_tab", "func": "is_expected_active_tab",
"result": { "result": {
"type": "active_tab_info" "type": "active_url_from_accessTree",
"goto_prefix": ""
}, },
"expected": { "expected": {
"type": "rule", "type": "rule",

View File

@@ -53,7 +53,7 @@
"key":"time" "key":"time"
}, },
{ {
"type": "active_tab_info", "type": "active_url_from_accessTree",
"goto_prefix": "https://www." "goto_prefix": "https://www."
}], }],
"expected":[ "expected":[

View File

@@ -55,7 +55,8 @@
"evaluator": { "evaluator": {
"func": "is_expected_active_tab", "func": "is_expected_active_tab",
"result": { "result": {
"type": "active_tab_info" "type": "active_url_from_accessTree",
"goto_prefix": "https://www."
}, },
"expected": { "expected": {
"type": "rule", "type": "rule",

View File

@@ -45,10 +45,7 @@
"evaluator": { "evaluator": {
"func": "is_expected_url_pattern_match", "func": "is_expected_url_pattern_match",
"result": { "result": {
"type": "active_tab_info", "type": "active_url_from_accessTree",
"selectors": [
"application[name=Chromium] entry[name=Address\\ and\\ search\\ bar]"
],
"goto_prefix": "https://www." "goto_prefix": "https://www."
}, },
"expected": { "expected": {
@@ -58,22 +55,6 @@
"forms\\?title=&field_component_target_id=431" "forms\\?title=&field_component_target_id=431"
] ]
} }
],
"trajectory": "trajectories/",
"related_apps": [
"chrome"
],
"evaluator": {
"func":"is_expected_url_pattern_match",
"result": {
"type": "active_tab_info",
"goto_prefix": "https://www."
},
"expected":{
"type": "rule",
"rules":{
"expected": ["forms\\?title=&field_component_target_id=431"]
}
} }
} }
} }

View File

@@ -55,7 +55,8 @@
"evaluator": { "evaluator": {
"func": "is_expected_active_tab", "func": "is_expected_active_tab",
"result": { "result": {
"type": "active_tab_info" "type": "active_url_from_accessTree",
"goto_prefix": "https://www."
}, },
"expected": { "expected": {
"type": "rule", "type": "rule",

View File

@@ -42,28 +42,6 @@
"related_apps": [ "related_apps": [
"chrome" "chrome"
], ],
"evaluator": {
"func": "exact_match",
"result": {
"type": "url_dashPart",
"selectors": [
"application[name=Chromium] entry[name=Address\\ and\\ search\\ bar]"
],
"goto_prefix": "https://www.",
"partIndex": -1,
"needDeleteId": false,
"returnType": "string"
},
"expected": {
"type": "rule",
"rules": {
"expected": "tamiflu.html#side-effects"
}
],
"trajectory": "trajectories/",
"related_apps": [
"chrome"
],
"evaluator": { "evaluator": {
"func": "exact_match", "func": "exact_match",
"result": { "result": {

View File

@@ -46,9 +46,6 @@
"func": "is_expected_url_pattern_match", "func": "is_expected_url_pattern_match",
"result": { "result": {
"type": "active_url_from_accessTree", "type": "active_url_from_accessTree",
"selectors": [
"application[name=Chromium] entry[name=Address\\ and\\ search\\ bar]"
],
"goto_prefix": "https://www." "goto_prefix": "https://www."
}, },
"expected": { "expected": {
@@ -60,22 +57,6 @@
"S=4" "S=4"
] ]
} }
],
"trajectory": "trajectories/",
"related_apps": [
"chrome"
],
"evaluator": {
"func":"is_expected_url_pattern_match",
"result": {
"type": "active_url_from_accessTree",
"goto_prefix": "https://www."
},
"expected":{
"type": "rule",
"rules":{
"expected": ["AgeAppropriate:Kids", "search=spider-man%20toys", "S=4"]
}
} }
} }
} }

View File

@@ -55,7 +55,8 @@
"evaluator": { "evaluator": {
"func": "is_expected_active_tab", "func": "is_expected_active_tab",
"result": { "result": {
"type": "active_tab_info" "type": "active_url_from_accessTree",
"goto_prefix": "https://www."
}, },
"expected": { "expected": {
"type": "rule", "type": "rule",

View File

@@ -55,7 +55,8 @@
"evaluator": { "evaluator": {
"func": "is_expected_active_tab", "func": "is_expected_active_tab",
"result": { "result": {
"type": "active_tab_info" "type": "active_url_from_accessTree",
"goto_prefix": "https://www."
}, },
"expected": { "expected": {
"type": "rule", "type": "rule",