This commit is contained in:
@@ -18,13 +18,17 @@ router = APIRouter()
|
||||
|
||||
class ClickInteraction(BaseModel):
|
||||
action: Literal["click"]
|
||||
selector: str
|
||||
selector: Optional[str] = None
|
||||
x: Optional[int] = None
|
||||
y: Optional[int] = None
|
||||
|
||||
|
||||
class TypeInteraction(BaseModel):
|
||||
action: Literal["type"]
|
||||
selector: str
|
||||
selector: Optional[str] = None
|
||||
text: str
|
||||
x: Optional[int] = None
|
||||
y: Optional[int] = None
|
||||
|
||||
|
||||
Interaction = Annotated[
|
||||
@@ -33,9 +37,17 @@ Interaction = Annotated[
|
||||
]
|
||||
|
||||
|
||||
class Viewport(BaseModel):
|
||||
width: int
|
||||
height: int
|
||||
|
||||
|
||||
class InteractionsRequest(BaseModel):
|
||||
url: str
|
||||
returnType: Optional[Literal["screenshot", "html"]] = "screenshot"
|
||||
viewport: Optional[Viewport] = None
|
||||
scrollX: Optional[int] = None
|
||||
scrollY: Optional[int] = None
|
||||
interactions: List[Interaction]
|
||||
|
||||
@router.get("/")
|
||||
@@ -99,7 +111,14 @@ async def run_interactions(payload: InteractionsRequest, x_api_key: Optional[str
|
||||
return_type = (payload.returnType or "screenshot").lower()
|
||||
|
||||
try:
|
||||
result = await interactions_service(decoded_url, payload.interactions, return_type)
|
||||
result = await interactions_service(
|
||||
decoded_url,
|
||||
payload.interactions,
|
||||
return_type,
|
||||
viewport=payload.viewport,
|
||||
scroll_x=payload.scrollX,
|
||||
scroll_y=payload.scrollY,
|
||||
)
|
||||
|
||||
if isinstance(result, dict) and result.get("status") == "error":
|
||||
raise HTTPException(status_code=500, detail=result.get("error", "Interaction failed"))
|
||||
|
||||
@@ -175,12 +175,34 @@ async def click_selector_service(decoded_url: str, selector: str, return_type: s
|
||||
|
||||
return await safe_browser_operation(decoded_url, click_operation)
|
||||
|
||||
async def interactions_service(decoded_url: str, interactions, return_type: str):
|
||||
async def interactions_service(
|
||||
decoded_url: str,
|
||||
interactions,
|
||||
return_type: str,
|
||||
viewport=None,
|
||||
scroll_x: int | None = None,
|
||||
scroll_y: int | None = None,
|
||||
):
|
||||
"""Open `decoded_url`, execute click/type interactions in order, then return screenshot or HTML."""
|
||||
print(f"Running interactions on {decoded_url} (return_type={return_type})")
|
||||
|
||||
async def interactions_operation(page):
|
||||
try:
|
||||
# Configure viewport if provided (before navigation).
|
||||
if viewport is not None:
|
||||
try:
|
||||
if isinstance(viewport, dict):
|
||||
vp_width = viewport.get("width")
|
||||
vp_height = viewport.get("height")
|
||||
else:
|
||||
vp_width = getattr(viewport, "width", None)
|
||||
vp_height = getattr(viewport, "height", None)
|
||||
|
||||
if vp_width and vp_height:
|
||||
await page.set_viewport_size({"width": vp_width, "height": vp_height})
|
||||
except Exception as e:
|
||||
print(f"Error setting viewport: {e}")
|
||||
|
||||
# Navigate and wait for initial DOM.
|
||||
await page.goto(decoded_url, wait_until="domcontentloaded", timeout=30000)
|
||||
await page.wait_for_load_state("load", timeout=30000)
|
||||
@@ -188,6 +210,17 @@ async def interactions_service(decoded_url: str, interactions, return_type: str)
|
||||
# Ensure the page is fully loaded (helps with JS-driven UIs).
|
||||
await page.wait_for_function("document.readyState === 'complete'", timeout=30000)
|
||||
|
||||
# Apply initial scroll position if provided.
|
||||
if scroll_x is not None or scroll_y is not None:
|
||||
try:
|
||||
await page.evaluate(
|
||||
"(x, y) => { window.scrollTo(x ?? window.scrollX, y ?? window.scrollY); }",
|
||||
scroll_x,
|
||||
scroll_y,
|
||||
)
|
||||
except Exception as e:
|
||||
print(f"Error applying initial scroll position: {e}")
|
||||
|
||||
# Best-effort cookie/banner handling (helps avoid click interception).
|
||||
cookie_buttons = [
|
||||
"Accept all",
|
||||
@@ -215,10 +248,19 @@ async def interactions_service(decoded_url: str, interactions, return_type: str)
|
||||
action = getattr(interaction, "action", None)
|
||||
selector = getattr(interaction, "selector", None)
|
||||
text = getattr(interaction, "text", None)
|
||||
# Coordinates may be present on interaction models.
|
||||
x = getattr(interaction, "x", None)
|
||||
y = getattr(interaction, "y", None)
|
||||
|
||||
if not action or not selector:
|
||||
raise ValueError("Interaction must contain action and selector")
|
||||
if isinstance(interaction, dict):
|
||||
x = interaction.get("x")
|
||||
y = interaction.get("y")
|
||||
|
||||
if not action:
|
||||
raise ValueError("Interaction must contain an action")
|
||||
|
||||
# Prefer selector-based interactions when selector is provided.
|
||||
if selector:
|
||||
if action == "click":
|
||||
await page.wait_for_selector(selector, timeout=30000)
|
||||
locator = page.locator(selector)
|
||||
@@ -236,6 +278,23 @@ async def interactions_service(decoded_url: str, interactions, return_type: str)
|
||||
else:
|
||||
raise ValueError(f"Unknown interaction action: {action}")
|
||||
|
||||
# Fallback to coordinate-based interactions when no selector is given.
|
||||
elif x is not None and y is not None:
|
||||
if action == "click":
|
||||
await page.mouse.click(x, y)
|
||||
|
||||
elif action == "type":
|
||||
if text is None:
|
||||
raise ValueError("Type interaction must contain text")
|
||||
await page.mouse.click(x, y)
|
||||
await page.keyboard.type(text)
|
||||
|
||||
else:
|
||||
raise ValueError(f"Unknown interaction action: {action}")
|
||||
|
||||
else:
|
||||
raise ValueError("Interaction must contain either selector or x/y coordinates")
|
||||
|
||||
# Let the UI settle after each action.
|
||||
await page.wait_for_timeout(1000)
|
||||
try:
|
||||
|
||||
Reference in New Issue
Block a user