BuiltInFunctions.py

"""Synchronous Playwright counterparts for ZeuZ Selenium actions."""

import ast
import asyncio
import base64
import functools
import inspect
import json
import socket
import sys
import time
import threading
from pathlib import Path
from urllib.parse import urldefrag, urlparse

from Framework.Built_In_Automation.Shared_Resources import LocateElement
from Framework.Built_In_Automation.Shared_Resources import (
    BuiltInFunctionSharedResources as sr,
)
from Framework.Built_In_Automation.Sequential_Actions.action_declarations.playwright import (
    declarations,
)
from Framework.Utilities import CommonUtil, ConfigModule
from Framework.Utilities.CommonUtil import failed_tag_list


MODULE_NAME = inspect.getmodulename(__file__)
temp_config = str(
    Path(__file__).resolve().parents[4]
    / "AutomationLog"
    / ConfigModule.get_config_value("Advanced Options", "_file")
)


class _BrowserState(threading.local):
    """Keep sync Playwright objects on their action or performance thread."""

    def __init__(self):
        self.playwright_details = {}
        self.playwright_page = None
        self.current_driver_id = None
        self._playwright = None


_browser_state = _BrowserState()


def _log(message, level=1):
    CommonUtil.ExecLog(MODULE_NAME, message, level)


def _fail(message):
    _log(message, 3)
    return "zeuz_failed"


_DIAGNOSTIC_SNAPSHOT = """element => {
    const describe = el => {
        if (!el || el.nodeType !== 1) return null;
        const info = {};
        for (const name of ['id', 'class', 'type', 'role', 'aria-expanded', 'aria-label',
                            'aria-selected', 'aria-checked', 'aria-busy'])
            info[name] = (el.getAttribute(name) || '').slice(0, 200);
        info.tag = el.tagName;
        if ('checked' in el) info.checked = el.checked;
        if ('value' in el && el.type !== 'password') info.value_length = el.value.length;
        return info;
    };
    if (!window.__zeuzPlaywrightDiagnostics) {
        const events = [];
        window.__zeuzPlaywrightDiagnostics = events;
        for (const type of ['pointerdown', 'mousedown', 'mouseup', 'click', 'dblclick', 'focusin', 'focusout', 'input', 'change'])
            document.addEventListener(type, event => {
                events.push({time: Date.now(), type, target: describe(event.target),
                             parent: describe(event.target.parentElement),
                             trusted: event.isTrusted, detail: event.detail,
                             input_type: event.inputType || null});
                if (events.length > 30) events.shift();
            }, true);
    }
    const rect = element ? element.getBoundingClientRect() : null;
    const overlays = [...document.querySelectorAll(
        '[role="dialog"], [role="listbox"], .p-overlaypanel, .popover'
    )].filter(el => el.getClientRects().length && getComputedStyle(el).visibility !== 'hidden');
    return {
        time: Date.now(), frame: location.origin + location.pathname,
        viewport: {width: innerWidth, height: innerHeight, scroll_x: scrollX, scroll_y: scrollY},
        target: describe(element), parent: describe(element && element.parentElement),
        rect: rect ? rect.toJSON() : null,
        hit_at_center: rect ? describe(document.elementFromPoint(
            rect.x + rect.width / 2, rect.y + rect.height / 2)) : null,
        active: describe(document.activeElement),
        hovered: [...document.querySelectorAll(':hover')].slice(-5).map(describe),
        overlay_count: overlays.length, overlays: overlays.slice(0, 10).map(describe),
        controls: [...document.querySelectorAll(
            'input[type="checkbox"], [role="switch"], [aria-busy="true"]'
        )].slice(0, 30).map(describe),
        events: window.__zeuzPlaywrightDiagnostics
    };
}"""


def _diagnostic(stage, element=None):
    """Opt in with zeuz_playwright_diagnostics=true or comma-separated test case IDs."""
    try:
        setting = str(sr.Get_Shared_Variables("zeuz_playwright_diagnostics", log=False)).strip()
        if (setting.lower() not in CommonUtil.affirmative_words
                and (not CommonUtil.current_tc_no
                     or CommonUtil.current_tc_no not in [case.strip() for case in setting.split(",")])):
            return
        record = {"stage": stage, "case": CommonUtil.current_tc_no,
                  "step_id": CommonUtil.current_step_id, "step": CommonUtil.current_step_no,
                  "action": CommonUtil.current_action_no}
        if hasattr(element, "evaluate"):
            try:
                if LocateElement._is_playwright_locator(element):
                    record["target"] = (element.evaluate(_DIAGNOSTIC_SNAPSHOT, timeout=500)
                                        if element.count() else {"missing": True})
                else:
                    record["target"] = element.evaluate(_DIAGNOSTIC_SNAPSHOT)
                box = (element.bounding_box(timeout=500) if LocateElement._is_playwright_locator(element)
                       else element.bounding_box())
                record["main_rect"] = box
                if box:
                    record["main_hit_at_center"] = get_page().evaluate("""point => {
                        const el = document.elementFromPoint(point.x, point.y);
                        return el ? {tag: el.tagName, id: el.id, class: el.getAttribute('class')} : null;
                    }""", {"x": box["x"] + box["width"] / 2, "y": box["y"] + box["height"] / 2})
            except Exception as error:
                record["target"] = {"unavailable": type(error).__name__}
        record["frames"] = []
        for frame in get_page().frames:
            try:
                record["frames"].append(frame.evaluate(_DIAGNOSTIC_SNAPSHOT))
            except Exception as error:
                record["frames"].append({"unavailable": type(error).__name__})
        _log("PW_DIAGNOSTIC " + json.dumps(record, ensure_ascii=False))
    except Exception:
        # Diagnostics must never change the action result, including during teardown.
        pass


def _rows(data_set):
    return [
        (str(left).strip(), str(middle).strip().lower(), str(right))
        for left, middle, right in data_set
    ]


def _key(value):
    return value.replace(" ", "").replace("_", "").replace("-", "").lower()


def _action(data_set, default=""):
    return next(
        (right for _left, middle, right in _rows(data_set) if "action" in middle),
        default,
    )


def _driver_id(data_set, default=None):
    for left, _middle, right in _rows(data_set):
        if _key(left) in ("driverid", "drivertag"):
            return right.strip() or "default"
    return default or _browser_state.current_driver_id or "default"


def _state():
    if _browser_state.current_driver_id not in _browser_state.playwright_details:
        raise RuntimeError(
            "No active Playwright browser. Run 'open browser' or 'go to link' first."
        )
    return _browser_state.playwright_details[_browser_state.current_driver_id]


def get_driver():
    state = _state()
    return state.get("frame") or state["page"]


def get_page():
    return _state()["page"]


def get_dom(_data_set=()):
    return get_page().evaluate("""() => {
        const html = document.documentElement.cloneNode(true);
        html.setAttribute('zeuz', 'aiplugin');
        html.querySelectorAll('head,link,script,style').forEach(node => node.remove());
        return html.outerHTML.replace(/[\\x00-\\x08\\x0B-\\x1F\\x7F]/g, '');
    }""")


def _element(data_set, all_elements=False, root=None):
    return LocateElement.Get_Element(
        data_set, root or get_driver(), return_all_elements=all_elements
    )


def _set_active(driver_id):
    _browser_state.current_driver_id = driver_id
    state = _browser_state.playwright_details[driver_id]
    bridge = state.get("selenium_bridge")
    if bridge is not None:
        session = state["context"].new_cdp_session(state["page"])
        try:
            target_id = session.send("Target.getTargetInfo")["targetInfo"]["targetId"]
        finally:
            session.detach()
        bridge.switch_to.window(target_id)
    _browser_state.playwright_page = state["page"]
    sr.Set_Shared_Variables("playwright_page", _browser_state.playwright_page)
    sr.Set_Shared_Variables("common_driver", state.get("frame") or _browser_state.playwright_page)
    _publish_selenium_bridge(state.get("selenium_bridge"))
    sr.Set_Shared_Variables("zeuz_active_browser_backend", "playwright")
    owners = sr.Get_Shared_Variables("zeuz_browser_backends", log=False)
    owners = owners if isinstance(owners, dict) else {}
    owners[driver_id] = "playwright"
    sr.Set_Shared_Variables("zeuz_browser_backends", owners)
    CommonUtil.set_screenshot_vars(sr.Shared_Variable_Export())


def _publish_selenium_bridge(driver):
    from Framework.Built_In_Automation.Web.Selenium import BuiltInFunctions as selenium

    selenium.selenium_driver = driver
    if driver is None:
        sr.Remove_From_Shared_Variables("selenium_driver")
    else:
        sr.Set_Shared_Variables("selenium_driver", driver)


def _free_local_port():
    with socket.socket() as sock:
        sock.bind(("127.0.0.1", 0))
        return sock.getsockname()[1]


def _configured_viewport():
    try:
        width = int(ConfigModule.get_config_value("RunDefinition", "window_size_x"))
        height = int(ConfigModule.get_config_value("RunDefinition", "window_size_y"))
        if width > 0 and height > 0:
            return {"width": width, "height": height}
    except (TypeError, ValueError):
        pass
    return {"width": 1920, "height": 1080}


def _selenium_cdp_address(browser_name, debugger, arguments):
    if not any(name in browser_name for name in ("chrome", "edge", "chromium")):
        return None
    if debugger:
        parsed = urlparse(debugger if "://" in debugger else "//" + debugger)
        if not parsed.hostname or not parsed.port:
            raise ValueError(f"Invalid debugger address '{debugger}'")
        return f"{parsed.hostname}:{parsed.port}"

    port = None
    for index, argument in enumerate(arguments):
        if str(argument).startswith("--remote-debugging-port="):
            port = int(str(argument).split("=", 1)[1])
            if port == 0:
                port = _free_local_port()
                arguments[index] = f"--remote-debugging-port={port}"
            break
    if port is None:
        port = _free_local_port()
        arguments.extend(
            [f"--remote-debugging-port={port}", "--remote-debugging-address=127.0.0.1"]
        )
    return f"127.0.0.1:{port}"


def _attach_selenium_bridge(state, browser_name, debugger_address, driver_path=None):
    if not debugger_address:
        state["selenium_bridge"] = None
        _log(
            "Selenium execute-python compatibility is unavailable for Playwright Firefox/WebKit",
            2,
        )
        return
    try:
        from selenium import webdriver

        if "edge" in browser_name:
            from selenium.webdriver.edge.options import Options

            options = Options()
            options.debugger_address = debugger_address
            driver = webdriver.Edge(options=options)
        else:
            from selenium.webdriver.chrome.options import Options
            from selenium.webdriver.chrome.service import Service

            options = Options()
            options.debugger_address = debugger_address
            driver = webdriver.Chrome(
                options=options,
                service=Service(executable_path=driver_path) if driver_path else None,
            )
        state["selenium_bridge"] = driver
        _log("Attached Selenium compatibility bridge to Playwright browser", 1)
    except Exception as error:
        state["selenium_bridge"] = None
        _log(f"Could not attach Selenium compatibility bridge: {error}", 2)


def _close_selenium_bridge(state):
    driver = state.pop("selenium_bridge", None)
    if driver is not None:
        try:
            driver.quit()
        except Exception:
            _log("Unable to close Selenium compatibility bridge", 2)


def _parse(value):
    try:
        return CommonUtil.parse_value_into_object(value)
    except Exception:
        try:
            return ast.literal_eval(value)
        except Exception:
            return value


def _alert_instruction(data_set):
    value = _action(data_set).strip()
    prompt = next((right.strip() for left, middle, right in _rows(data_set)
                   if middle == "input parameter" and left.lower() in ("prompt text", "text", "send text")), None)
    if _key(value).startswith("sendtext") and "=" in value:
        prompt = value.split("=", 1)[1].strip()
    return ("dismiss" if value.lower() in ("reject", "decline", "fail", "no", "cancel", "dismiss") else "accept"), prompt


def _next_alert_instruction():
    try:
        step_data = sr.Get_Shared_Variables("step_data", log=False)
        index = int(CommonUtil.current_action_no)
        if not isinstance(step_data, list) or index >= len(step_data):
            return "dismiss", None
        for left, middle, right in _rows(step_data[index]):
            if left.lower() == "handle alert" and "action" in middle:
                return _alert_instruction(step_data[index])
    except Exception:
        pass
    return "dismiss", None


def _on_dialog(state, dialog):
    instruction, prompt = (
        _alert_instruction(state["alert_rows"])
        if "alert_rows" in state else _next_alert_instruction()
    )
    state["dialog"] = {
        "text": dialog.message,
        "type": dialog.type,
        "result": instruction,
    }
    if instruction == "accept":
        dialog.accept(prompt)
    else:
        dialog.dismiss()


def _wire_page(state, page, activate=True):
    if activate:
        state["page"] = page
        state["frame"] = None
        state["frame_stack"] = []
    wired_pages = state.setdefault("wired_pages", set())
    if id(page) in wired_pages:
        return
    wired_pages.add(id(page))
    page.on("dialog", lambda dialog: _on_dialog(state, dialog))
    page.on("download", lambda download: state["downloads"].append(download))
    if state["capturing_network"]:
        _capture_network_page(state, page)
    if not state.get("page_listener"):
        page.context.on("page", lambda new_page: _wire_page(state, new_page, activate=False))
        state["page_listener"] = True


def _launch(data_set):
    from playwright.sync_api import Error, sync_playwright

    rows = _rows(data_set)
    for left, _middle, right in rows:
        if _key(left) in ("waittimetoappearelement", "waitforelement"):
            sr.Set_Shared_Variables("element_wait", float(right))

    dependency = sr.Get_Shared_Variables("dependency")
    browser_name = str(dependency.get("Browser", "Chrome")).strip().lower()
    if "opera" in browser_name:
        raise RuntimeError("Opera is not supported by Playwright")

    driver_id = _driver_id(data_set, "default")
    if driver_id in _browser_state.playwright_details:
        state = _browser_state.playwright_details[driver_id]
        page = state["page"]
        try:
            # A live request also processes close events queued while Python or
            # Selenium actions were running outside the Playwright thread.
            page.title()
        except Error:
            browser, context = state["browser"], state["context"]
            if not browser.is_connected() or context not in browser.contexts:
                Tear_Down_Selenium([("driver id", "optional parameter", driver_id)])
                state = None
            elif page.is_closed():
                pages = [page for page in context.pages if not page.is_closed()]
                page = pages[-1] if pages else context.new_page()
                _wire_page(state, page)
            else:
                raise
        if state is not None:
            _set_active(driver_id)
            return state

    if _browser_state._playwright is None:
        _browser_state._playwright = sync_playwright().start()

    headless = "headless" in browser_name
    launch = {"headless": headless}
    context_options = {"accept_downloads": True, "viewport": _configured_viewport()}
    wait_until = "load"
    debugger = None
    chrome_version = None
    chrome_driver_path = None
    firefox_prefs = {}
    arguments = []

    for left, middle, right in rows:
        key = _key(left)
        value = _parse(right)
        if key in ("waittimetopageload", "pageloadtimeout"):
            context_options["timeout"] = float(right) * 1000
        elif key == "resolution":
            width, height = (
                int(part.strip())
                for part in right.lower().replace("x", ",").split(",", 1)
            )
            context_options["viewport"] = {"width": width, "height": height}
        elif key in ("addargument", "argument"):
            supplied = value if isinstance(value, list) else [right]
            for argument in supplied:
                if "load-extension" in str(argument) and any(
                    name in browser_name for name in ("chrome", "edge")
                ):
                    _log(
                        "Playwright ignores extension sideloading on branded Chrome/Edge",
                        2,
                    )
                else:
                    arguments.append(argument)
        elif key == "pageloadstrategy":
            wait_until = {
                "normal": "load",
                "eager": "domcontentloaded",
                "none": "commit",
            }.get(right.strip().lower(), "load")
        elif key == "debuggeraddress":
            debugger = right
        elif key in ("chromeversion", "chrome:version"):
            chrome_version = right.strip()
        elif key in ("proxy", "proxyserver"):
            context_options["proxy"] = (
                value if isinstance(value, dict) else {"server": right}
            )
        elif key == "locale":
            context_options["locale"] = right
        elif key in ("useragent", "user-agent"):
            context_options["user_agent"] = right
        elif key == "permissions":
            context_options["permissions"] = (
                value
                if isinstance(value, list)
                else [part.strip() for part in right.split(",")]
            )
        elif key in ("ignorehttpserrors", "accepthttpscerts"):
            context_options["ignore_https_errors"] = (
                str(right).lower() in CommonUtil.affirmative_words
            )
        elif key == "setpreference" and "firefox" in middle:
            if isinstance(value, dict):
                firefox_prefs.update(value)
        elif key in (
            "addextension",
            "addencodedextension",
            "setcapability",
            "addexperimentaloption",
        ):
            _log(f"Playwright ignores unsupported Selenium option '{left}'", 2)
        elif middle == "shared capability":
            _log(f"Playwright ignores unsupported Selenium capability '{left}'", 2)

    selenium_cdp_address = _selenium_cdp_address(browser_name, debugger, arguments)
    launch["args"] = arguments
    if firefox_prefs:
        launch["firefox_user_prefs"] = firefox_prefs
    state = {
        "browser": None,
        "context": None,
        "page": None,
        "frame": None,
        "backend": "playwright",
        "downloads": [],
        "dialog": None,
        "network": [],
        "capturing_network": False,
        "wait_until": wait_until,
        "selenium_bridge": None,
    }

    if debugger:
        if not any(name in browser_name for name in ("chrome", "edge", "chromium")):
            raise RuntimeError(
                "CDP debugger attachment is only supported for Chromium browsers"
            )
        endpoint = debugger if "://" in debugger else "http://" + debugger
        browser = _browser_state._playwright.chromium.connect_over_cdp(endpoint)
        context = (
            browser.contexts[0]
            if browser.contexts
            else browser.new_context(**context_options)
        )
    else:
        if "firefox" in browser_name:
            browser_type = _browser_state._playwright.firefox
        elif "safari" in browser_name or "webkit" in browser_name:
            browser_type = _browser_state._playwright.webkit
        else:
            browser_type = _browser_state._playwright.chromium
            if "edge" in browser_name:
                launch["channel"] = "msedge"
            elif "chrome" in browser_name:
                from Framework.Built_In_Automation.Web.Selenium.utils import (
                    ChromeForTesting,
                )

                channel = None
                if chrome_version and chrome_version.lower() in (
                    "beta",
                    "dev",
                    "canary",
                ):
                    channel, chrome_version = chrome_version.capitalize(), None
                chrome_bin, chrome_driver = ChromeForTesting().setup_chrome_for_testing(
                    chrome_version, channel
                )
                if chrome_bin:
                    launch["executable_path"] = str(chrome_bin)
                    chrome_driver_path = str(chrome_driver)
                    _log(
                        f"Using Chrome for Testing {chrome_version or channel or 'latest'}",
                        1,
                    )
                else:
                    launch["channel"] = "chrome"
            else:
                launch["channel"] = "chrome"
        browser = browser_type.launch(**launch)
        context = browser.new_context(
            **{key: value for key, value in context_options.items() if key != "timeout"}
        )
        if "timeout" in context_options:
            context.set_default_navigation_timeout(context_options["timeout"])

    state["browser"], state["context"] = browser, context
    page = context.pages[-1] if context.pages else context.new_page()
    _wire_page(state, page)
    _attach_selenium_bridge(
        state, browser_name, selenium_cdp_address, chrome_driver_path
    )
    _browser_state.playwright_details[driver_id] = state
    _set_active(driver_id)
    return state


def Go_To_Link(data_set):
    try:
        state = _launch(data_set)
        _wire_page(state, state["page"])
        _set_active(_browser_state.current_driver_id)
        url = next(
            (
                right
                for left, _middle, right in _rows(data_set)
                if _key(left) in ("gotolink", "gotolinkv2")
            ),
            None,
        )
        if url:
            page = state["page"]
            same_document_hash = (
                urlparse(url).fragment
                and urldefrag(page.url).url == urldefrag(url).url
            )
            page.goto(url, wait_until=state["wait_until"])
            if same_document_hash:
                page.reload(wait_until=state["wait_until"])
        return "passed"
    except Exception:
        return CommonUtil.Exception_Handler(
            sys.exc_info(), None, "Failed to open Playwright browser"
        )


Go_To_Link_V2 = Go_To_Link


def Open_Electron_App(data_set):
    from Framework.Built_In_Automation.Web.Selenium import BuiltInFunctions as selenium

    return selenium.Open_Electron_App(data_set)


def Tear_Down_Selenium(data_set=()):
    try:
        requested = (
            _driver_id(data_set)
            if any(_key(left) in ("driverid", "drivertag") for left, _, _ in _rows(data_set))
            else None
        )
        ids = [requested] if requested else list(_browser_state.playwright_details)
        for driver_id in ids:
            state = _browser_state.playwright_details.pop(driver_id, None)
            if not state:
                continue
            try:
                _stop_network_capture(state)
                _close_selenium_bridge(state)
                state["context"].close()
            finally:
                state["browser"].close()
        owners = sr.Get_Shared_Variables("zeuz_browser_backends", log=False)
        owners = owners if isinstance(owners, dict) else {}
        for driver_id in ids:
            if owners.get(driver_id) == "playwright":
                owners.pop(driver_id, None)
        sr.Set_Shared_Variables("zeuz_browser_backends", owners)
        if _browser_state.playwright_details:
            _set_active(next(iter(_browser_state.playwright_details)))
        else:
            if _browser_state._playwright is not None:
                _browser_state._playwright.stop()
            _browser_state._playwright = None
            _browser_state.current_driver_id = _browser_state.playwright_page = None
            sr.Remove_From_Shared_Variables("playwright_page")
            _publish_selenium_bridge(None)
        return "passed"
    except Exception:
        _log("Unable to tear down Playwright browsers", 2)
        return "passed"


def Switch_Browser(data_set):
    driver_id = _driver_id(data_set, "default")
    if driver_id not in _browser_state.playwright_details:
        return _fail(f"Driver_id='{driver_id}' not found")
    _set_active(driver_id)
    return "passed"


def Get_Current_URL(data_set):
    name = _action(data_set)
    return (
        sr.Set_Shared_Variables(name, get_page().url)
        if name
        else _fail("Missing variable name")
    )


def Navigate(data_set):
    value = _action(data_set).strip().lower()
    page = get_page()
    if value == "back":
        page.go_back()
    elif value == "forward":
        page.go_forward()
    elif value == "refresh":
        page.reload()
    else:
        return _fail("Only back, forward, and refresh are supported")
    return "passed"


def Click_Element(data_set):
    element = _element(data_set)
    if element in failed_tag_list:
        return "zeuz_failed"
    _diagnostic("click.target", element)
    options = {}
    for left, middle, right in _rows(data_set):
        if (
            middle in ("optional parameter", "optional option")
            and _key(left) == "offset"
        ):
            x, y = (float(part) for part in right.split(",", 1))
            box = element.bounding_box()
            if not box:
                return "zeuz_failed"
            options["position"] = {
                "x": box["width"] / 2 * (1 + x / 100),
                "y": box["height"] / 2 * (1 + y / 100),
            }
        elif _key(left) == "usejs" and right.strip().lower() in CommonUtil.affirmative_words:
            # Match Selenium's explicit JS click, including mouse-dependent widgets.
            element.evaluate("""element => {
                for (const type of ['mousedown', 'mouseup', 'click'])
                    element.dispatchEvent(new MouseEvent(type, {
                        bubbles: true, cancelable: true, view: window, buttons: 1
                    }));
            }""")
            return "passed"
    from playwright.sync_api import TimeoutError as PlaywrightTimeoutError

    for attempt in range(2):
        try:
            element.click(**options)
            break
        except PlaywrightTimeoutError as error:
            outside = "outside of the viewport" in str(error)
            intercepted = "intercepts pointer events" in str(error)
            if (options or not (outside or intercepted)
                    or not element.is_visible() or not element.is_enabled()):
                raise
            if outside and attempt == 0:
                element.evaluate("el => el.scrollIntoView({block: 'center', inline: 'center'})")
                # An iframe-local scroll may leave the outer page unchanged.
                # Playwright bounding boxes use main-frame viewport coordinates.
                box = element.bounding_box()
                if box:
                    get_page().evaluate("""box => window.scrollBy({
                        left: box.x + box.width / 2 - window.innerWidth / 2,
                        top: box.y + box.height / 2 - window.innerHeight / 2,
                        behavior: 'instant'
                    })""", box)
                _log("Requested parent-page scroll for offscreen target; retrying the click", 2)
                _diagnostic("click.repositioned", element)
                continue
            # Match Selenium's intercepted-click fallback, not unrelated timeouts.
            if not intercepted:
                raise
            element.evaluate("element => element.click()")
            _log("Element click was intercepted; clicked the element using JavaScript", 2)
            break
    return "passed"


def Click_and_Download(data_set):
    timeout = (
        next(
            (
                float(right)
                for left, _, right in _rows(data_set)
                if _key(left) == "waitfordownload"
            ),
            20,
        )
        * 1000
    )
    path = next(
        (
            CommonUtil.path_parser(right)
            for left, middle, right in _rows(data_set)
            if _key(left) in ("folderpath", "directory", "filepath", "file", "folder")
            and middle.startswith("optional")
        ),
        None,
    )
    with get_page().expect_download(timeout=timeout) as info:
        result = Click_Element(data_set)
    if result in failed_tag_list:
        return result
    download = info.value
    if path:
        destination = Path(path)
        if destination.is_dir() or not destination.suffix:
            destination.mkdir(parents=True, exist_ok=True)
            destination /= download.suggested_filename
        else:
            destination.parent.mkdir(parents=True, exist_ok=True)
        download.save_as(str(destination))
    return "passed"


def Right_Click_Element(data_set):
    element = _element(data_set)
    if element in failed_tag_list:
        return "zeuz_failed"
    element.click(button="right")
    return "passed"


def Double_Click_Element(data_set):
    element = _element(data_set)
    if element in failed_tag_list:
        return "zeuz_failed"
    _diagnostic("double_click.target", element)
    element.dblclick()
    return "passed"


def Hover_Over_Element(data_set):
    element = _element(data_set)
    if element in failed_tag_list:
        return "zeuz_failed"
    element.hover()
    return "passed"


_KEYS = {
    "CTRL": "Control",
    "CONTROL": "Control",
    "CMD": "Meta",
    "COMMAND": "Meta",
    "ALT": "Alt",
    "SHIFT": "Shift",
    "ENTER": "Enter",
    "RETURN": "Enter",
    "ESC": "Escape",
    "ESCAPE": "Escape",
    "TAB": "Tab",
    "SPACE": "Space",
    "BACKSPACE": "Backspace",
    "DELETE": "Delete",
    "HOME": "Home",
    "END": "End",
    "PAGEUP": "PageUp",
    "PAGEDOWN": "PageDown",
    "UP": "ArrowUp",
    "DOWN": "ArrowDown",
    "LEFT": "ArrowLeft",
    "RIGHT": "ArrowRight",
}


def Keystroke_For_Element(data_set):
    field, value = next(
        (
            (left.lower(), right)
            for left, middle, right in _rows(data_set)
            if "action" in middle
        ),
        ("", ""),
    )
    element = (
        _element(data_set)
        if any("element parameter" in middle for _, middle, _ in _rows(data_set))
        else None
    )
    keyboard = get_page().keyboard
    if element in failed_tag_list:
        return "zeuz_failed"
    if "chars" in field:
        _type_text(element, value)
    else:
        key, _, count = value.partition(",")
        combo = "+".join(
            _KEYS.get(_key(part).upper(), part.strip()) for part in key.split("+")
        )
        paste_image = any(_key(left) == "pasteimage" and right.strip().lower() == "true"
                          for left, _, right in _rows(data_set))
        if paste_image and combo.lower() in ("control+v", "meta+v"):
            combo = "Meta+v" if get_page().evaluate("/Mac/.test(navigator.platform)") else "Control+v"
            try:
                if element:
                    element.focus()
                keyboard.press(combo)
            except Exception:
                import pyperclip
                text = pyperclip.paste()
                if element:
                    element.evaluate("(el, text) => { el.focus(); el.value = text; }", text)
                else:
                    get_page().evaluate("text => { document.activeElement.value += text; }", text)
            return "passed"
        for _ in range(int(count or 1)):
            element.press(combo) if element else keyboard.press(combo)
    return "passed"


def _type_text(element, value, delay=0):
    if delay or not value:
        (element if element is not None else get_page().keyboard).type(value, delay=delay)
        return

    if element is None:
        keyboard = get_page().keyboard
    else:
        element.type("")  # Preserve native typing's focus and caret placement.
        page = element.page if LocateElement._is_playwright_locator(element) else element.owner_frame().page
        keyboard = page.keyboard

    async def send_keys():
        commands = []
        for char in value:
            # Match Playwright's US-layout keys and insertText for other characters.
            if " " <= char <= "~" or char in "\n\r":
                commands.extend((keyboard._impl_obj.down(char), keyboard._impl_obj.up(char)))
            else:
                commands.append(keyboard._impl_obj.insert_text(char))
        commands = [asyncio.create_task(command) for command in commands]
        # Reuse the action's call site; collecting a stack per key delays dispatch.
        for command in commands:
            command.__pw_stack_trace__ = asyncio.current_task().__pw_stack_trace__
        results = await asyncio.gather(*commands, return_exceptions=True)
        for result in results:
            if isinstance(result, BaseException):
                raise result

    # Queue native events like ChromeDriver send_keys, without per-event round trips.
    # ponytail: sync/async bridge uses Playwright internals; replace when a public
    # batched-key API exists. Browser event-parity tests cover dependency upgrades.
    keyboard._sync(keyboard._impl_obj._channel._connection.wrap_api_call(send_keys))


def Enter_Text_In_Text_Box(data_set):
    element = _element(data_set)
    if element in failed_tag_list:
        return "zeuz_failed"
    _diagnostic("text.target", element)
    value = _action(data_set)
    if (element.get_attribute("type") or "").lower() == "file":
        element.set_input_files(CommonUtil.path_parser(value))
        return "passed"

    if any(_key(left) == "usejs" and right.strip().lower() in ("true", "yes", "1")
           for left, _, right in _rows(data_set)):
        element.evaluate("""(el, value) => {
            el.click();
            el.value = value;
            el.dispatchEvent(new Event('input', {bubbles: true}));
            el.dispatchEvent(new Event('change', {bubbles: true}));
            el.click();
        }""", value)
        return "passed"
    append = any(
        _key(left) == "append" and right.strip().lower() in CommonUtil.affirmative_words
        for left, _, right in _rows(data_set)
    )
    clear = not any(
        _key(left) == "clear" and right.strip().lower() in ("no", "false")
        for left, _, right in _rows(data_set)
    )
    delay = next(
        (float(right) * 1000 for left, _, right in _rows(data_set) if _key(left) == "delay"),
        0,
    )
    if not 0 <= delay < float("inf"):
        return _fail("Typing delay must be a finite, non-negative number of seconds")
    # Clicking/clearing can change attributes used by the locator (e.g. ng-pristine).
    from playwright.sync_api import Error as PlaywrightError, TimeoutError as PlaywrightTimeoutError

    handle = element.element_handle()
    try:
        try:
            handle.click(timeout=5000)
        except PlaywrightTimeoutError:
            _log("Entering text without clicking the element", 2)
        if clear and not append:
            try:
                handle.press("ControlOrMeta+A")
                handle.press("Delete")
                if handle.evaluate("el => el.isContentEditable ? el.textContent : el.value"):
                    handle.fill("")
            except PlaywrightError:
                pass  # Selenium treats clearing as best effort before send_keys.
        else:
            handle.press("ControlOrMeta+End")
        _type_text(handle, value, delay=delay)
        try:
            handle.click(timeout=5000)
        except PlaywrightError:
            pass  # Match Selenium's best-effort click after typing.
    finally:
        handle.dispose()
    return "passed"


def Validate_Text(data_set):
    element = _element(data_set)
    if element in failed_tag_list:
        return "zeuz_failed"
    name, expected = next(
        (
            (left.lower(), right)
            for left, middle, right in _rows(data_set)
            if "action" in middle
        ),
        ("", ""),
    )
    actual = element.inner_text().strip()
    ignore_case = any(
        _key(left) == "ignorecase" and right.strip().lower() in CommonUtil.affirmative_words
        for left, _, right in _rows(data_set)
    )
    if ignore_case:
        actual, expected = actual.lower(), expected.lower()
    lines = [line for line in actual.split("\n") if line != ""]
    for left, middle, right in _rows(data_set):
        if middle == "text classifier offset":
            from Framework.AI.NLP import binary_classification
            return binary_classification(" ".join(lines), [left], float(right))["status"]
    valid = any(expected in line for line in lines) if "partial" in name else expected in lines
    return (
        "passed"
        if valid
        else _fail(f"Expected text {expected!r}; actual text {actual!r}")
    )


def Select_Deselect(data_set):
    element = _element(data_set)
    if element in failed_tag_list:
        return "zeuz_failed"
    name, value = next(
        (
            (left.lower(), right)
            for left, middle, right in _rows(data_set)
            if "action" in middle
        ),
        ("", ""),
    )
    if name == "deselect all":
        element.select_option([])
        return "passed"
    if name.startswith("deselect"):
        if "visible text" in name:
            remove = element.locator("option", has_text=value).first.get_attribute(
                "value"
            )
        elif "index" in name:
            remove = element.locator("option").nth(int(value)).get_attribute("value")
        else:
            remove = value
        selected = element.locator("option:checked").evaluate_all(
            "(options, remove) => options.filter(o => o.value !== remove).map(o => o.value)",
            remove,
        )
        element.select_option(selected)
    elif "visible text" in name:
        element.select_option(label=value)
    elif "value" in name:
        element.select_option(value=value)
    elif "index" in name:
        element.select_option(index=int(value))
    else:
        return _fail("Unknown select action")
    return "passed"


def Handle_Browser_Alert(data_set):
    state = _state()
    if not state.get("dialog"):
        from playwright.sync_api import TimeoutError as PlaywrightTimeoutError

        timeout = next((float(right) for left, middle, right in _rows(data_set)
                        if middle == "optional parameter" and left.lower() in ("wait", "timeout")), 5)
        state["alert_rows"] = data_set
        try:
            get_page().wait_for_event("dialog", timeout=timeout * 1000)
        except PlaywrightTimeoutError:
            return _fail("No alert appeared within timeout")
        finally:
            state.pop("alert_rows", None)
    cached = state.pop("dialog")
    value = _action(data_set).strip()
    name = next((left for left, middle, _ in _rows(data_set) if middle == "save parameter"), None)
    if _key(value).startswith("gettext") and "=" in value:
        name = value.split("=", 1)[1].strip()
    if name:
        sr.Set_Shared_Variables(name, cached["text"])
    return "passed"


def open_new_tab(data_set):
    state = _state()
    page = state["context"].new_page()
    _wire_page(state, page)
    url = _action(data_set)
    if url and url.lower() not in ("open new tab", "new tab"):
        page.goto(url)
    _set_active(_browser_state.current_driver_id)
    return "passed"


def _find_page(data_set):
    from playwright.sync_api import TimeoutError as PlaywrightTimeoutError

    context = _state()["context"]
    rows = _rows(data_set)
    partial_title = any(left.startswith("*") and _key(left).lstrip("*") in ("tabtitle", "windowtitle", "title")
                        for left, _, _ in rows)
    timeout = next((float(right) for left, middle, right in rows
                    if middle == "optional parameter" and _key(left) in ("wait", "timeout")), 10)
    if not 0 <= timeout < float("inf"):
        raise ValueError("Tab wait must be a finite, non-negative number")
    deadline = time.monotonic() + timeout
    interval = 1
    while True:
        # Pump queued page/close events before reading Playwright's cached pages.
        try:
            context.wait_for_event("page", timeout=interval)
        except PlaywrightTimeoutError:
            pass
        pages = [page for page in context.pages if not page.is_closed()]
        # Selenium lets later title/index rows override earlier ones.
        for left, _middle, right in reversed(rows):
            right = right.strip()
            partial = left.startswith("*")
            key = _key(left).lstrip("*")
            if key in ("tabindex", "windowindex", "index"):
                index = int(right)
                page = pages[index] if -len(pages) <= index < len(pages) else None
                break
            if key in ("tabtitle", "windowtitle", "title"):
                page = next((page for page in pages if (
                    right.lower() in page.title().lower() if partial_title
                    else right.lower() == page.title().lower()
                )), None)
                break
            if key in ("url", "taburl", "windowurl"):
                page = next((page for page in pages
                             if (right in page.url if partial else right == page.url)), None)
                break
        else:
            page = pages[-1] if pages else None
        if page is not None:
            return page
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            return None
        interval = min(100, remaining * 1000)


def switch_window_or_tab(data_set):
    page = _find_page(data_set)
    if page is None:
        return _fail("Requested tab/window was not found")
    _wire_page(_state(), page)
    page.bring_to_front()
    _set_active(_browser_state.current_driver_id)
    return "passed"


def close_tab(data_set):
    state = _state()
    tabs = next((_parse(right) for left, _, right in _rows(data_set) if _key(left) == "tabs"), None)
    if tabs is not None:
        if not isinstance(tabs, list):
            return _fail("tabs must be a list of titles or indices")
        selected = []
        for tab in tabs:
            field = "tab index" if isinstance(tab, int) else "tab title"
            page = _find_page([(field, "input parameter", str(tab))] + [
                row for row in data_set if _key(row[0]) in ("wait", "timeout")
            ])
            if page is None:
                return _fail("Requested tab/window was not found")
            if page not in selected:
                selected.append(page)
    else:
        selected = [_find_page(data_set)]
    if None in selected:
        return _fail("Requested tab/window was not found")
    for page in selected:
        page.close()
    if state["context"].pages:
        _wire_page(state, state["context"].pages[-1])
        _set_active(_browser_state.current_driver_id)
    return "passed"


def switch_iframe(data_set):
    state = _state()
    stack = state.setdefault("frame_stack", [])
    command = _action(data_set).strip().lower()
    frame_rows = [
        row
        for row in _rows(data_set)
        if row[1] in ("iframe parameter", "frame parameter")
    ]
    reset = any(
        right.lower() in ("default", "default content", "main", "top")
        for _, _, right in frame_rows
    )
    if reset:
        state["frame"] = None
        stack.clear()
        frame_rows = [
            row
            for row in frame_rows
            if row[2].lower() not in ("default", "default content", "main", "top")
        ]
    indexed = next(
        (
            right
            for left, _, right in frame_rows
            if left.lower() == "index" and right.lstrip("-").isdigit()
        ),
        None,
    )
    if command in ("default", "default content", "main", "top"):
        state["frame"] = None
        stack.clear()
    elif reset and not frame_rows:
        pass
    elif command in ("parent", "parent frame"):
        state["frame"] = stack.pop() if stack else None
    elif command.lstrip("-").isdigit() or indexed is not None:
        index = int(indexed if indexed is not None else command)
        frames = [
            frame for frame in state["page"].frames if frame != state["page"].main_frame
        ]
        try:
            stack.append(state.get("frame"))
            state["frame"] = frames[index]
        except IndexError:
            return _fail(f"Iframe index {index} was not found")
    else:
        frame_name = next(
            (
                right
                for left, _, right in _rows(data_set)
                if _key(left) in ("framename", "iframename", "name")
            ),
            None,
        )
        frame_url = next(
            (
                right
                for left, _, right in _rows(data_set)
                if _key(left) in ("frameurl", "iframeurl", "url")
            ),
            None,
        )
        if frame_name or frame_url:
            stack.append(state.get("frame"))
            state["frame"] = next(
                (
                    frame
                    for frame in state["page"].frames
                    if (not frame_name or frame.name == frame_name)
                    and (not frame_url or frame_url in frame.url)
                ),
                None,
            )
        else:
            locator_rows = [
                (
                    left,
                    "element parameter"
                    if middle in ("iframe parameter", "frame parameter")
                    else middle,
                    right,
                )
                for left, middle, right in _rows(data_set)
                if not (
                    middle in ("iframe parameter", "frame parameter")
                    and right.lower()
                    in ("default", "default content", "main", "top")
                )
            ]
            element = _element(locator_rows, root=state["page"])
            if element in failed_tag_list:
                return "zeuz_failed"
            stack.append(state.get("frame"))
            content_frame = element.content_frame
            state["frame"] = (
                content_frame() if callable(content_frame) else content_frame
            )
        if state["frame"] is None:
            return _fail("Located element is not an iframe")
    sr.Set_Shared_Variables("common_driver", state.get("frame") or state["page"])
    return "passed"


def upload_file(data_set):
    locator_rows = any(
        middle.endswith("parameter")
        and middle
        not in ("optional parameter", "input parameter", "save parameter", "get parameter")
        for _left, middle, _right in _rows(data_set)
    )
    element = (
        _element(data_set)
        if locator_rows
        else get_driver().locator("input[type=file]").first
    )
    if element in failed_tag_list:
        return "zeuz_failed"
    files = _parse(_action(data_set))
    files = (
        [CommonUtil.path_parser(str(path)) for path in files]
        if isinstance(files, list)
        else CommonUtil.path_parser(str(files))
    )
    paths = files if isinstance(files, list) else [files]
    if any(not Path(path).is_file() for path in paths):
        return _fail("One or more upload files were not found")
    element.set_input_files(files)
    return "passed"


def drag_and_drop(data_set):
    source, target, shared_options = [], [], []
    destination_offset = delay = None
    optional_types = ("option", "optional parameter", "optional option")
    for raw_left, raw_middle, raw_right in data_set:
        left = str(raw_left).strip()
        middle = str(raw_middle).strip().lower()
        right = str(raw_right)
        if middle.startswith(("source ", "src ")):
            source.append((left, middle.split(" ", 1)[1], right))
        elif middle.startswith(("destination ", "dst ")):
            target.append((left, middle.split(" ", 1)[1], right))
        elif _key(left) in ("wait", "allowdisable", "allowhidden") and middle in optional_types:
            shared_options.append((left, "optional parameter", right))
        elif _key(left) == "destinationoffset" and middle in optional_types:
            destination_offset = right
        elif _key(left) == "delay":
            delay = float(right.strip())

    non_locators = (*optional_types, "save parameter")
    if not source or all(middle in non_locators for _, middle, _ in source):
        return _fail("Please provide a source element locator")
    if not target or all(middle in non_locators for _, middle, _ in target):
        return _fail("Please provide a destination element locator")
    source.extend(shared_options)
    target.extend(shared_options)

    src = _element(source)
    if src in failed_tag_list:
        return "zeuz_failed"
    dst = _element(target)
    if dst in failed_tag_list:
        return "zeuz_failed"

    target_position = None
    if destination_offset:
        x, y = (float(part) for part in destination_offset.split(",", 1))
        box = dst.bounding_box()
        if not box:
            return _fail("Destination element has no bounding box")
        target_position = {
            "x": box["width"] / 2 * (1 + x / 100),
            "y": box["height"] / 2 * (1 + y / 100),
        }

    if delay is None:
        src.drag_to(dst, **({"target_position": target_position} if target_position else {}))
    else:
        src.hover()
        page = get_page()
        page.mouse.down()
        try:
            hover_options = {"position": target_position} if target_position else {}
            dst.hover(**hover_options)
            dst.hover(**hover_options)
            page.wait_for_timeout(delay * 1000)
        finally:
            page.mouse.up()
    return "passed"


def get_element_info(data_set):
    element = _element(data_set)
    if element in failed_tag_list:
        return "zeuz_failed"
    name = _action(data_set)
    box = element.bounding_box() or {}
    value = {
        "size": {"width": box.get("width", 0), "height": box.get("height", 0)},
        "location": {"x": box.get("x", 0), "y": box.get("y", 0)},
    }
    return sr.Set_Shared_Variables(name, value)


def Save_Attribute(data_set):
    variable = attribute = None
    clean = []
    for left, middle, right in _rows(data_set):
        if middle == "save parameter":
            attribute, variable = left.lower(), right
        else:
            clean.append((left, middle, right))
    element = _element(clean)
    if element in failed_tag_list or not variable:
        return "zeuz_failed"
    _diagnostic("attribute.target", element)
    if attribute == "text":
        value = element.inner_text().strip()
    elif attribute == "tag":
        value = element.evaluate("element => element.tagName.toLowerCase()")
    elif attribute == "checked":
        value = element.is_checked()
    elif attribute == "value":
        value = element.evaluate("el => el.value ?? el.getAttribute('value')")
    else:
        value = element.get_attribute(attribute)
    return sr.Set_Shared_Variables(variable, value)


def _target_specs(data_set):
    specs = []
    for _left, middle, right in _rows(data_set):
        if "target parameter" not in middle:
            continue
        spec = {}
        for item in right.replace(",\n", ",").split(","):
            item = item.strip()
            if "(" in item and item.split("(", 1)[0].strip() in ("return_contains", "return_does_not_contain"):
                key, value = item.split("(", 1)
                spec.setdefault(key.strip(), []).append(value.rstrip(")"))
                continue
            if "=" in item:
                key, value = item.split("=", 1)
                spec.setdefault(key.strip(), []).append(value.strip().strip('"'))
        specs.append(spec)
    return specs


def _return_attribute(element, attribute):
    if attribute == "text":
        return element.inner_text().strip()
    if attribute == "tag":
        return element.evaluate("el => el.tagName.toLowerCase()")
    if attribute == "checked":
        return str(element.evaluate("el => Boolean(el.checked || el.selected)"))
    return element.get_attribute(attribute)


def _target_rows(spec):
    return [
        ("allow hidden", "optional parameter", values[0])
        if _key(key) in ("allowhidden", "allowdisable")
        else (key, "element parameter", values[0])
        for key, values in spec.items()
        if not key.startswith("return")
    ]


def _return_matches(value, contains, excludes):
    return (not contains or any(not isinstance(value, str) or not part or part in value
                                for part in contains)) and not any(
        isinstance(value, str) and part and part in value for part in excludes)


def save_attribute_values_in_list(data_set):
    parent = _element(data_set)
    if parent in failed_tag_list:
        return "zeuz_failed"
    result = []
    for spec in _target_specs(data_set):
        elements = _element(_target_rows(spec), all_elements=True, root=parent)
        if elements in failed_tag_list:
            return "zeuz_failed"
        values = []
        for element in elements:
            attribute = spec.get("return", ["text"])[0]
            value = _return_attribute(element, attribute)
            values.append(value if _return_matches(value, spec.get("return_contains", []),
                                                   spec.get("return_does_not_contain", [])) else None)
        result.append(values)
    name = next(
        (
            right
            for left, _, right in _rows(data_set)
            if left.lower() == "save attribute values in list"
        ),
        "",
    )
    paired = not any(_key(left) == "paired" and right.strip().lower() == "no"
                     for left, _, right in _rows(data_set))
    if len(result) == 1:
        result = result[0]
    else:
        result = [[group[index] for group in result if index < len(group)]
                  for index in range(max(map(len, result), default=0))]
        if not paired:
            result = list(map(list, zip(*result)))
    return sr.Set_Shared_Variables(name, result)


def Extract_Table_Data(data_set):
    table = _element(data_set)
    if table in failed_tag_list:
        return "zeuz_failed"
    values = table.locator("tr").evaluate_all(
        "rows => rows.map(r => [...r.querySelectorAll('td')].map(c => c.textContent.trim()))"
    )
    row = next(
        (right for left, _, right in _rows(data_set) if "row" in left.lower()), ""
    )
    column = next(
        (right for left, _, right in _rows(data_set) if "column" in left.lower()), ""
    )
    if row:
        values = [values[int(row)]]
    if column:
        values = [[item[int(column)]] for item in values]
    name = next(
        (
            right
            for left, _, right in _rows(data_set)
            if left.lower() == "extract table data"
        ),
        "",
    )
    return sr.Set_Shared_Variables(name, values)


def save_web_elements_in_list(data_set):
    name = _action(data_set)
    has_parent = any(
        middle
        in (
            "element parameter",
            "parent parameter",
            "child parameter",
            "sibling parameter",
        )
        for _, middle, _ in _rows(data_set)
    )
    root = _element(data_set) if has_parent else get_driver()
    if root in failed_tag_list:
        return "zeuz_failed"
    groups = []
    for spec in _target_specs(data_set):
        elements = _element(_target_rows(spec), all_elements=True, root=root)
        if elements in failed_tag_list:
            return "zeuz_failed"
        for filter_name in ("return_contains", "return_does_not_contain"):
            filters = spec.get(filter_name, [])
            if not filters:
                continue
            retained = []
            for element in elements:
                matches = []
                for rule in filters:
                    attribute, expected = rule.split("=", 1)
                    value = _return_attribute(element, attribute.strip().strip('"'))
                    matches.append(value is None or expected.strip().strip('"') in value)
                if any(matches) == (filter_name == "return_contains"):
                    retained.append(element)
            elements = retained
        groups.append(elements)
    value = groups[0] if len(groups) == 1 else groups
    return sr.Set_Shared_Variables(name, value)


def take_screenshot_selenium(data_set):
    filename_format = next(
        (
            right
            for left, _, right in _rows(data_set)
            if "take screenshot web" in left.lower() and "default" not in right.lower()
        ),
        "%Y_%m_%d_%H-%M-%S",
    )
    full_page = any(
        "fullscreen" in left.lower() and right.strip().lower() in CommonUtil.affirmative_words
        for left, _, right in _rows(data_set)
    )
    folder = ConfigModule.get_config_value(
        "sectionOne", "screen_capture_folder", temp_config
    )
    Path(folder).mkdir(parents=True, exist_ok=True)
    filename = time.strftime(filename_format) + ".png"
    get_page().screenshot(path=str(Path(folder) / filename), full_page=full_page)
    sr.Set_Shared_Variables("zeuz_screenshot", filename)
    return "passed"


def execute_javascript(data_set):
    script = next(
        (right for left, _, right in _rows(data_set) if "javascript" in left.lower()),
        None,
    )
    variable = next(
        (right for left, _, right in _rows(data_set) if left.lower() == "variable"),
        None,
    )
    if script is None:
        return _fail("Missing JavaScript")
    if any("element parameter" in middle for _, middle, _ in _rows(data_set)):
        element = _element(data_set)
        if element in failed_tag_list:
            return "zeuz_failed"
        body = script.replace("$elem", "element").replace("arguments[0]", "element")
        result = element.evaluate(f"element => {{ {body} }}")
    else:
        stripped = script.strip()
        if stripped.startswith("return "):
            script = f"() => ({stripped[7:].rstrip(';')})"
        elif ";" in stripped and "=>" not in stripped:
            script = f"() => {{ {stripped} }}"
        result = get_page().evaluate(script)
    return sr.Set_Shared_Variables(variable, result) if variable else "passed"


def Scroll(data_set):
    direction = _action(data_set).strip().lower()
    pixels = next(
        (int(right) for left, _, right in _rows(data_set) if left.lower() == "pixels"),
        750,
    )
    x = pixels if direction == "right" else -pixels if direction == "left" else 0
    y = pixels if direction == "down" else -pixels if direction == "up" else 0
    element = (
        _element(data_set)
        if any("element parameter" in middle for _, middle, _ in _rows(data_set))
        else None
    )
    (element or get_page()).evaluate(
        "(target, delta) => (target === document ? window : target).scrollBy(delta.x, delta.y)",
        {"x": x, "y": y},
    ) if element else get_page().evaluate(
        "delta => window.scrollBy(delta.x, delta.y)", {"x": x, "y": y}
    )
    return "passed"


def scroll_to_element(data_set):
    element = _element(data_set)
    if element in failed_tag_list:
        return "zeuz_failed"
    method, align, additional = "js", True, 0.1
    direction = ""
    for left, _, right in _rows(data_set):
        key, value = _key(left), right.strip().lower()
        if key == "usejs":
            method = "js" if value in ("true", "yes", "1") else "action chain"
        elif key == "method":
            method = value
        elif key == "aligntotop":
            align = value in ("true", "yes", "1")
        elif key.startswith("additionalscroll"):
            additional = float(value)
            direction = key.removeprefix("additionalscroll")
    page = get_page()
    before = page.evaluate("[window.scrollX, window.scrollY]")
    if method == "js":
        element.evaluate("(el, align) => el.scrollIntoView(align)", align)
    elif method == "webdriver":
        element.evaluate("el => el.scrollIntoView(true)")
    else:
        element.hover()
    if additional > 0:
        after = page.evaluate("[window.scrollX, window.scrollY]")
        if not direction:
            dx, dy = after[0] - before[0], after[1] - before[1]
            direction = "down" if dy > 0 else "up" if dy < 0 else "right" if dx > 0 else "left" if dx < 0 else ""
            if (method in ("js", "webdriver") and align and direction in ("down", "right")) or (not align and direction in ("up", "left")):
                direction = ""
        if direction:
            page.evaluate("""({direction, fraction}) => {
                const x = direction === 'right' ? 1 : direction === 'left' ? -1 : 0;
                const y = direction === 'down' ? 1 : direction === 'up' ? -1 : 0;
                window.scrollBy(x * Math.round(window.innerWidth * fraction),
                                y * Math.round(window.innerHeight * fraction));
            }""", {"direction": direction, "fraction": additional})
    return "passed"


def scroll_to_top(data_set):
    get_page().evaluate("window.scrollTo(window.scrollX, 0)")
    return "passed"


def _set_checked(element, checked, data_set):
    if any(_key(left) == "usejs" and right.strip().lower() in ("true", "yes", "ok")
           for left, _, right in _rows(data_set)):
        element.evaluate("""(el, checked) => {
            if (Boolean(el.checked || el.selected) !== checked) el.click();
        }""", checked)
    else:
        element.check() if checked else element.uncheck()


def check_uncheck(data_set):
    element = _element(data_set)
    if element in failed_tag_list:
        return "zeuz_failed"
    command = _action(data_set).strip().lower()
    _set_checked(element, "uncheck" not in command, data_set)
    return "passed"


def check_uncheck_all(data_set):
    parent = _element(data_set)
    if parent in failed_tag_list:
        return "zeuz_failed"
    targets = [
        (left, "element parameter", right)
        for left, middle, right in _rows(data_set)
        if middle == "target parameter"
    ]
    elements = _element(targets, all_elements=True, root=parent)
    uncheck = "uncheck" in _action(data_set).lower()
    for element in elements:
        _set_checked(element, not uncheck, data_set)
    return "passed"


def multiple_check_uncheck(data_set):
    parent = _element(data_set)
    if parent in failed_tag_list:
        return "zeuz_failed"
    for _left, middle, right in _rows(data_set):
        if middle != "target parameter":
            continue
        for attribute, value, command in _parse("[" + right + "]"):
            element = _element([(attribute, "element parameter", value)], root=parent)
            if element not in failed_tag_list:
                _set_checked(element, "uncheck" not in command.lower(), data_set)
    return "passed"


def slider_bar(data_set):
    element = _element(data_set)
    if element in failed_tag_list:
        return "zeuz_failed"
    value = int(_action(data_set))
    box = element.bounding_box()
    if not box:
        return "zeuz_failed"
    get_page().mouse.click(
        box["x"] + box["width"] * value / 100, box["y"] + box["height"] / 2
    )
    return "passed"


def resize_window(data_set):
    page = get_page()
    size = page.viewport_size or {"width": 1920, "height": 1080}
    for left, middle, right in _rows(data_set):
        if "element parameter" in middle and left.lower() in ("width", "height"):
            size[left.lower()] = round(
                size[left.lower()] * float(right.rstrip("%")) / 100
            )
    page.set_viewport_size(size)
    return "passed"


def Change_Attribute_Value(data_set):
    element = _element(data_set)
    if element in failed_tag_list:
        return "zeuz_failed"
    attribute, value = next(
        (
            (left, right)
            for left, middle, right in _rows(data_set)
            if "input parameter" in middle
        ),
        (None, None),
    )
    if not attribute:
        return _fail("Missing attribute input parameter")
    element.evaluate(
        "(element, pair) => { element[pair.name] = pair.value; element.setAttribute(pair.name, pair.value); }",
        {"name": attribute, "value": value},
    )
    return "passed"


def _network_listener(state, emitter, event, callback):
    emitter.on(event, callback)
    state.setdefault("network_listeners", []).append((emitter, event, callback))


def _capture_network_target(state, target):
    sessions = state.setdefault("network_sessions", {})
    if target in sessions:
        return
    try:
        session = state["context"].new_cdp_session(target)
    except Exception as error:
        # In-process iframes are already covered by their parent's session.
        if "does not have a separate CDP session" in str(error):
            return
        raise
    sessions[target] = session
    pending = {}
    _network_listener(state, session, "close", lambda _: sessions.pop(target, None))

    def response_received(event):
        if not state["capturing_network"]:
            return
        response = event["response"]
        item = pending.setdefault(event["requestId"], {"method": ""})
        item.update({
            "url": response["url"],
            "status": response["status"],
            "mimeType": response.get("mimeType", "").split(";", 1)[0],
            "type": event.get("type", "other").lower(),
            "timestamp": time.time(),
            "_session": session,
            "_request_id": event["requestId"],
            "_finished": False,
        })
        state["network"].append(item)

    def request_started(event):
        if not state["capturing_network"]:
            return
        if "redirectResponse" in event:
            response_received({**event, "response": event["redirectResponse"]})
        pending[event["requestId"]] = {"method": event["request"]["method"]}

    def request_finished(event):
        item = pending.pop(event["requestId"], None)
        if item is not None:
            item["_finished"] = "errorText" not in event

    _network_listener(state, session, "Network.requestWillBeSent", request_started)
    _network_listener(state, session, "Network.responseReceived", response_received)
    _network_listener(state, session, "Network.loadingFinished", request_finished)
    _network_listener(state, session, "Network.loadingFailed", request_finished)
    # Let Chrome manage its bounded response cache; do not copy POST bodies or
    # create Playwright Request/Response objects for this long-lived browser.
    session.send("Network.enable", {"maxPostDataSize": 0})


def _capture_network_page(state, page):
    if state["browser"].browser_type.name == "chromium":
        _capture_network_target(state, page)
        for frame in page.frames:
            if frame != page.main_frame:
                _capture_network_target(state, frame)
        _network_listener(
            state, page, "framenavigated",
            lambda frame: _capture_network_target(state, frame)
            if state["capturing_network"] and frame != page.main_frame else None,
        )
    else:
        # ponytail: Firefox/WebKit have no CDP. Keep their existing response API, but
        # subscribe only during capture (their retained history still requires
        # page/context teardown for long-running sessions).
        _network_listener(state, page, "response", lambda response: state["network"].append({
            "url": response.url,
            "status": response.status,
            "method": response.request.method,
            "mimeType": response.headers.get("content-type", "").split(";", 1)[0],
            "type": response.request.resource_type,
            "timestamp": time.time(),
            "_response": response,
        }) if state["capturing_network"] else None)


def _stop_network_capture(state):
    state["capturing_network"] = False
    state["network"] = []
    for emitter, event, callback in state.pop("network_listeners", []):
        emitter.remove_listener(event, callback)
    for session in state.pop("network_sessions", {}).values():
        try:
            session.send("Network.disable")
        except Exception:
            pass  # The page may already have closed or crashed.
        finally:
            try:
                session.detach()
            except Exception:
                pass


def cleanup_network_captures(data_set=()):
    for state in _browser_state.playwright_details.values():
        _stop_network_capture(state)
    return "passed"


cleanup_network_captures._zeuz_thread_affine = True
get_dom._zeuz_thread_affine = True
Tear_Down_Selenium._zeuz_thread_affine = True


def capture_network_log(data_set):
    state = _state()
    command = _action(data_set).strip().lower()
    if command == "start":
        _stop_network_capture(state)
        state["capturing_network"] = True
        try:
            for page in state["context"].pages:
                _capture_network_page(state, page)
        except Exception:
            _stop_network_capture(state)
            raise
        return "passed"
    try:
        state["capturing_network"] = False
        captured_network, state["network"] = state["network"], []
        variable = next(
            (
                right
                for left, _, right in _rows(data_set)
                if _key(left) in ("save", "variable", "variablename", "saveas")
            ),
            None,
        )
        domains = [
            item.strip()
            for left, _, right in _rows(data_set)
            if _key(left) == "filterdomain"
            for item in right.split(",")
        ]
        methods = [
            item.strip().upper()
            for left, _, right in _rows(data_set)
            if _key(left) == "includerequestmethod"
            for item in right.split(",")
        ]
        statuses = set()
        for left, _, right in _rows(data_set):
            if _key(left) != "includestatuscode":
                continue
            for part in right.split(","):
                bounds = [int(item) for item in part.strip().split("-", 1)]
                statuses.update(range(bounds[0], bounds[-1] + 1))
        include_body = any(
            _key(left) == "includeresponsebody"
            and right.strip().lower() in CommonUtil.affirmative_words
            for left, _, right in _rows(data_set)
        )
        static = (
            ".js",
            ".css",
            ".png",
            ".jpg",
            ".jpeg",
            ".gif",
            ".svg",
            ".ico",
            ".woff",
            ".woff2",
            ".ttf",
            ".map",
            ".txt",
            ".webp",
            ".eot",
        )
        logs = []
        for captured in captured_network:
            if captured.get("type") == "request" or captured["url"].lower().endswith(
                static
            ) or captured.get("mimeType", "").startswith((
                "image/", "font/", "text/css", "application/javascript",
                "text/javascript", "application/font-", "application/x-font-",
            )):
                continue
            if domains and not any(domain in captured["url"] for domain in domains):
                continue
            if methods and captured["method"].upper() not in methods:
                continue
            if statuses and captured["status"] not in statuses:
                continue
            item = {key: value for key, value in captured.items() if not key.startswith("_")}
            if include_body:
                try:
                    if "_session" in captured:
                        if not captured["_finished"]:
                            item["body"] = "Unavailable"
                        else:
                            result = captured["_session"].send(
                                "Network.getResponseBody", {"requestId": captured["_request_id"]}
                            )
                            item["body"] = (
                                base64.b64decode(result["body"]).decode(errors="replace")
                                if result.get("base64Encoded") else result["body"]
                            )
                    else:
                        response = captured["_response"]
                        item["body"] = (
                            response.body().decode(errors="replace")
                            if response.request.timing["responseEnd"] != -1
                            else "Unavailable"
                        )
                except Exception:
                    item["body"] = "Unavailable"
            logs.append(item)
        if variable:
            sr.Set_Shared_Variables(variable, logs)
    finally:
        _stop_network_capture(state)
    return "passed"


def if_element_exists(data_set):
    value, variable = (part.strip() for part in _action(data_set).split("=", 1))
    sr.Set_Shared_Variables(
        variable, value if _element(data_set) not in failed_tag_list else "false"
    )
    return "passed"


def copy_image_into_browser(data_set):
    image_variable = next((right.strip() for left, _, right in _rows(data_set)
                           if _key(left) == "imagevariable"), "")
    path = next(
        (
            right
            for left, _, right in _rows(data_set)
            if _key(left) in ("imagefile", "filepath", "file")
        ),
        "",
    )
    if not path and image_variable:
        path = image_variable if Path(image_variable).is_file() else sr.Get_Shared_Variables(image_variable)
    path = CommonUtil.path_parser(path)
    if not path or not Path(path).is_file():
        return _fail("Image file was not found")
    mime = "image/svg+xml" if str(path).lower().endswith(".svg") else "image/png"
    encoded = base64.b64encode(Path(path).read_bytes()).decode()
    try:
        _state()["context"].grant_permissions(
            ["clipboard-read", "clipboard-write"], origin=get_page().url
        )
    except Exception:
        pass
    get_page().evaluate(
        """async ({data, mime}) => {
        const blob = await (await fetch(`data:${mime};base64,${data}`)).blob();
        await navigator.clipboard.write([new ClipboardItem({[mime]: blob})]);
    }""",
        {"data": encoded, "mime": mime},
    )
    return "passed"


Tear_Down_Playwright = Tear_Down_Selenium


def _action_guard(function):
    @functools.wraps(function)
    def guarded(data_set=()):
        try:
            result = function(data_set)
        except Exception:
            if function.__name__ in ("Click_Element", "Double_Click_Element", "Enter_Text_In_Text_Box", "Save_Attribute"):
                _diagnostic(function.__name__ + ".error")
            return CommonUtil.Exception_Handler(sys.exc_info())
        if function.__name__ in ("Click_Element", "Double_Click_Element", "Enter_Text_In_Text_Box", "Save_Attribute",
                                 "take_screenshot_selenium"):
            _diagnostic(function.__name__ + ".after")
        return result

    return guarded


# Keep direct calls consistent with Sequential Actions: action errors are results,
# not exceptions escaping into callers.
for _name in {_declaration["function"] for _declaration in declarations}:
    globals()[_name] = _action_guard(globals()[_name])