ck_pro/ck_web/utils.py

#

# utils for our web-agent

import re
import os
import subprocess
import signal
import time
import requests
import base64
import markdownify
from ..agents.utils import KwargsInitializable, rprint, zwarn, zlog

# --
# web state
class WebState:
    def __init__(self, **kwargs):
        # not-changed
        self.browser_id = ""
        self.page_id = ""
        self.target_url = ""
        # from tree-results
        self.get_accessibility_tree_succeed = False
        self.current_accessibility_tree = ""
        self.step_url = ""
        self.html_md = ""
        self.snapshot = ""
        self.boxed_screenshot = ""  # always store the screenshot here
        self.downloaded_file_path = []
        self.current_has_cookie_popup = False
        self.expanded_part = None
        # step info
        self.curr_step = 0  # step to the root
        self.curr_screenshot_mode = False  # whether we are using screenshot or not?
        self.total_actual_step = 0  # [no-rev] total actual steps including reverting (can serve as ID)
        self.num_revert_state = 0  # [no-rev] number of state reversion
        # (last) action information
        self.action_string = ""
        self.action = None
        self.error_message = ""
        # --
        self.update(**kwargs)

    def get_id(self):  # use these as ID
        return (self.browser_id, self.page_id, self.total_actual_step)

    def update(self, **kwargs):
        for k, v in kwargs.items():
            assert (k in self.__dict__), f"Attribute not found for {k} <- {v}"
        self.__dict__.update(**kwargs)

    def to_dict(self):
        return self.__dict__.copy()

    def copy(self):
        return WebState(**self.to_dict())

    def __repr__(self):
        return f"WebState({self.__dict__})"

# --
class MyMarkdownify(markdownify.MarkdownConverter):
    def convert_img(self, el, text, parent_tags):
        return ""  # simply ignore image

    def convert_a(self, el, text, parent_tags):
        if (not text) or (not text.strip()):
            return ""  # empty
        text = text.strip()  # simply strip!
        href = el.get("href")
        if not href:
            href = ""
        if not any(href.startswith(z) for z in ["http", "https"]):
            ret = text  # simply no links
            # ret = ""  # more aggressively remove things! (nope, removing too much...)
        else:
            ret = f"[{text}]({href})"
        return ret

    @staticmethod
    def md_convert(html: str):
        html_md = MyMarkdownify().convert(html)
        valid_lines = []
        for line in html_md.split("\n"):
            line = line.rstrip()
            if not line: continue
            valid_lines.append(line)
        ret = "\n".join(valid_lines)
        return ret

    @classmethod
    def create_from_dict(cls, data):
        """Create WebState instance from dictionary"""
        return cls(**data)

# an opened web browser
class WebEnv(KwargsInitializable):
    def __init__(self, settings=None, starting=True, starting_target_url=None, logger=None, **kwargs):
        # Use configuration from settings - unified web config from [web.env]
        if settings and hasattr(settings, 'web') and hasattr(settings.web, 'env'):
            self.web_ip = settings.web.env.web_ip
            self.web_command = settings.web.env.web_command
            self.web_timeout = settings.web.env.web_timeout
            self.screenshot_boxed = settings.web.env.screenshot_boxed
            self.target_url = settings.web.env.target_url
        else:
            # Fallback defaults if no settings provided
            self.web_ip = "localhost:3000"
            self.web_command = ""
            self.web_timeout = 600
            self.screenshot_boxed = True
            self.target_url = "https://www.bing.com/"
        self.web_ip = settings.web.env.web_ip  # use TOML config from [web.env]
        self.web_command = settings.web.env.web_command  # use TOML config
        self.web_timeout = settings.web.env.web_timeout  # use TOML config
        # self.use_screenshot = False  # add screenshot? -> for simplicity, always store it!
        self.screenshot_boxed = settings.web.env.screenshot_boxed  # use TOML config
        # self.target_url = "https://duckduckgo.com/"  # by default
        self.target_url = settings.web.env.target_url  # use TOML config
        # self.target_url = "https://duckduckgo.com/"  # by default
        self.logger = logger  # 诊断日志器
        # --
        super().__init__(**kwargs)
        # --
        self.state: WebState = None
        self.popen = None  # popen obj for subprocess running
        if starting:
            self.start(starting_target_url)  # start at the beginning
        # --

    def start(self, target_url=None):
        self.stop()  # stop first
        # --
        # optionally start one
        if self.web_command:
            self.popen = subprocess.Popen(self.web_command, shell=True, preexec_fn=os.setsid)  # make a new one
            time.sleep(15)  # wait for some time
            rprint(f"Web-Utils-Start {self.popen}")
        # --
        target_url = target_url if target_url is not None else self.target_url  # otherwise use default
        ### hard code: replace google to bing
        if 'www.google.com' in target_url:
            if not 'www.google.com/maps' in target_url:
                target_url = target_url.replace('www.google.com', 'www.bing.com')
        self.init_state(target_url)

    def stop(self):
        if self.state is not None:
            self.end_state()
            self.state = None
        if self.popen is not None:
            os.killpg(self.popen.pid, signal.SIGKILL)  # kill the PG
            self.popen.kill()
            time.sleep(1)  # slightly wait
            rprint(f"Web-Utils-Kill {self.popen} with {self.popen.poll()}")
            self.popen = None

    def __del__(self):
        self.stop()

    # note: return a copy!
    def get_state(self, export_to_dict=True, return_copy=True):
        assert self.state is not None, "Current state is None, should first start it!"
        if export_to_dict:
            ret = self.state.to_dict()
        elif return_copy:
            ret = self.state.copy()
        else:
            ret = self.state
        return ret

    def get_target_url(self):
        return self.target_url

    # --
    # helpers

    def get_browser(self, storage_state, geo_location):
        url = f"http://{self.web_ip}/getBrowser"
        data = {"storageState": storage_state, "geoLocation": geo_location}

        # 埋点：获取浏览器请求
        if self.logger:
            self.logger.info("[WEB_HTTP] Get_Browser_Request: %s", url)
            self.logger.debug("[WEB_HTTP] Get_Browser_Data: %s", data)

        response = requests.post(url, json=data, timeout=self.web_timeout)

        if response.status_code == 200:
            browser_data = response.json()
            zlog(f"==> Get browser {browser_data}")
            # 埋点：获取浏览器成功
            if self.logger:
                self.logger.info("[WEB_HTTP] Get_Browser_Success: %s", browser_data)
            return browser_data["browserId"]
        else:
            # 埋点：获取浏览器失败
            if self.logger:
                self.logger.error("[WEB_HTTP] Get_Browser_Failed: Status: %s | Response: %s",
                                response.status_code, response.text)
            raise requests.RequestException(f"Getting browser failed: {response}")

    def close_browser(self, browser_id):
        url = f"http://{self.web_ip}/closeBrowser"
        data = {"browserId": browser_id}
        zlog(f"==> Closing browser {browser_id}")
        try:  # put try here
            response = requests.post(url, json=data, timeout=self.web_timeout)
            if response.status_code == 200:
                return None
            else:
                zwarn(f"Bad response when closing browser: {response}")
        except requests.RequestException as e:
            zwarn(f"Request Error: {e}")
        return None

    def open_page(self, browser_id, target_url):
        url = f"http://{self.web_ip}/openPage"
        data = {"browserId": browser_id, "url": target_url}

        # 埋点：打开页面请求
        if self.logger:
            self.logger.info("[WEB_HTTP] Open_Page_Request: %s", url)
            self.logger.info("[WEB_HTTP] Open_Page_Data: Browser: %s | Target: %s", browser_id, target_url)

        response = requests.post(url, json=data, timeout=self.web_timeout)

        if response.status_code == 200:
            page_data = response.json()
            # 埋点：打开页面成功
            if self.logger:
                self.logger.info("[WEB_HTTP] Open_Page_Success: %s", page_data)
            return page_data["pageId"]
        else:
            # 埋点：打开页面失败
            if self.logger:
                self.logger.error("[WEB_HTTP] Open_Page_Failed: Status: %s | Response: %s",
                                response.status_code, response.text)
            raise requests.RequestException(f"Open page Request failed: {response}")

    def goto_url(self, browser_id, page_id, target_url):
        url = f"http://{self.web_ip}/gotoUrl"
        data = {"browserId": browser_id, "pageId": page_id, "targetUrl": target_url}
        response = requests.post(url, json=data, timeout=self.web_timeout)
        if response.status_code == 200:
            return True
        else:
            raise requests.RequestException(f"GOTO page Request failed: {response}")

    def process_html(self, html: str):
        if not html.strip():
            return html  # empty
        return MyMarkdownify.md_convert(html)

    def process_axtree(self, res_json):
        # --
        def _parse_tree_str(_s):
            if "[2]" in _s:
                _lines = _s.split("[2]", 1)[1].split("\n")
                _lines = [z for z in _lines if z.strip().startswith("[")]
                _lines = [" ".join(z.split()[1:]) for z in _lines]
                return _lines
            else:
                return []
        # --
        def _process_tree_str(_s):
            _s = _s.strip()
            if _s.startswith("Tab 0 (current):"):  # todo(+N): sometimes this line can be strange, simply remove it!
                _s = _s.split("\n", 1)[-1].strip()
            return _s
        # --
        html_md = self.process_html(res_json.get("html", ""))
        AccessibilityTree = _process_tree_str(res_json.get("yaml", ""))
        curr_url = res_json.get("url", "")
        snapshot = res_json.get("snapshot", "")
        fulltree = _process_tree_str(res_json.get("fulltree", ""))
        screenshot = res_json.get("boxed_screenshot", "") if self.screenshot_boxed else res_json.get("nonboxed_screenshot", "")
        downloaded_file_path = res_json.get("downloaded_file_path", [])
        all_at, all_ft = _parse_tree_str(AccessibilityTree), _parse_tree_str(fulltree)
        # all_ft_map = {v: i for i, v in enumerate(all_ft)}
        all_ft_map = {}
        for ii, vv in enumerate(all_ft):
            if vv not in all_ft_map:  # no overwritten to get the minumum one
                all_ft_map[vv] = ii
        _hit_at_idxes = [all_ft_map[z] for z in all_at if z in all_ft_map]
        if _hit_at_idxes:
            _last_hit_idx = max(_hit_at_idxes)
            _remaining = len(all_ft) - (_last_hit_idx + 1)
            if _remaining >= len(_hit_at_idxes) * 0.5:  # note: a simple heuristic
                AccessibilityTree = AccessibilityTree.strip() + "\n(* Scroll down to see more items)"
        # --
        ret = {"current_accessibility_tree": AccessibilityTree, "step_url": curr_url, "html_md": html_md, "snapshot": snapshot, "boxed_screenshot": screenshot, "downloaded_file_path": downloaded_file_path}
        return ret

    def get_accessibility_tree(self, browser_id, page_id, current_round):
        url = f"http://{self.web_ip}/getAccessibilityTree"
        data = {
            "browserId": browser_id,
            "pageId": page_id,
            "currentRound": current_round,
        }
        default_axtree = ""  # default empty
        default_res = {"current_accessibility_tree": default_axtree, "step_url": "", "html_md": "", "snapshot": "", "boxed_screenshot": "", "downloaded_file_path": []}
        try:
            response = requests.post(url, json=data, timeout=self.web_timeout)
            if response.status_code == 200:
                res_json = response.json()
                res_dict = self.process_axtree(res_json)
                return True, res_dict
            else:
                zwarn(f"Get accessibility tree Request failed with status code: {response.status_code}")
                return False, default_res
        except requests.RequestException as e:
            zwarn(f"Request failed: {e}")
            return False, default_res

    def action(self, browser_id, page_id, action):
        url = f"http://{self.web_ip}/performAction"
        data = {
            "browserId": browser_id,
            "pageId": page_id,
            "actionName": action["action_name"],
            "targetId": action["target_id"],
            "targetElementType": action["target_element_type"],
            "targetElementName": action["target_element_name"],
            "actionValue": action["action_value"],
            "needEnter": action["need_enter"],
        }

        # 埋点：HTTP 请求详情
        if self.logger:
            self.logger.info("[WEB_HTTP] Request_URL: %s", url)
            self.logger.info("[WEB_HTTP] Request_Data: %s", data)
            self.logger.debug("[WEB_HTTP] Timeout: %s seconds", self.web_timeout)

        try:
            response = requests.post(url, json=data, timeout=self.web_timeout)

            # 埋点：HTTP 响应详情
            if self.logger:
                self.logger.info("[WEB_HTTP] Response_Status: %s", response.status_code)
                if response.status_code != 200:
                    self.logger.error("[WEB_HTTP] Response_Text: %s", response.text)

            if response.status_code == 200:
                return True
            else:
                zwarn(f"Request failed with status code: {response.status_code} {response.text}")
                return False
        except requests.RequestException as e:
            # 埋点：HTTP 请求异常
            if self.logger:
                self.logger.error("[WEB_HTTP] Request_Exception: %s", str(e))
            zwarn(f"Request failed: {e}")
            return False

    # --
    # other helpers

    def is_annoying(self, current_accessbility_tree):
        if "See results closer to you?" in current_accessbility_tree and len(current_accessbility_tree.split("\n")) <= 10:
            return True
        return False

    def parse_action_string(self, action_string: str, state):
        patterns = {"click": r"click\s+\[?(\d+)\]?", "type": r"type\s+\[?(\d+)\]?\s+\{?(.+)\}?", "scroll": r"scroll\s+(down|up)", "wait": "wait", "goback": "goback", "restart": "restart", "stop": r"stop(.*)", "goto": r"goto(.*)", "save": r"save(.*)", "screenshot": r"screenshot(.*)", "nop": r"nop(.*)"}
        action = {"action_name": "", "target_id": None, "action_value": None, "need_enter": None, "target_element_type": None, "target_element_name": None}  # assuming these fields
        if action_string:
            for key, pat in patterns.items():
                m = re.match(pat, action_string, flags=(re.IGNORECASE|re.DOTALL))  # ignore case and allow \n
                if m:
                    action["action_name"] = key
                    if key in ["click", "type"]:
                        action["target_id"] = m.groups()[0]  # target ID
                    if key in ["type", "scroll", "stop", "goto", "save", "screenshot"]:
                        action["action_value"] = m.groups()[-1].strip()  # target value
                        if key == "type":  # quick fix
                            action["action_value"] = action["action_value"].rstrip("}]").rstrip().strip("\"'").strip()
                    # if key == "restart":
                    #     action["action_value"] = state.target_url  # restart
                    break
        return action

    @staticmethod
    def find_target_element_info(current_accessibility_tree, target_id, action_name):
        if target_id is None:
            return None, None, None
        if action_name == "type":
            tree_to_check = current_accessibility_tree.split("\n")[int(target_id) - 1:]
            for i, line in enumerate(tree_to_check):
                if f"[{target_id}]" in line and ("combobox" in line or "box" not in line):
                    num_tabs = len(line) - len(line.lstrip("\t"))
                    for j in range(i + 1, len(tree_to_check)):
                        curr_num_tabs = len(tree_to_check[j]) - len(tree_to_check[j].lstrip("\t"))
                        if curr_num_tabs <= num_tabs:
                            break
                        if "textbox" in tree_to_check[j] or "searchbox" in tree_to_check[j]:
                            target_element_id = tree_to_check[j].split("]")[0].strip()[1:]
                            # print("CATCHED ONE MISSED TYPE ACTION, changing the type action to", target_element_id)
                            target_id = target_element_id
        target_pattern = r"\[" + re.escape(target_id) + r"\] ([a-z]+) '(.*)'"
        matches = re.finditer(target_pattern, current_accessibility_tree, re.IGNORECASE)
        for match in matches:
            target_element_type, target_element_name = match.groups()
            return target_id, target_element_type, target_element_name
        return target_id, None, None

    @staticmethod
    def get_skip_action(current_accessbility_tree):
        # action_name, target_id, action_value, need_enter = extract_info_from_action("click [5]")
        action_name, target_id, action_value, need_enter = "click", "5", "", None
        target_id, target_element_type, target_element_name = WebEnv.find_target_element_info(current_accessbility_tree, target_id, action_name)
        return {
            "action_name": action_name,
            "target_id": target_id,
            "action_value": action_value,
            "need_enter": need_enter,
            "target_element_type": target_element_type,
            "target_element_name": target_element_name,
        }

    @staticmethod
    def check_if_menu_is_expanded(accessibility_tree, snapshot):
        node_to_expand = {}
        lines = accessibility_tree.split("\n")
        for i, line in enumerate(lines):
            if 'hasPopup: menu' in line and 'expanded: true' in line:
                num_tabs = len(line) - len(line.lstrip("\t"))
                next_tabs = len(lines[i + 1]) - len(lines[i + 1].lstrip("\t"))
                if next_tabs <= num_tabs:
                    # In this case, the menu should be expanded but is not present in the tree
                    target_pattern = r"\[(\d+)\] ([a-z]+) '(.*)'"
                    matches = re.finditer(target_pattern, line, re.IGNORECASE)
                    target_id = None
                    target_element_type = None
                    target_element_name = None
                    for match in matches:
                        target_id, target_element_type, target_element_name = match.groups()
                        break
                    if target_element_type is not None:
                        # locate the menu items from the snapshot instead
                        children = WebEnv.find_node_with_children(snapshot, target_element_type, target_element_name)
                        if children is not None:
                            node_to_expand[i] = (num_tabs + 1, children, target_id, target_element_type, target_element_name)
        new_lines = []
        curr = 1
        if len(node_to_expand) == 0:
            return accessibility_tree, None
        expanded_part = {}
        # add the menu items to the correct location in the tree
        for i, line in enumerate(lines):
            if not line.strip().startswith('['):
                new_lines.append(line)
                continue
            num_tabs = len(line) - len(line.lstrip("\t"))
            content = line.split('] ')[1]
            new_lines.append('\t' * num_tabs + f"[{curr}] {content}")
            curr += 1
            if i in node_to_expand:
                for child in node_to_expand[i][1]:
                    child_content = f"{child.get('role', '')} '{child.get('name', '')}' " + ' '.join([f"{k}: {v}" for k, v in child.items() if k not in ['role', 'name']])
                    tabs = '\t' * node_to_expand[i][0]
                    new_lines.append(f"{tabs}[{curr}] {child_content}")
                    expanded_part[curr] = (node_to_expand[i][2], node_to_expand[i][3], node_to_expand[i][4])
                    curr += 1
        return '\n'.join(new_lines), expanded_part

    @staticmethod
    def find_node_with_children(node, target_role, target_name):
        # Check if the current node matches the target role and name
        if node.get('role') == target_role and node.get('name') == target_name:
            return node.get('children', None)
        # If the node has children, recursively search through them
        children = node.get('children', [])
        for child in children:
            result = WebEnv.find_node_with_children(child, target_role, target_name)
            if result is not None:
                return result
        # If no matching node is found, return None
        return None

    # --
    # main step

    def init_state(self, target_url: str):
        # 埋点：开始初始化浏览器状态
        if self.logger:
            self.logger.info("[WEB_INIT] Starting browser initialization")
            self.logger.info("[WEB_INIT] Target_URL: %s", target_url)
            self.logger.info("[WEB_INIT] Web_IP: %s", self.web_ip)

        browser_id = self.get_browser(None, None)

        # 埋点：浏览器创建成功
        if self.logger:
            self.logger.info("[WEB_INIT] Browser_Created: %s", browser_id)

        page_id = self.open_page(browser_id, target_url)

        # 埋点：页面打开成功
        if self.logger:
            self.logger.info("[WEB_INIT] Page_Opened: %s", page_id)

        curr_step = 0
        state = WebState(browser_id=browser_id, page_id=page_id, target_url=target_url, curr_step=curr_step, total_actual_step=curr_step)  # start from 0
        results = self._get_accessibility_tree_results(state)
        state.update(**results)  # update it!

        # 埋点：状态初始化完成
        if self.logger:
            actual_url = getattr(state, 'step_url', 'unknown')
            self.logger.info("[WEB_INIT] State_Initialized: Actual_URL: %s", actual_url)
            if actual_url != target_url:
                self.logger.warning("[WEB_INIT] URL_Mismatch: Expected: %s | Actual: %s", target_url, actual_url)

        # --
        self.state = state  # set the new state!
        # --

    def end_state(self):
        state = self.state
        self.close_browser(state.browser_id)

    def reset_to_state(self, target_state):
        state = self.state
        if isinstance(target_state, dict):
            target_state = WebState.create_from_dict(target_state)
        # assert state.browser_id == target_state.browser_id and state.page_id == target_state.page_id, "Mismatched basic IDs"
        if state.get_id() != target_state.get_id():  # need to revert to another URL
            self.goto_url(target_state.browser_id, target_state.page_id, target_state.step_url)
            state.update(browser_id=target_state.browser_id, page_id=target_state.page_id)
            results = self._get_accessibility_tree_results(state)
            state.update(**results)  # update it!
            # --
            # revert other state info
            state.update(curr_step=target_state.curr_step, action_string=target_state.action_string, action=target_state.action, error_message=target_state.error_message)  # no change of total_step!
            state.num_revert_state += 1
            # --
            zlog(f"Reset state with URL={target_state.step_url}")
            return True
        else:
            assert state.to_dict() == target_state.to_dict(), "Mismatched state!"
            zlog("No need for state resetting!")
            return False
        # --

    def _get_accessibility_tree_results(self, state):
        get_accessibility_tree_succeed, curr_res = self.get_accessibility_tree(state.browser_id, state.page_id, state.curr_step)
        current_accessibility_tree = curr_res.get("current_accessibility_tree", "")
        if not get_accessibility_tree_succeed:
            zwarn("Failed to get current_accessibility_tree!!")
        if self.is_annoying(current_accessibility_tree):
            skip_this_action = self.get_skip_action(current_accessibility_tree)
            self.action(state.browser_id, state.page_id, skip_this_action)
            get_accessibility_tree_succeed, curr_res = self.get_accessibility_tree(state.browser_id, state.page_id, state.curr_step)
        # try to close cookie popup
        if "Cookie banner" in current_accessibility_tree:
            current_has_cookie_popup = True  # note: only mark here!
        else:
            current_has_cookie_popup = False
        current_accessibility_tree, expanded_part = self.check_if_menu_is_expanded(current_accessibility_tree, curr_res["snapshot"])
        # --
        # if (not self.use_screenshot) and ("boxed_screenshot" in curr_res):  # note: no storing of snapshot since it is too much
        #     del curr_res["boxed_screenshot"]  # for simplicity, always store it
        # --
        # more checking on axtree
        if not current_accessibility_tree or ("[2]" not in current_accessibility_tree):  # at least we should have some elements!
            curr_res["current_accessibility_tree"] = current_accessibility_tree + "\n**Warning**: The accessibility tree is currently unavailable. Please try some alternative actions. If the issue persists after multiple attempts, consider goback or restart."
        # --
        curr_res.update(get_accessibility_tree_succeed=get_accessibility_tree_succeed, current_has_cookie_popup=current_has_cookie_popup, expanded_part=expanded_part)
        return curr_res

    def step_state(self, action_string: str):
        state = self.state

        # 埋点：WebEnv 开始执行动作
        if self.logger:
            self.logger.info("[WEB_ENV] Step_State_Start: %s", action_string)
            self.logger.debug("[WEB_ENV] Current_URL: %s", getattr(state, 'step_url', 'unknown'))

        # --
        need_enter = True
        if "[NOENTER]" in action_string:
            need_enter = False
            action_string = action_string.replace("[NOENTER]", "")  # note: ugly quick fix ...
        # --
        action_string = action_string.strip()
        # parse action
        action = self.parse_action_string(action_string, state)

        # 埋点：动作解析结果
        if self.logger:
            self.logger.info("[WEB_ENV] Parsed_Action: %s", action)
        if action["action_name"]:
            if action["action_name"] in ["click", "type"]:  # need more handling
                target_id, target_element_type, target_element_name = self.find_target_element_info(state.current_accessibility_tree, action["target_id"], action["action_name"])
                if state.expanded_part and int(target_id) in state.expanded_part:
                    expand_target_id, expand_target_type, expand_target_name = state.expanded_part[int(target_id)]
                    action.update({"action_name": "select", "target_id": expand_target_id, "action_value": target_element_name, "target_element_type": expand_target_type, "target_element_name": expand_target_name})
                else:
                    action.update({"target_id": target_id, "target_element_type": target_element_type, "target_element_name": target_element_name})
            if action["action_name"] == "type":
                action["need_enter"] = need_enter
        zlog(f"[CallWeb:{state.curr_step}:{state.total_actual_step}] ACTION={action} ACTION_STR={action_string}", timed=True)
        # --
        # execution
        state.curr_step += 1
        state.total_actual_step += 1
        state.update(action=action, action_string=action_string, error_message="")  # first update some of the things
        if not action["action_name"]:  # UNK action
            state.error_message = f"The action you previously choose is not well-formatted: {action_string}. Please double-check if you have selected the correct element or used correct action format."
            ret = state.error_message
            # 埋点：动作格式错误
            if self.logger:
                self.logger.error("[WEB_ENV] Action_Parse_Error: %s", action_string)
        elif action["action_name"] in ["stop", "save", "nop"]:  # ok, nothing to do
            ret = f"Browser step: {action_string}"
            # 埋点：简单动作执行
            if self.logger:
                self.logger.info("[WEB_ENV] Simple_Action: %s", action["action_name"])
        elif action["action_name"] == "screenshot":
            _old_mode = state.curr_screenshot_mode
            _fields = action["action_value"].split() + [""] * 2
            _new_mode = _fields[0].lower() in ["1", "true", "yes"]
            _save_path = _fields[1].strip()
            if _save_path:
                try:
                    assert state.boxed_screenshot.strip(), "Screenshot not available!"
                    file_bytes = base64.b64decode(state.boxed_screenshot)
                    _dir = os.path.dirname(_save_path)
                    if _dir:
                        os.makedirs(_dir, exist_ok=True)
                    with open(_save_path, 'wb') as fd:
                        fd.write(file_bytes)
                    save_info = f" (Current screenshot saved to {_save_path}.)"
                except Exception as e:
                    save_info = f" (Error {e} when saving screenshot.)"
            else:
                save_info = ""
            state.curr_screenshot_mode = _new_mode
            ret = f"Browser step: {action_string} -> Changing curr_screenshot_mode from {_old_mode} to {_new_mode}" + save_info
        else:
            # actually perform action
            # 埋点：即将执行浏览器动作
            if self.logger:
                self.logger.info("[WEB_ENV] Executing_Browser_Action: %s | Browser_ID: %s | Page_ID: %s",
                                action["action_name"], state.browser_id, state.page_id)

            action_succeed = self.action(state.browser_id, state.page_id, action)

            if not action_succeed:  # no succeed
                state.error_message = f"The action you have chosen cannot be executed: {action_string}. Please double-check if you have selected the correct element or used correct action format."
                ret = state.error_message
                # 埋点：浏览器动作执行失败
                if self.logger:
                    self.logger.error("[WEB_ENV] Browser_Action_Failed: %s", action_string)
            else:  # get new states
                # 埋点：浏览器动作执行成功，获取新状态
                if self.logger:
                    self.logger.info("[WEB_ENV] Browser_Action_Success: %s", action_string)
                    self.logger.debug("[WEB_ENV] Getting_New_State...")

                results = self._get_accessibility_tree_results(state)
                state.update(**results)  # update it!
                ret = f"Browser step: {action_string}"

                # 埋点：状态更新完成
                if self.logger:
                    new_url = getattr(state, 'step_url', 'unknown')
                    self.logger.info("[WEB_ENV] State_Updated: New_URL: %s", new_url)
        return ret
        # --

    # sync files between remote and local dirs
    def sync_files(self):
        # --
        def _get_file(_f: str):
            url = f"http://{self.web_ip}/getFile"
            data = {"filename": _f}
            try:
                response = requests.post(url, json=data, timeout=self.web_timeout)
                if response.status_code == 200:
                    res_json = response.json()
                    base64_str = res_json["file"]
                    file_bytes = base64.b64decode(base64_str)
                    if _f:
                        _dir = os.path.dirname(_f)
                        if _dir:
                            os.makedirs(_dir, exist_ok=True)
                    with open(_f, 'wb') as fd:  # Change output filename as needed
                        fd.write(file_bytes)
                    return True
                else:
                    zwarn(f"Get file failed with status code: {response.status_code}")
                    return False
            except Exception as e:
                zwarn(f"Request failed: {e}")
                return False
        # --
        files = {}
        for file in self.state.downloaded_file_path:
            if not os.path.exists(file):
                fres = _get_file(file)
                files[file] = f"Get[res={fres}]"
            else:
                files[file] = "Exist"
        zlog(f"Sync files: {files}")

    def screenshot_mode(self, flag=None):
        old_mode = self.state.curr_screenshot_mode
        new_mode = old_mode
        if flag is not None:  # set as flag
            self.state.curr_screenshot_mode = flag
        return old_mode, new_mode