handler_utils.input_sequentially (#2722)

skyvern/constants.py

@@ -36,3 +36,7 @@ SCRAPE_TYPE_ORDER = [ScrapeType.NORMAL, ScrapeType.NORMAL, ScrapeType.RELOAD]
 DEFAULT_MAX_TOKENS = 100000
 MAX_IMAGE_MESSAGES = 10
 SCROLL_AMOUNT_MULTIPLIER = 100
+
+# Text input constants
+TEXT_INPUT_DELAY = 10  # 10ms between each character input
+TEXT_PRESS_MAX_LENGTH = 20

skyvern/webeye/actions/handler.py

@@ -73,7 +73,7 @@ from skyvern.forge.sdk.services.bitwarden import BitwardenConstants
 from skyvern.forge.sdk.services.credentials import OnePasswordConstants
 from skyvern.schemas.runs import CUA_RUN_TYPES
 from skyvern.utils.prompt_engine import CheckPhoneNumberFormatResponse, load_prompt_with_elements
-from skyvern.webeye.actions import actions
+from skyvern.webeye.actions import actions, handler_utils
 from skyvern.webeye.actions.action_types import ActionType
 from skyvern.webeye.actions.actions import (
     Action,
@@ -88,7 +88,6 @@ from skyvern.webeye.actions.actions import (
     UploadFileAction,
     WebAction,
 )
-from skyvern.webeye.actions.handler_utils import download_file_safe
 from skyvern.webeye.actions.responses import ActionAbort, ActionFailure, ActionResult, ActionSuccess
 from skyvern.webeye.scraper.scraper import (
     CleanupElementTreeFunc,
@@ -1888,7 +1887,7 @@ async def chain_click(
     file: list[str] | str = []
     if action.file_url:
         file_url = await get_actual_value_of_parameter_if_secret(task, action.file_url)
-        file = await download_file_safe(file_url, action.model_dump())
+        file = await handler_utils.download_file(file_url, action.model_dump())
 
     is_filechooser_trigger = False
 

skyvern/webeye/actions/handler_utils.py

@@ -1,15 +1,18 @@
 from typing import Any
 
 import structlog
+from playwright.async_api import Locator
 
-from skyvern.forge.sdk.api.files import download_file
+from skyvern.config import settings
+from skyvern.constants import TEXT_INPUT_DELAY, TEXT_PRESS_MAX_LENGTH
+from skyvern.forge.sdk.api.files import download_file as download_file_api
 
 LOG = structlog.get_logger()
 
 
-async def download_file_safe(file_url: str, action: dict[str, Any] | None = None) -> str | list[str]:
+async def download_file(file_url: str, action: dict[str, Any] | None = None) -> str | list[str]:
     try:
-        return await download_file(file_url)
+        return await download_file_api(file_url)
     except Exception:
         LOG.exception(
             "Failed to download file, continuing without it",
@@ -17,3 +20,14 @@ async def download_file_safe(file_url: str, action: dict[str, Any] | None = None
             file_url=file_url,
         )
         return []
+
+
+async def input_sequentially(locator: Locator, text: str, timeout: float = settings.BROWSER_ACTION_TIMEOUT_MS) -> None:
+    length = len(text)
+    if length > TEXT_PRESS_MAX_LENGTH:
+        # if the text is longer than TEXT_PRESS_MAX_LENGTH characters, we will locator.fill in initial texts until the last TEXT_PRESS_MAX_LENGTH characters
+        # and then type the last TEXT_PRESS_MAX_LENGTH characters with locator.press_sequentially
+        await locator.fill(text, timeout=timeout)
+        text = text[length - TEXT_PRESS_MAX_LENGTH :]
+
+    await locator.press_sequentially(text, delay=TEXT_INPUT_DELAY, timeout=timeout)

skyvern/webeye/utils/dom.py

@@ -11,7 +11,7 @@ import structlog
 from playwright.async_api import ElementHandle, Frame, FrameLocator, Locator, Page, TimeoutError
 
 from skyvern.config import settings
-from skyvern.constants import SKYVERN_ID_ATTR
+from skyvern.constants import SKYVERN_ID_ATTR, TEXT_INPUT_DELAY
 from skyvern.exceptions import (
     ElementIsNotLabel,
     InteractWithDisabledElement,
@@ -24,15 +24,13 @@ from skyvern.exceptions import (
     NoneFrameError,
     SkyvernException,
 )
+from skyvern.webeye.actions import handler_utils
 from skyvern.webeye.scraper.scraper import IncrementalScrapePage, ScrapedPage, json_to_html, trim_element
 from skyvern.webeye.utils.page import SkyvernFrame
 
 LOG = structlog.get_logger()
 COMMON_INPUT_TAGS = {"input", "textarea", "select"}
 
-TEXT_INPUT_DELAY = 10  # 10ms between each character input
-TEXT_PRESS_MAX_LENGTH = 20
-
 
 async def resolve_locator(scrape_page: ScrapedPage, page: Page, frame: str, css: str) -> tuple[Locator, Page | Frame]:
     iframe_path: list[str] = []
@@ -574,14 +572,7 @@ class SkyvernElement:
         await self.get_locator().focus(timeout=timeout)
 
     async def input_sequentially(self, text: str, default_timeout: float = settings.BROWSER_ACTION_TIMEOUT_MS) -> None:
-        length = len(text)
+        await handler_utils.input_sequentially(self.get_locator(), text, timeout=default_timeout)
-        if length > TEXT_PRESS_MAX_LENGTH:
-            # if the text is longer than TEXT_PRESS_MAX_LENGTH characters, we will locator.fill in initial texts until the last TEXT_PRESS_MAX_LENGTH characters
-            # and then type the last TEXT_PRESS_MAX_LENGTH characters with locator.press_sequentially
-            await self.input_fill(text[: length - TEXT_PRESS_MAX_LENGTH])
-            text = text[length - TEXT_PRESS_MAX_LENGTH :]
-
-        await self.press_fill(text, timeout=default_timeout)
 
     async def press_key(self, key: str, timeout: float = settings.BROWSER_ACTION_TIMEOUT_MS) -> None:
         await self.get_locator().press(key=key, timeout=timeout)