Finish Chrome v2 loading

desktop_env/envs/desktop_env.py

@@ -104,6 +104,9 @@ class DesktopEnv(gym.Env):
         self.action_history: List[Dict[str, any]] = []
 
     def _config_screen_size(self):
+        """
+        fixme: Experimental features, will cause unexpected error when system corrupt
+        """
         def calculate_vram_size(width, height, bits_per_pixel=32):
             """
             Calculate VRAM size for given width, height, and color depth.

desktop_env/evaluators/getters/__init__.py

@@ -1,10 +1,12 @@
 from .chrome import get_default_search_engine, get_cookie_data, get_bookmarks, get_open_tabs_info, get_pdf_from_url, \
-    get_shortcuts_on_desktop
+    get_shortcuts_on_desktop, get_history, get_enabled_experiments, get_chrome_language, get_chrome_font_size, \
+    get_profile_name, \
+    get_number_of_search_results
 from .file import get_cloud_file, get_vm_file, get_cache_file
 from .general import get_vm_command_line
+from .impress import get_audio_in_slide
 from .info import get_vm_screen_size, get_vm_window_size, get_vm_wallpaper
 from .misc import get_rule, get_accessibility_tree
 from .replay import get_replay
 from .vlc import get_vlc_playing_info, get_vlc_config
 from .vscode import get_vscode_config
-from .impress import get_audio_in_slide

desktop_env/evaluators/getters/chrome.py

@@ -84,6 +84,168 @@ def get_cookie_data(env, config: Dict[str, str]):
         return None
 
 
+def get_history(env, config: Dict[str, str]):
+    os_type = env.vm_platform
+    if os_type == 'Windows':
+        chrome_history_path = env.controller.execute_python_command(
+            """import os; print(os.path.join(os.getenv('USERPROFILE'), "AppData", "Local", "Google", "Chrome", "User Data", "Default", "History"))""")[
+            'output'].strip()
+    elif os_type == 'Darwin':
+        chrome_history_path = env.controller.execute_python_command(
+            """import os; print(os.path.join(os.getenv('HOME'), "Library", "Application Support", "Google", "Chrome", "Default", "History"))""")[
+            'output'].strip()
+    elif os_type == 'Linux':
+        chrome_history_path = env.controller.execute_python_command(
+            "import os; print(os.path.join(os.getenv('HOME'), '.config', 'google-chrome', 'Default', 'History'))")[
+            'output'].strip()
+    else:
+        raise Exception('Unsupported operating system')
+
+    try:
+        content = env.controller.get_file(chrome_history_path)
+        _path = os.path.join(env.cache_dir, config["dest"])
+
+        with open(_path, "wb") as f:
+            f.write(content)
+
+        conn = sqlite3.connect(_path)
+        cursor = conn.cursor()
+
+        # Query to check for OpenAI cookies
+        cursor.execute("SELECT url, title, last_visit_time FROM urls")
+        history_items = cursor.fetchall()
+        return history_items
+    except Exception as e:
+        logger.error(f"Error: {e}")
+        return None
+
+
+def get_enabled_experiments(env, config: Dict[str, str]):
+    os_type = env.vm_platform
+    if os_type == 'Windows':
+        preference_file_path = env.controller.execute_python_command("""import os; print(os.path.join(os.getenv('LOCALAPPDATA'),
+                                                'Google\\Chrome\\User Data\\Local State'))""")[
+            'output'].strip()
+    elif os_type == 'Darwin':
+        preference_file_path = env.controller.execute_python_command(
+            "import os; print(os.path.join(os.getenv('HOME'), 'Library/Application Support/Google/Chrome/Local State'))")[
+            'output'].strip()
+    elif os_type == 'Linux':
+        preference_file_path = env.controller.execute_python_command(
+            "import os; print(os.path.join(os.getenv('HOME'), '.config/google-chrome/Local State'))")[
+            'output'].strip()
+    else:
+        raise Exception('Unsupported operating system')
+
+    try:
+        content = env.controller.get_file(preference_file_path)
+        data = json.loads(content)
+
+        # The path within the JSON data to the default search engine might vary
+        enabled_labs_experiments = data.get('browser', {}).get('enabled_labs_experiments', [])
+        return enabled_labs_experiments
+    except Exception as e:
+        logger.error(f"Error: {e}")
+        return []
+
+
+def get_profile_name(env, config: Dict[str, str]):
+    """
+    Get the username from the Chrome browser.
+    Assume the cookies are stored in the default location, not encrypted and not large in size.
+    """
+    os_type = env.vm_platform
+    if os_type == 'Windows':
+        preference_file_path = env.controller.execute_python_command("""import os; print(os.path.join(os.getenv('LOCALAPPDATA'),
+                                            'Google\\Chrome\\User Data\\Default\\Preferences'))""")['output'].strip()
+    elif os_type == 'Darwin':
+        preference_file_path = env.controller.execute_python_command(
+            "import os; print(os.path.join(os.getenv('HOME'), 'Library/Application Support/Google/Chrome/Default/Preferences'))")[
+            'output'].strip()
+    elif os_type == 'Linux':
+        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:
+        raise Exception('Unsupported operating system')
+
+    try:
+        content = env.controller.get_file(preference_file_path)
+        data = json.loads(content)
+
+        # The path within the JSON data to the default search engine might vary
+        profile_name = data.get('profile', {}).get('name', None)
+        return profile_name
+    except Exception as e:
+        logger.error(f"Error: {e}")
+        return None
+
+
+def get_chrome_language(env, config: Dict[str, str]):
+    os_type = env.vm_platform
+    if os_type == 'Windows':
+        preference_file_path = env.controller.execute_python_command("""import os; print(os.path.join(os.getenv('LOCALAPPDATA'),
+                                                    'Google\\Chrome\\User Data\\Local State'))""")[
+            'output'].strip()
+    elif os_type == 'Darwin':
+        preference_file_path = env.controller.execute_python_command(
+            "import os; print(os.path.join(os.getenv('HOME'), 'Library/Application Support/Google/Chrome/Local State'))")[
+            'output'].strip()
+    elif os_type == 'Linux':
+        preference_file_path = env.controller.execute_python_command(
+            "import os; print(os.path.join(os.getenv('HOME'), '.config/google-chrome/Local State'))")[
+            'output'].strip()
+    else:
+        raise Exception('Unsupported operating system')
+
+    try:
+        content = env.controller.get_file(preference_file_path)
+        data = json.loads(content)
+
+        # The path within the JSON data to the default search engine might vary
+        enabled_labs_experiments = data.get('intl', {}).get('app_locale', "en-US")
+        return enabled_labs_experiments
+    except Exception as e:
+        logger.error(f"Error: {e}")
+        return "en-US"
+
+
+def get_chrome_font_size(env, config: Dict[str, str]):
+    os_type = env.vm_platform
+    if os_type == 'Windows':
+        preference_file_path = env.controller.execute_python_command("""import os; print(os.path.join(os.getenv('LOCALAPPDATA'),
+                                                'Google\\Chrome\\User Data\\Default\\Preferences'))""")[
+            'output'].strip()
+    elif os_type == 'Darwin':
+        preference_file_path = env.controller.execute_python_command(
+            "import os; print(os.path.join(os.getenv('HOME'), 'Library/Application Support/Google/Chrome/Default/Preferences'))")[
+            'output'].strip()
+    elif os_type == 'Linux':
+        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:
+        raise Exception('Unsupported operating system')
+
+    try:
+        content = env.controller.get_file(preference_file_path)
+        data = json.loads(content)
+
+        # The path within the JSON data to the default search engine might vary
+        search_engine = data.get('webkit', {}).get('webprefs', {
+            "default_fixed_font_size": 13,
+            "default_font_size": 16,
+            "minimum_font_size": 13
+        })
+        return search_engine
+    except Exception as e:
+        logger.error(f"Error: {e}")
+        return {
+            "default_fixed_font_size": 13,
+            "default_font_size": 16
+        }
+
+
 def get_bookmarks(env, config: Dict[str, str]):
     os_type = env.vm_platform
     if os_type == 'Windows':
@@ -227,8 +389,7 @@ def get_pdf_from_url(env, config: Dict[str, str]) -> str:
 
 # fixme: needs to be changed (maybe through post-processing) since it's not working
 def get_chrome_saved_address(env, config: Dict[str, str]):
-    # host = env.vm_ip
-    host = "192.168.13.130"
+    host = env.vm_ip
     port = 9222  # fixme: this port is hard-coded, need to be changed from config file
 
     remote_debugging_url = f"http://{host}:{port}"
@@ -278,6 +439,25 @@ def get_shortcuts_on_desktop(env, config: Dict[str, str]):
 
     for shortcut_path in shortcuts_paths:
         short_cuts[shortcut_path] = env.controller.get_file(env.controller.execute_python_command(
-            f"import os; print(os.path.join(os.path.expanduser('~'), 'Desktop', '{shortcut_path}'))")['output'].strip()).decode('utf-8')
+            f"import os; print(os.path.join(os.path.expanduser('~'), 'Desktop', '{shortcut_path}'))")[
+                                                                'output'].strip()).decode('utf-8')
 
     return short_cuts
+
+
+def get_number_of_search_results(env, config: Dict[str, str]):
+    # todo: move into the config file
+    url, result_selector = "https://google.com/search?q=query", '.search-result'
+    host = env.vm_ip
+    port = 9222  # fixme: this port is hard-coded, need to be changed from config file
+
+    remote_debugging_url = f"http://{host}:{port}"
+    with sync_playwright() as p:
+        browser = p.chromium.connect_over_cdp(remote_debugging_url)
+        page = browser.new_page()
+        page.goto(url)
+        search_results = page.query_selector_all(result_selector)
+        actual_count = len(search_results)
+        browser.close()
+
+    return actual_count

desktop_env/evaluators/metrics/chrome.py

@@ -95,3 +95,45 @@ def is_shortcut_on_desktop(shortcuts: Dict[str, str], rule):
         raise TypeError(f"{rule['type']} not support yet!")
     else:
         raise TypeError(f"{rule['type']} not support yet!")
+
+
+def check_history_deleted(history_data, rule):
+    """
+    Check if the history is deleted.
+    """
+
+    if rule['type'] == 'keywords':
+        history_domains = [history[0] for history in history_data]
+        for keyword in rule['keywords']:
+            for history_domain in history_domains:
+                if keyword in history_domain:
+                    return 0.
+        return 1.
+    else:
+        raise TypeError(f"{rule['type']} not support yet!")
+
+
+def check_enabled_experiments(enabled_experiments, rule):
+    """
+    Check if the enabled experiments are as expected.
+    """
+    enabled_experiments_names = [experiment.split("@")[0] for experiment in enabled_experiments]
+
+    if rule['type'] == 'names':
+        return 1. if enabled_experiments_names == rule['names'] else 0.
+    else:
+        raise TypeError(f"{rule['type']} not support yet!")
+
+
+def check_font_size(font_size, rule):
+    """
+    Check if the font size is as expected.
+    """
+
+    default_font_size = font_size['default_font_size']
+    if rule['type'] == 'value':
+        return 1. if default_font_size == rule['value'] else 0.
+    elif rule['type'] == 'range':
+        return 1. if rule['min'] < default_font_size < rule['max'] else 0.
+    else:
+        raise TypeError(f"{rule['type']} not support yet!")