This commit is contained in:
@@ -188,84 +188,21 @@ async def interactions_service(
|
||||
|
||||
async def interactions_operation(page):
|
||||
try:
|
||||
# When the API caller sets viewport.height = -1, treat that as "capture full webpage".
|
||||
# We implement this by resizing the viewport to the document's scroll height just
|
||||
# before the final screenshot. (This keeps existing coordinate-click behavior intact.)
|
||||
full_page_via_viewport = False
|
||||
requested_viewport_width = None
|
||||
# Interactions endpoint policy:
|
||||
# - Disable all scrolling features (ignore scrollX/scrollY and do not auto-scroll).
|
||||
# - Force viewport width to 1280.
|
||||
# - Return a full-height screenshot (full_page=True) when return_type != "html".
|
||||
forced_viewport_width = 1280
|
||||
|
||||
async def resolve_viewport_point(px: int, py: int) -> tuple[float, float]:
|
||||
"""
|
||||
`document.elementFromPoint(x, y)` uses *viewport* (CSS pixel) coordinates.
|
||||
|
||||
This API also supports *document* coordinates (e.g. from a full-page
|
||||
screenshot). In that case we must scroll so the point is inside the viewport,
|
||||
then convert to viewport coordinates by subtracting the current scroll offsets.
|
||||
With scrolling disabled, we keep the page scrolled to the top and expand the
|
||||
viewport height to the document height (best-effort). That makes document and
|
||||
viewport coordinates equivalent for typical pages.
|
||||
"""
|
||||
x = float(px)
|
||||
y = float(py)
|
||||
|
||||
vp = await page.evaluate(
|
||||
"() => ({ innerWidth: window.innerWidth, innerHeight: window.innerHeight, scrollX: window.scrollX, scrollY: window.scrollY })"
|
||||
)
|
||||
inner_w = float(vp.get("innerWidth") or 0)
|
||||
inner_h = float(vp.get("innerHeight") or 0)
|
||||
cur_scroll_x = float(vp.get("scrollX") or 0)
|
||||
cur_scroll_y = float(vp.get("scrollY") or 0)
|
||||
|
||||
# If the point is plausibly already a viewport coordinate, use it directly.
|
||||
# (We allow a small slack to account for fractional/rounded coordinates.)
|
||||
if inner_w and inner_h and (-5.0 <= x <= inner_w + 5.0) and (-5.0 <= y <= inner_h + 5.0):
|
||||
return x, y
|
||||
|
||||
# Otherwise treat as document coordinates and auto-scroll to bring it into view.
|
||||
# We may need to iterate because some pages clamp/adjust scroll requests.
|
||||
if inner_w and inner_h:
|
||||
for _ in range(6):
|
||||
metrics = await page.evaluate(
|
||||
"() => ({ innerWidth: window.innerWidth, innerHeight: window.innerHeight, scrollX: window.scrollX, scrollY: window.scrollY, scrollW: Math.max(document.body?.scrollWidth||0, document.documentElement?.scrollWidth||0), scrollH: Math.max(document.body?.scrollHeight||0, document.documentElement?.scrollHeight||0) })"
|
||||
)
|
||||
inner_w = float(metrics.get("innerWidth") or inner_w or 0)
|
||||
inner_h = float(metrics.get("innerHeight") or inner_h or 0)
|
||||
cur_scroll_x = float(metrics.get("scrollX") or 0)
|
||||
cur_scroll_y = float(metrics.get("scrollY") or 0)
|
||||
scroll_w = float(metrics.get("scrollW") or 0)
|
||||
scroll_h = float(metrics.get("scrollH") or 0)
|
||||
|
||||
max_scroll_x = max(0.0, scroll_w - inner_w) if inner_w else 0.0
|
||||
max_scroll_y = max(0.0, scroll_h - inner_h) if inner_h else 0.0
|
||||
|
||||
vx = x - cur_scroll_x
|
||||
vy = y - cur_scroll_y
|
||||
|
||||
# If the mapped viewport point is inside the viewport, we're done.
|
||||
if (0.0 <= vx <= max(inner_w - 1.0, 0.0)) and (0.0 <= vy <= max(inner_h - 1.0, 0.0)):
|
||||
# Avoid exact-edge coordinates which sometimes hit nothing (scrollbars).
|
||||
vx = min(max(vx, 0.5), max(inner_w - 1.5, 0.5))
|
||||
vy = min(max(vy, 0.5), max(inner_h - 1.5, 0.5))
|
||||
return vx, vy
|
||||
|
||||
# Scroll so the point lands near the center of the viewport.
|
||||
target_scroll_x = x - (inner_w / 2.0)
|
||||
target_scroll_y = y - (inner_h / 2.0)
|
||||
|
||||
target_scroll_x = min(max(target_scroll_x, 0.0), max_scroll_x)
|
||||
target_scroll_y = min(max(target_scroll_y, 0.0), max_scroll_y)
|
||||
|
||||
await page.evaluate(
|
||||
"""({ sx, sy }) => {
|
||||
window.scrollTo({ left: sx, top: sy, behavior: 'auto' });
|
||||
}""",
|
||||
{"sx": target_scroll_x, "sy": target_scroll_y},
|
||||
)
|
||||
|
||||
# Give the browser a moment to reflow after scroll.
|
||||
await page.wait_for_timeout(150)
|
||||
|
||||
# If we couldn't get reliable viewport metrics, fall back to best-effort conversion.
|
||||
vx = x - cur_scroll_x
|
||||
vy = y - cur_scroll_y
|
||||
return vx, vy
|
||||
return float(px), float(py)
|
||||
|
||||
async def element_from_point_retry(vx: float, vy: float, attempts: int = 10):
|
||||
"""Retry `elementFromPoint` to avoid timing issues around scroll/render."""
|
||||
@@ -296,27 +233,12 @@ async def interactions_service(
|
||||
await page.wait_for_timeout(100)
|
||||
raise ValueError(last_error or f"No element found at viewport point ({vx}, {vy})")
|
||||
|
||||
# Configure viewport if provided (before navigation).
|
||||
if viewport is not None:
|
||||
# Force viewport width to 1280 (ignore payload viewport for interactions endpoint).
|
||||
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)
|
||||
|
||||
requested_viewport_width = vp_width
|
||||
if vp_width and vp_height:
|
||||
if int(vp_height) == -1:
|
||||
# Defer height calculation until after navigation/interactions.
|
||||
full_page_via_viewport = True
|
||||
vp_size = page.viewport_size or {"width": 1280, "height": 720}
|
||||
await page.set_viewport_size({"width": int(vp_width), "height": int(vp_size.get("height", 720))})
|
||||
else:
|
||||
await page.set_viewport_size({"width": int(vp_width), "height": int(vp_height)})
|
||||
vp_size = page.viewport_size or {"width": forced_viewport_width, "height": 720}
|
||||
await page.set_viewport_size({"width": forced_viewport_width, "height": int(vp_size.get("height", 720))})
|
||||
except Exception as e:
|
||||
print(f"Error setting viewport: {e}")
|
||||
print(f"Error setting forced viewport width: {e}")
|
||||
|
||||
# Navigate and wait for initial DOM.
|
||||
await page.goto(decoded_url, wait_until="domcontentloaded", timeout=30000)
|
||||
@@ -342,60 +264,25 @@ async def interactions_service(
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Apply initial scroll position if provided.
|
||||
# Do this AFTER cookie/banner clicks since those can change scroll position.
|
||||
if scroll_x is not None or scroll_y is not None:
|
||||
# Disable scrolling: always keep page at the top.
|
||||
try:
|
||||
await page.evaluate(
|
||||
"""(x, y) => {
|
||||
const left = (x === null ? window.scrollX : x);
|
||||
const top = (y === null ? window.scrollY : y);
|
||||
window.scrollTo({ left, top, behavior: 'auto' });
|
||||
}""",
|
||||
scroll_x if scroll_x is not None else None,
|
||||
scroll_y if scroll_y is not None else None,
|
||||
)
|
||||
|
||||
# Wait for scroll offsets to actually settle before doing coordinate clicks.
|
||||
if scroll_x is not None:
|
||||
await page.wait_for_function(
|
||||
"(x) => Math.abs(window.scrollX - x) <= 2",
|
||||
scroll_x,
|
||||
timeout=10000,
|
||||
)
|
||||
if scroll_y is not None:
|
||||
await page.wait_for_function(
|
||||
"(y) => Math.abs(window.scrollY - y) <= 2",
|
||||
scroll_y,
|
||||
timeout=10000,
|
||||
)
|
||||
await page.wait_for_timeout(300)
|
||||
except Exception as e:
|
||||
print(f"Error applying initial scroll position: {e}")
|
||||
|
||||
# If the caller requested full-page via viewport (viewport.height = -1), do the
|
||||
# viewport expansion *now* so coordinate-based interactions can use viewport
|
||||
# coordinates without requiring scrolling.
|
||||
if full_page_via_viewport:
|
||||
try:
|
||||
max_vp_h = 16384
|
||||
|
||||
# Expand viewport height to the document height (capped), and reset scroll.
|
||||
await page.evaluate("() => window.scrollTo({ left: 0, top: 0, behavior: 'auto' })")
|
||||
await page.wait_for_timeout(250)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Best-effort: expand viewport height to the document height (capped by Chromium).
|
||||
# This keeps coordinate interactions working without scrolling.
|
||||
try:
|
||||
max_vp_h = 16384
|
||||
full_h = await page.evaluate(
|
||||
"() => Math.max(document.body?.scrollHeight || 0, document.documentElement?.scrollHeight || 0)"
|
||||
)
|
||||
target_h = int(min(max(int(full_h or 0), 1), max_vp_h))
|
||||
|
||||
cur_vp = page.viewport_size or {"width": 1280, "height": 720}
|
||||
target_w = int(requested_viewport_width or cur_vp.get("width", 1280))
|
||||
|
||||
await page.set_viewport_size({"width": target_w, "height": target_h})
|
||||
await page.wait_for_timeout(500)
|
||||
await page.set_viewport_size({"width": forced_viewport_width, "height": target_h})
|
||||
await page.wait_for_timeout(250)
|
||||
except Exception as e:
|
||||
print(f"Error resizing viewport for full-page interactions: {e}")
|
||||
print(f"Error resizing viewport for full-height interactions: {e}")
|
||||
|
||||
for interaction in interactions:
|
||||
# Support both Pydantic model instances and raw dicts.
|
||||
@@ -422,17 +309,34 @@ async def interactions_service(
|
||||
if selector:
|
||||
if action == "click":
|
||||
await page.wait_for_selector(selector, timeout=30000)
|
||||
locator = page.locator(selector)
|
||||
await locator.scroll_into_view_if_needed(timeout=30000)
|
||||
await locator.click(force=True, timeout=30000)
|
||||
# Avoid Playwright auto-scrolling: click via DOM.
|
||||
await page.evaluate(
|
||||
"""(sel) => {
|
||||
const el = document.querySelector(sel);
|
||||
if (!el) throw new Error(`No element for selector: ${sel}`);
|
||||
el.click();
|
||||
}""",
|
||||
selector,
|
||||
)
|
||||
|
||||
elif action == "type":
|
||||
if text is None:
|
||||
raise ValueError("Type interaction must contain text")
|
||||
await page.wait_for_selector(selector, timeout=30000)
|
||||
locator = page.locator(selector)
|
||||
await locator.scroll_into_view_if_needed(timeout=30000)
|
||||
await locator.fill(text)
|
||||
# Avoid Playwright auto-scrolling: focus/fill via DOM.
|
||||
await page.evaluate(
|
||||
"""({ sel, value }) => {
|
||||
const el = document.querySelector(sel);
|
||||
if (!el) throw new Error(`No element for selector: ${sel}`);
|
||||
el.focus?.();
|
||||
if ('value' in el) {
|
||||
el.value = value;
|
||||
el.dispatchEvent(new Event('input', { bubbles: true }));
|
||||
el.dispatchEvent(new Event('change', { bubbles: true }));
|
||||
}
|
||||
}""",
|
||||
{"sel": selector, "value": text},
|
||||
)
|
||||
|
||||
else:
|
||||
raise ValueError(f"Unknown interaction action: {action}")
|
||||
@@ -491,31 +395,8 @@ async def interactions_service(
|
||||
if return_type == "html":
|
||||
return await page.content()
|
||||
|
||||
if full_page_via_viewport:
|
||||
try:
|
||||
# Chromium has practical limits on viewport height; cap to avoid errors.
|
||||
max_vp_h = 16384
|
||||
|
||||
# Ensure we capture from the top of the document.
|
||||
await page.evaluate("() => window.scrollTo({ left: 0, top: 0, behavior: 'auto' })")
|
||||
await page.wait_for_timeout(250)
|
||||
|
||||
full_h = await page.evaluate(
|
||||
"() => Math.max(document.body?.scrollHeight || 0, document.documentElement?.scrollHeight || 0)"
|
||||
)
|
||||
target_h = int(min(max(full_h, 1), max_vp_h))
|
||||
|
||||
cur_vp = page.viewport_size or {"width": 1280, "height": 720}
|
||||
target_w = int(requested_viewport_width or cur_vp.get("width", 1280))
|
||||
|
||||
await page.set_viewport_size({"width": target_w, "height": target_h})
|
||||
await page.wait_for_timeout(500)
|
||||
except Exception as e:
|
||||
print(f"Error resizing viewport for full-page capture: {e}")
|
||||
|
||||
# For coordinate-based workflows, default is viewport screenshot so pixel coords align.
|
||||
# If full_page_via_viewport is enabled, the viewport has been expanded to the full document height.
|
||||
return await page.screenshot(full_page=False, type="png")
|
||||
# Always return full-page screenshot (full height).
|
||||
return await page.screenshot(full_page=True, type="png")
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error during interactions capture: {e}")
|
||||
|
||||
Reference in New Issue
Block a user