opencode/yao
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge pull request #1452 from trheyi/main

Browse source 
Add Chrome support to sandbox module and update documentation
This commit is contained in:
Max 2026-02-06 17:49:21 +08:00 • committed by GitHub
commit 9eeec111b9
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
10 changed files with 1878 additions and 8 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
.gitignore vendored
View file

@ -64,3 +64,4 @@ sandbox/docker/claude/claude-proxy-*
sandbox/proxy/claude-proxy-linux-*
release/*
sandbox/TODO-VNC.md
sandbox/docker/chrome/PLAN.md

19
sandbox/README.md
View file

@ -37,13 +37,13 @@ The sandbox module enables Yao to safely run external AI coding agents (like Cla
│ │ │
│ ┌───────────────┼───────────────┐ │
│ ▼ ▼ ▼ │
│ ┌────────────┐ ┌────────────┐ ┌────────────┐ │
│ │ sandbox- │ │ sandbox- │ │ sandbox- │ │
│ │ claude │ │ playwright │ │ desktop │ │
│ │ (No VNC) │ │ (VNC) │ │ (VNC) │ │
│ └─────┬──────┘ └─────┬──────┘ └─────┬──────┘ │
│ │ │ │ │
│ ──────┴───────────────┴───────────────┴──── │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ sandbox- │ │ sandbox- │ │ sandbox- │ │ sandbox- │ │
│ │ claude │ │ browser │ │ desktop │ │ chrome │ │
│ │ (No VNC) │ │ (VNC) │ │ (VNC) │ │ (VNC) │ │
│ └────┬─────┘ └────┬─────┘ └────┬─────┘ └────┬─────┘ │
│ │ │ │ │ │
│ ─────┴────────────┴────────────┴────────────┴──── │
│ Unix Socket IPC │
│ (one socket per container) │
└───────────────────────────────────────────────────────────────┘
@ -62,6 +62,7 @@ cd sandbox/docker
# Build VNC-enabled images
./build.sh browser # Browser (Playwright) + Fluxbox + VNC
./build.sh desktop # XFCE Desktop + VNC
./build.sh chrome # Real Chrome + CDP + VNC (amd64 only)
# Build all images
./build.sh all
@ -134,6 +135,7 @@ When enabled, VNC ports (6080, 5900) are automatically mapped to random availabl
| `yaoapp/sandbox-claude:full` | ❌ | + Go 1.23 |
| `yaoapp/sandbox-claude-browser:latest` | ✅ | + Playwright, Fluxbox, VNC (~3.4GB) |
| `yaoapp/sandbox-claude-desktop:latest` | ✅ | + XFCE Desktop, VNC (~3.1GB) |
| `yaoapp/sandbox-claude-chrome:latest` | ✅ | + Real Chrome, CDP, PyAutoGUI, VNC (~3.4GB, amd64 only) |
## IPC Communication
@ -174,6 +176,9 @@ sandbox/
│ ├── claude/
│ ├── browser/ # Browser (Playwright) + VNC image
│ ├── desktop/ # XFCE Desktop + VNC image
│ ├── chrome/ # Real Chrome + CDP + VNC image (amd64 only)
│ │ ├── config/ # Chrome preferences, stealth scripts
│ │ └── tests/ # LLM-driven browser automation demos
│ ├── vnc/ # Shared VNC scripts
│ └── build.sh
├── ipc/ # IPC system

33
sandbox/docker/build.sh
View file

@ -67,6 +67,25 @@ setup_buildx() {
fi
}
# Build single-arch image (amd64 only, for Chrome which has no arm64 build)
build_amd64() {
local IMAGE_NAME=$1
local DOCKERFILE=$2
local PUSH_FLAG=$3
echo ""
echo "=== Building $IMAGE_NAME (linux/amd64 only) ==="
BUILD_ARGS="--platform linux/amd64 -t ${REGISTRY}/${IMAGE_NAME}:latest"
if [ "$PUSH_FLAG" = "true" ]; then
BUILD_ARGS="$BUILD_ARGS --push"
else
BUILD_ARGS="$BUILD_ARGS --load"
fi
docker buildx build $BUILD_ARGS -f "$DOCKERFILE" .
}
# Build multi-arch image
build_multiarch() {
local IMAGE_NAME=$1
@ -121,6 +140,11 @@ case $TOOL in
echo "=== Building Claude Desktop image ==="
build_multiarch "sandbox-claude-desktop" "desktop/Dockerfile" "$PUSH"
;;
chrome)
echo ""
echo "=== Building Claude Chrome image (amd64 only) ==="
build_amd64 "sandbox-claude-chrome" "chrome/Dockerfile" "$PUSH"
;;
cursor)
echo ""
echo "=== Building Cursor images ==="
@ -135,17 +159,20 @@ case $TOOL in
# Claude VNC variants
build_multiarch "sandbox-claude-browser" "browser/Dockerfile" "$PUSH"
build_multiarch "sandbox-claude-desktop" "desktop/Dockerfile" "$PUSH"
# Chrome (amd64 only - Google Chrome has no arm64 Linux build)
build_amd64 "sandbox-claude-chrome" "chrome/Dockerfile" "$PUSH"
# Cursor (uncomment when ready)
# build_multiarch "sandbox-cursor" "cursor/Dockerfile" "$PUSH"
;;
*)
echo "Unknown tool: $TOOL"
echo "Usage: $0 [claude|claude-vnc|browser|desktop|cursor|all] [true|false]"
echo "Usage: $0 [claude|claude-vnc|browser|desktop|chrome|cursor|all] [true|false]"
echo " $0 claude # Build Claude images locally"
echo " $0 claude true # Build and push Claude images"
echo " $0 claude-vnc # Build Claude VNC images (Browser + Desktop)"
echo " $0 browser # Build Claude Browser image only"
echo " $0 desktop # Build Claude Desktop image only"
echo " $0 chrome # Build Claude Chrome image (amd64 only)"
echo " $0 all true # Build and push all images"
exit 1
;;
@ -174,11 +201,15 @@ if [ "$PUSH" = "true" ]; then
desktop)
echo " - ${REGISTRY}/sandbox-claude-desktop:latest"
;;
chrome)
echo " - ${REGISTRY}/sandbox-claude-chrome:latest"
;;
all)
echo " - ${REGISTRY}/sandbox-claude:latest"
echo " - ${REGISTRY}/sandbox-claude-full:latest"
echo " - ${REGISTRY}/sandbox-claude-browser:latest"
echo " - ${REGISTRY}/sandbox-claude-desktop:latest"
echo " - ${REGISTRY}/sandbox-claude-chrome:latest"
;;
esac
fi

166
sandbox/docker/chrome/Dockerfile Normal file
View file

@ -0,0 +1,166 @@
# Claude sandbox with real Google Chrome + anti-detection stealth
# Image: sandbox-claude-chrome
# Base: sandbox-claude (Ubuntu 24.04 + Node.js + Python + Claude CLI)
# Adds: Xvfb + x11vnc + noVNC + Fluxbox + Real Chrome + Patchright + PyAutoGUI
#
# Anti-bot detection browser environment for web research tasks
# amd64 architecture only (Google Chrome has no official arm64 Linux build)
ARG REGISTRY=yaoapp
FROM ${REGISTRY}/sandbox-claude:latest
USER root
# ============================================
# 1. VNC + Window Manager (same as browser image)
# ============================================
RUN apt-get update && apt-get install -y --no-install-recommends \
# Sudo for sandbox user
sudo \
# Virtual display
xvfb \
# VNC server
x11vnc \
# noVNC (HTML5 VNC client) and websockify
novnc \
python3-websockify \
# Minimal window manager
fluxbox \
# Background/wallpaper utilities
feh \
imagemagick \
# Fonts (required for proper browser rendering)
fonts-liberation \
fonts-noto-cjk \
fonts-noto-color-emoji \
# X11 utilities
x11-utils \
xdotool \
# Audio (for video playback)
pulseaudio \
# PyAutoGUI X11 dependencies
python3-tk \
python3-dev \
scrot \
# Misc
xterm \
&& rm -rf /var/lib/apt/lists/*
# Configure passwordless sudo for sandbox user
RUN echo "sandbox ALL=(ALL) NOPASSWD:ALL" > /etc/sudoers.d/sandbox && \
chmod 0440 /etc/sudoers.d/sandbox
# ============================================
# 2. Real Google Chrome (amd64 only)
# ============================================
RUN curl -fsSL https://dl.google.com/linux/linux_signing_key.pub \
| gpg --dearmor -o /usr/share/keyrings/google-chrome.gpg && \
echo "deb [arch=amd64 signed-by=/usr/share/keyrings/google-chrome.gpg] \
http://dl.google.com/linux/chrome/deb/ stable main" \
> /etc/apt/sources.list.d/google-chrome.list && \
apt-get update && apt-get install -y google-chrome-stable && \
rm -rf /var/lib/apt/lists/*
# ============================================
# 3. Playwright system deps (for Patchright compatibility)
# ============================================
RUN npx playwright install-deps chromium || true
# ============================================
# 4. Python anti-detection libraries
# ============================================
USER sandbox
# Install Patchright (stealth Playwright fork) + PyAutoGUI + stealth libs
RUN pip install --user --break-system-packages \
patchright \
pyautogui \
playwright-stealth \
playwright && \
# Install Patchright browser deps (uses system Chrome, no Chromium download)
python3 -m patchright install chromium || true
# Install Node.js Playwright + stealth plugins
RUN npm install -g playwright playwright-extra puppeteer-extra-plugin-stealth
USER root
# ============================================
# 5. Copy config files
# ============================================
RUN mkdir -p /usr/local/share/yao
# VNC startup scripts (shared with browser/desktop)
COPY vnc/start-vnc.sh /usr/local/bin/start-vnc.sh
COPY vnc/entrypoint-vnc.sh /usr/local/bin/entrypoint.sh
# Chrome-specific config files
COPY chrome/config/setup-fluxbox.sh /usr/local/bin/setup-fluxbox.sh
COPY chrome/config/chrome-stealth.sh /usr/local/bin/chrome-stealth
COPY chrome/config/stealth-init.js /usr/local/share/yao/stealth-init.js
COPY chrome/config/chrome-preferences.json /usr/local/share/yao/chrome-preferences.json
# Reuse yao-logo from browser image
COPY browser/config/yao-logo.png /usr/local/share/yao/yao-logo.png
RUN chmod +x /usr/local/bin/start-vnc.sh \
/usr/local/bin/entrypoint.sh \
/usr/local/bin/setup-fluxbox.sh \
/usr/local/bin/chrome-stealth
# ============================================
# 6. Default Chrome profile + X11 auth
# ============================================
RUN mkdir -p /home/sandbox/.config/google-chrome/Default && \
cp /usr/local/share/yao/chrome-preferences.json \
/home/sandbox/.config/google-chrome/Default/Preferences && \
# Mark first run as done
touch /home/sandbox/.config/google-chrome/First\ Run && \
# Create .Xauthority for PyAutoGUI (Xvfb runs without auth)
touch /home/sandbox/.Xauthority && \
chown -R sandbox:sandbox /home/sandbox/.config/google-chrome /home/sandbox/.Xauthority
# ============================================
# 7. Environment variables
# ============================================
ENV DISPLAY=:99
ENV VNC_PORT=5900
ENV NOVNC_PORT=6080
ENV RESOLUTION=1920x1080x24
ENV SANDBOX_VNC_ENABLED=true
ENV SANDBOX_DESKTOP=fluxbox
# Node.js environment
ENV NODE_PATH=/home/sandbox/.npm-global/lib/node_modules
# Timezone
ENV TZ=America/New_York
# Expose VNC ports (internal use only, accessed via proxy)
EXPOSE 5900 6080
USER sandbox
WORKDIR /workspace
# ============================================
# 8. Verify installations
# ============================================
RUN echo "=== Verifying installations ===" && \
google-chrome-stable --version && \
node --version && \
npm --version && \
python3 --version && \
python3 -c "from patchright.sync_api import sync_playwright; print('Patchright: OK')" && \
python3 -c "from playwright.sync_api import sync_playwright; print('Playwright: OK')" && \
python3 -c "from playwright_stealth import Stealth; print('Playwright-Stealth: OK')" && \
pip3 show pyautogui | head -2 && echo "PyAutoGUI: OK" && \
which fluxbox && \
which x11vnc && \
which Xvfb && \
which chrome-stealth && \
test -f /usr/local/share/yao/stealth-init.js && \
test -f /home/sandbox/.config/google-chrome/Default/Preferences && \
echo "=== All installations verified ==="
ENTRYPOINT ["/usr/local/bin/entrypoint.sh"]
CMD ["sleep", "infinity"]

34
sandbox/docker/chrome/config/chrome-preferences.json Normal file
View file

@ -0,0 +1,34 @@
{
"browser": {
"enabled_labs_experiments": ["disable-search-engine-collection@2"],
"check_default_browser": false,
"has_seen_welcome_page": true
},
"profile": {
"default_content_setting_values": {
"notifications": 2
}
},
"credentials_enable_service": false,
"translate": {
"enabled": false
},
"intl": {
"accept_languages": "en-US,en"
},
"distribution": {
"skip_first_run_ui": true,
"show_welcome_page": false,
"import_bookmarks": false,
"import_history": false,
"import_search_engine": false,
"suppress_first_run_bubble": true,
"do_not_create_desktop_shortcut": true,
"do_not_create_quick_launch_shortcut": true,
"do_not_create_taskbar_shortcut": true,
"do_not_launch_chrome": true,
"do_not_register_for_update_launch": true,
"make_chrome_default": false,
"make_chrome_default_for_user": false
}
}

44
sandbox/docker/chrome/config/stealth-init.js Normal file
View file

@ -0,0 +1,44 @@
// Yao Sandbox - Chrome Stealth Initialization Script
// Injected before page load to mask automation fingerprints
// Location: /usr/local/share/yao/stealth-init.js
// Remove webdriver flag
Object.defineProperty(navigator, 'webdriver', { get: () => undefined });
// Fake chrome.runtime (Chrome Extension API)
if (!window.chrome) window.chrome = {};
if (!window.chrome.runtime) {
window.chrome.runtime = {
connect: function() {},
sendMessage: function() {},
onMessage: { addListener: function() {} },
id: undefined
};
}
// Fake navigator.plugins (simulate Chrome default plugins)
Object.defineProperty(navigator, 'plugins', {
get: () => [
{ name: 'Chrome PDF Plugin', filename: 'internal-pdf-viewer', description: 'Portable Document Format' },
{ name: 'Chrome PDF Viewer', filename: 'mhjfbmdgcfjbbpaeojofohoefgiehjai', description: '' },
{ name: 'Native Client', filename: 'internal-nacl-plugin', description: '' }
]
});
// Fake navigator.languages
Object.defineProperty(navigator, 'languages', { get: () => ['en-US', 'en'] });
// Fix permissions API behavior
const originalQuery = window.navigator.permissions.query;
window.navigator.permissions.query = (parameters) =>
parameters.name === 'notifications'
? Promise.resolve({ state: Notification.permission })
: originalQuery(parameters);
// WebGL vendor/renderer spoofing
const getParameter = WebGLRenderingContext.prototype.getParameter;
WebGLRenderingContext.prototype.getParameter = function(parameter) {
if (parameter === 37445) return 'Google Inc. (Intel)'; // UNMASKED_VENDOR_WEBGL
if (parameter === 37446) return 'ANGLE (Intel, Mesa Intel(R) UHD Graphics, OpenGL 4.6)'; // UNMASKED_RENDERER_WEBGL
return getParameter.call(this, parameter);
};

105
sandbox/docker/chrome/tests/README.md Normal file
View file

@ -0,0 +1,105 @@
# Chrome Browser Automation Demo Tests
Example scripts demonstrating browser automation inside the `sandbox-claude-chrome` Docker image.
## Scripts
| Script | Description |
|--------|-------------|
| `demo-llm-vision.py` | **LLM-driven universal automation** — works with any search engine, no hardcoded selectors. LLM reads page DOM and decides what to click. |
| `demo-baidu.py` | Baidu search demo — hardcoded selectors |
| `demo-duckduckgo.py` | DuckDuckGo search demo — hardcoded selectors |
## demo-llm-vision.py
The main demo. Uses a layered architecture where each component does what it's best at:
```
LLM reads HTML → returns CSS selectors → DOM locates elements → CDP clicks
```
- **Playwright**: Opens pages, extracts DOM, keyboard input
- **LLM**: Reads page structure, returns CSS selectors for target elements (any cheap text model works)
- **DOM**: Uses LLM's selectors to get precise bounding boxes
- **CDP**: Chrome DevTools Protocol mouse events (`isTrusted=true`) for clicking
No hardcoded selectors — LLM figures out the page structure dynamically. Works with Google, Bing, Baidu, DuckDuckGo, Sogou, and any other search engine.
### Key Features
- **Concurrent LLM Race**: DOM is split into chunks, sent to LLM concurrently. First valid response wins — faster than sequential.
- **CDP Click**: Uses `Input.dispatchMouseEvent` via Chrome DevTools Protocol. Coordinates match `bounding_box()` exactly, no offset issues.
- **Ctrl+Click New Tab**: Search results open in new tabs, keeping the results list intact for clicking more links.
- **Fallback Chain**: CDP click → PyAutoGUI OS-level click → Playwright `.click()` → form submit. Always gets through.
### Environment Variables
| Variable | Description |
|----------|-------------|
| `LLM_API_KEY` | API key for the LLM service |
| `LLM_API_BASE` | OpenAI-compatible endpoint URL |
| `LLM_MODEL` | Model name (e.g. `gpt-4o-mini`) |
### Quick Start
```bash
# Start the container
docker run -d --name chrome-test \
--platform linux/amd64 \
-p 6080:6080 \
yaoapp/sandbox-claude-chrome:latest
# Wait for VNC to start
sleep 5
# Copy the script
docker cp tests/demo-llm-vision.py chrome-test:/workspace/
# Run with any search engine
docker exec \
-e LLM_API_KEY="your-key" \
-e LLM_API_BASE="https://api.openai.com/v1/" \
-e LLM_MODEL="gpt-4o-mini" \
chrome-test bash -c \
'DISPLAY=:99 python3 /workspace/demo-llm-vision.py "https://www.bing.com" "Yao App Engine"'
```
Open `http://localhost:6080` in your browser to watch the automation in real-time via VNC.
### Tested Search Engines
| Engine | Status | Notes |
|--------|--------|-------|
| Bing | Passed | gpt-4o-mini, ~100s |
| Sogou | Passed | gpt-4o-mini, ~83s |
| DuckDuckGo | Passed | gpt-4o-mini, ~87s |
| Baidu | Passed | glm-4-7, ~160s |
| Google | Passed | May show CAPTCHA on shared IPs |
### Flow
```
Phase 1 Open search engine homepage
Phase 2 [LLM Race] Analyze homepage DOM → get input/button selectors
Phase 3 [CDP] Click search input, type query
Phase 4 [CDP] Click search button (fallback: Enter key → form submit)
Phase 5 [LLM Race] Analyze results page DOM → get link selector
Phase 7+ [CDP Ctrl+Click] Open results in new tabs, screenshot, close
```
## Screenshots
Each demo saves screenshots to `/workspace/` at key steps:
| File | Content |
|------|---------|
| `llm-01-homepage.png` | Search engine homepage |
| `llm-02-typed.png` | Query typed in search box |
| `llm-03-results.png` | Search results page |
| `llm-detail.png` | Result detail page (new tab) |
## Notes
- **Google** may show reCAPTCHA due to IP-based rate limiting. Use a clean IP or proxy.
- **Model choice**: `gpt-4o-mini` recommended for speed. Slower models (e.g. `glm-4-7`) may timeout on large DOMs.
- **Concurrent Race** splits DOM into ~2000-char chunks and sends all chunks + full DOM to LLM simultaneously. First valid JSON response wins.

349
sandbox/docker/chrome/tests/demo-baidu.py Normal file
View file

@ -0,0 +1,349 @@
"""
Baidu Search Demo — PyAutoGUI OS-level Mouse + Smart Keyboard Fallback
Demonstrates anti-detection browser automation inside sandbox-claude-chrome:
1. Open Baidu homepage
2. Click search box with PyAutoGUI (real OS mouse event)
3. Type query with PyAutoGUI keyboard (auto-fallback to Playwright if needed)
4. Submit search
5. Click first search result (PyAutoGUI mouse) → view detail page
6. Go back
7. Click second search result (PyAutoGUI mouse) → view detail page
All mouse clicks are OS-level X11 events via PyAutoGUI — undetectable by websites.
Prerequisites:
- Running inside sandbox-claude-chrome container
- DISPLAY=:99 (Xvfb virtual display)
- VNC optional for live observation (http://localhost:6080)
Usage:
DISPLAY=:99 python3 demo-baidu.py
"""
from playwright.sync_api import sync_playwright
from playwright_stealth import Stealth
import pyautogui
import time
import random
import os
# ---------------------------------------------------------------------------
# PyAutoGUI config
# ---------------------------------------------------------------------------
pyautogui.FAILSAFE = False
pyautogui.PAUSE = 0.1
# Screenshot output directory
SCREENSHOT_DIR = os.environ.get("SCREENSHOT_DIR", "/workspace")
# ---------------------------------------------------------------------------
# Human-like helpers
# ---------------------------------------------------------------------------
def human_move(x, y):
"""Move mouse with randomized speed to simulate human behavior."""
duration = random.uniform(0.4, 0.8)
pyautogui.moveTo(x, y, duration=duration)
time.sleep(random.uniform(0.1, 0.3))
def human_click(x, y):
"""Move to (x, y) then click — mimics a real user click."""
human_move(x, y)
time.sleep(random.uniform(0.05, 0.15))
pyautogui.click()
time.sleep(random.uniform(0.2, 0.5))
def smart_type(element, text):
"""Type text with PyAutoGUI first; fallback to Playwright if it didn't land.
On native amd64 Linux, PyAutoGUI keyboard works perfectly.
On ARM Mac (Rosetta 2), X11 keyboard events may not reach Chrome,
so we detect and automatically fallback to Playwright's type().
"""
# Attempt PyAutoGUI keyboard (OS-level X11 events)
for ch in text:
pyautogui.press(ch)
time.sleep(random.uniform(0.05, 0.12))
time.sleep(0.5)
# Verify input landed
actual = element.input_value()
if actual and len(actual) >= len(text) * 0.8:
print(" Keyboard: PyAutoGUI (OS-level) ✓", flush=True)
return
# Fallback: Playwright type()
print(" PyAutoGUI keyboard didn't land — fallback to Playwright", flush=True)
element.fill("")
element.type(text, delay=80)
actual = element.input_value()
print(" Keyboard: Playwright fallback — '{}'".format(actual), flush=True)
def find_element(page, selectors, min_width=50, timeout=2000):
"""Try multiple CSS selectors, return (element, bounding_box) or (None, None)."""
for selector in selectors:
try:
el = page.locator(selector).first
box = el.bounding_box(timeout=timeout)
if box and box["width"] >= min_width:
return el, box
except Exception:
continue
return None, None
def find_results(page, selectors, min_count=2, min_width=100):
"""Find clickable search result elements with bounding boxes.
For Baidu: results are <h3><a href="..." target="_blank">title</a></h3>.
We need the <a> element — it's the actual clickable link.
"""
results = []
for selector in selectors:
elements = page.locator(selector).all()
for el in elements[:10]:
try:
box = el.bounding_box(timeout=1000)
title = el.text_content().strip()
if box and box["width"] >= min_width and title and len(title) > 5:
if not any(r["title"] == title for r in results):
results.append({"box": box, "title": title, "el": el})
except Exception:
pass
if len(results) >= min_count:
break
return results
def click_result(ctx, page, result, label):
"""Click a search result using PyAutoGUI and handle new tab navigation.
Baidu results have target=_blank, so clicking opens a new tab.
We scroll the element into view first, then use PyAutoGUI for the OS-level click.
"""
el = result["el"]
# Scroll element to a safe click zone.
# Baidu results page has a tall fixed search bar at the top (~150px).
# We need the element at y > 300 to avoid clicking the search input.
#
# Strategy: use PyAutoGUI mouse wheel scroll (real OS event) to position
# the element in the middle of the viewport, then re-read coordinates.
el.scroll_into_view_if_needed(timeout=3000)
time.sleep(0.3)
box = el.bounding_box(timeout=2000)
if box and box["y"] < 300:
# Element is too close to top — behind the fixed search bar.
# Use Playwright mouse.wheel to scroll page UP (negative deltaY)
# so the element moves DOWN in the viewport to a safe y > 350.
delta = int(box["y"]) - 400 # negative value scrolls page up
print(" Scrolling page (delta={}) to clear fixed header (y={})".format(
delta, int(box["y"])), flush=True)
page.mouse.wheel(0, delta)
time.sleep(0.8)
# Re-read bounding box after scroll
box = el.bounding_box(timeout=2000)
if not box:
print(" ⚠ Lost element after scroll", flush=True)
return False
# Click the left portion of the link text (more reliable than center)
rx = int(box["x"] + min(box["width"] * 0.3, 150))
ry = int(box["y"] + box["height"] / 2)
print(" [PyAutoGUI] Clicking at ({},{})".format(rx, ry), flush=True)
pages_before = len(ctx.pages)
human_click(rx, ry)
# Wait for new tab — Baidu links have target=_blank, so clicking should
# open a new tab. Give it enough time for the Baidu redirect.
for _ in range(10):
page.wait_for_timeout(500)
if len(ctx.pages) > pages_before:
break
if len(ctx.pages) > pages_before:
target = ctx.pages[-1]
target.wait_for_timeout(6000)
title = target.title()
url = target.url
print(" [New Tab] {} | {}".format(title[:50], url[:80]), flush=True)
screenshot(target, label)
target.close()
page.wait_for_timeout(1000)
return True
else:
# Check if URL changed (same-tab navigation via Baidu redirect)
current_url = page.url
if "baidu.com/s?" not in current_url:
page.wait_for_timeout(5000)
print(" Landed: {} | {}".format(page.title()[:50], page.url[:80]), flush=True)
screenshot(page, label)
print(" [PyAutoGUI] Going back...", flush=True)
pyautogui.hotkey("alt", "Left")
page.wait_for_timeout(3000)
return True
else:
print(" ⚠ Click didn't navigate — still on search page", flush=True)
print(" Trying Playwright click as fallback...", flush=True)
el.click(timeout=5000)
page.wait_for_timeout(3000)
if len(ctx.pages) > pages_before:
target = ctx.pages[-1]
target.wait_for_timeout(6000)
print(" [New Tab via Playwright] {} | {}".format(
target.title()[:50], target.url[:80]), flush=True)
screenshot(target, label)
target.close()
page.wait_for_timeout(1000)
return True
elif "baidu.com/s?" not in page.url:
page.wait_for_timeout(5000)
print(" [Playwright] Landed: {}".format(page.title()[:50]), flush=True)
screenshot(page, label)
pyautogui.hotkey("alt", "Left")
page.wait_for_timeout(3000)
return True
return False
def screenshot(page, name):
"""Save screenshot to SCREENSHOT_DIR."""
path = os.path.join(SCREENSHOT_DIR, name)
page.screenshot(path=path)
print(" Screenshot: " + path, flush=True)
# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
def main():
print("=" * 60, flush=True)
print(" Baidu Search — PyAutoGUI OS-Level Demo", flush=True)
print("=" * 60, flush=True)
with Stealth().use_sync(sync_playwright()) as p:
browser = p.chromium.launch(
channel="chrome",
headless=False,
args=[
"--no-sandbox",
"--disable-blink-features=AutomationControlled",
"--disable-dev-shm-usage",
"--window-size=1920,1080",
"--window-position=0,0",
],
)
ctx = browser.new_context(
viewport={"width": 1920, "height": 1080},
locale="zh-CN",
timezone_id="Asia/Shanghai",
)
page = ctx.new_page()
# Inject stealth script
stealth_path = "/usr/local/share/yao/stealth-init.js"
if os.path.exists(stealth_path):
page.add_init_script(open(stealth_path).read())
# ---- Step 1: Open Baidu ----
print("\n[1/7] Opening Baidu...", flush=True)
page.goto("https://www.baidu.com", timeout=30000)
page.wait_for_timeout(3000)
print(" Title: " + page.title(), flush=True)
screenshot(page, "baidu-01-homepage.png")
# ---- Step 2: Find & click search box ----
print("\n[2/7] Finding search box...", flush=True)
search_selectors = [
"#kw", "input[name=wd]", "input[name=word]",
"input.s_ipt", "input[type=text]", "input[type=search]",
]
search_el, box = find_element(page, search_selectors)
if not box:
# Fallback: Baidu search box typical position
print(" Using fallback coordinates", flush=True)
box = {"x": 600, "y": 350, "width": 600, "height": 40}
cx = int(box["x"] + box["width"] / 2)
cy = int(box["y"] + box["height"] / 2)
print(" [PyAutoGUI] Clicking search box at ({}, {})".format(cx, cy), flush=True)
human_move(100, 100)
time.sleep(0.3)
human_click(cx, cy)
# ---- Step 3: Type search query ----
print("\n[3/7] Typing search query...", flush=True)
query = "yao app engine"
if search_el:
smart_type(search_el, query)
else:
# No element reference — type blindly with PyAutoGUI
for ch in query:
pyautogui.press(ch)
time.sleep(random.uniform(0.06, 0.12))
time.sleep(1)
screenshot(page, "baidu-02-typed.png")
# ---- Step 4: Submit search ----
print("\n[4/7] Submitting search...", flush=True)
pyautogui.press("enter")
page.wait_for_timeout(5000)
print(" URL: " + page.url[:100], flush=True)
print(" Title: " + page.title(), flush=True)
screenshot(page, "baidu-03-results.png")
# ---- Step 5: Find results ----
print("\n[5/7] Finding search results...", flush=True)
result_selectors = [".c-container h3 a", "h3 a", "a:has(h3)"]
results = find_results(page, result_selectors)
print(" Found {} results".format(len(results)), flush=True)
for i, r in enumerate(results[:5]):
print(" [{}] {}".format(i + 1, r["title"][:60]), flush=True)
if len(results) < 3:
print("\n ⚠ Not enough results. Page may show CAPTCHA.", flush=True)
screenshot(page, "baidu-04-no-results.png")
else:
# Baidu results page has a fixed search bar at the top (~150px).
# The first result (index 0) is often right under it, making
# PyAutoGUI click hit the search input instead of the link.
# So we click results starting from index 1 (second result).
# ---- Step 6: Click result #2 ----
print("\n[6/7] Clicking result #2: " + results[1]["title"][:60], flush=True)
click_result(ctx, page, results[1], "baidu-04-page1.png")
# ---- Step 7: Click result #3 ----
# Re-find results (page may have scrolled, coordinates changed)
page.evaluate("window.scrollTo(0, 0)")
page.wait_for_timeout(1000)
results2 = find_results(page, result_selectors)
print("\n Re-found {} results".format(len(results2)), flush=True)
if len(results2) >= 3:
print("\n[7/7] Clicking result #3: " + results2[2]["title"][:60], flush=True)
click_result(ctx, page, results2[2], "baidu-05-page2.png")
else:
print("\n[7/7] ⚠ Could not re-find results for second click", flush=True)
# Keep browser open for VNC observation
print("\n" + "=" * 60, flush=True)
print(" Demo complete! Browser stays open 30s for observation.", flush=True)
print(" Connect via VNC: http://localhost:6080", flush=True)
print("=" * 60, flush=True)
page.wait_for_timeout(30000)
browser.close()
print("Done.", flush=True)
if __name__ == "__main__":
main()

275
sandbox/docker/chrome/tests/demo-duckduckgo.py Normal file
View file

@ -0,0 +1,275 @@
"""
DuckDuckGo Search Demo — PyAutoGUI OS-level Mouse + Smart Keyboard Fallback
Demonstrates anti-detection browser automation inside sandbox-claude-chrome:
1. Open DuckDuckGo homepage
2. Click search box with PyAutoGUI (real OS mouse event)
3. Type query with PyAutoGUI keyboard (auto-fallback to Playwright if needed)
4. Submit search
5. Click first search result (PyAutoGUI mouse)
6. Go back
7. Click second search result (PyAutoGUI mouse)
DuckDuckGo has no IP-based rate limiting or reCAPTCHA, making it ideal for
demonstrating pure anti-fingerprint capabilities without IP interference.
Prerequisites:
- Running inside sandbox-claude-chrome container
- DISPLAY=:99 (Xvfb virtual display)
- VNC optional for live observation (http://localhost:6080)
Usage:
DISPLAY=:99 python3 demo-duckduckgo.py
"""
from playwright.sync_api import sync_playwright
from playwright_stealth import Stealth
import pyautogui
import time
import random
import os
# ---------------------------------------------------------------------------
# PyAutoGUI config
# ---------------------------------------------------------------------------
pyautogui.FAILSAFE = False
pyautogui.PAUSE = 0.1
# Screenshot output directory
SCREENSHOT_DIR = os.environ.get("SCREENSHOT_DIR", "/workspace")
# ---------------------------------------------------------------------------
# Human-like helpers
# ---------------------------------------------------------------------------
def human_move(x, y):
"""Move mouse with randomized speed to simulate human behavior."""
duration = random.uniform(0.4, 0.8)
pyautogui.moveTo(x, y, duration=duration)
time.sleep(random.uniform(0.1, 0.3))
def human_click(x, y):
"""Move to (x, y) then click — mimics a real user click."""
human_move(x, y)
time.sleep(random.uniform(0.05, 0.15))
pyautogui.click()
time.sleep(random.uniform(0.2, 0.5))
def smart_type(element, text):
"""Type text with PyAutoGUI first; fallback to Playwright if it didn't land.
On native amd64 Linux, PyAutoGUI keyboard works perfectly.
On ARM Mac (Rosetta 2), X11 keyboard events may not reach Chrome,
so we detect and automatically fallback to Playwright's type().
"""
# Attempt PyAutoGUI keyboard (OS-level X11 events)
for ch in text:
pyautogui.press(ch)
time.sleep(random.uniform(0.05, 0.12))
time.sleep(0.5)
# Verify input landed
actual = element.input_value()
if actual and len(actual) >= len(text) * 0.8:
print(" Keyboard: PyAutoGUI (OS-level) ✓", flush=True)
return
# Fallback: Playwright type()
print(" PyAutoGUI keyboard didn't land — fallback to Playwright", flush=True)
element.fill("")
element.type(text, delay=80)
actual = element.input_value()
print(" Keyboard: Playwright fallback — '{}'".format(actual), flush=True)
def find_element(page, selectors, min_width=50, timeout=2000):
"""Try multiple CSS selectors, return (element, bounding_box) or (None, None)."""
for selector in selectors:
try:
el = page.locator(selector).first
box = el.bounding_box(timeout=timeout)
if box and box["width"] >= min_width:
return el, box
except Exception:
continue
return None, None
def find_results(page, selectors, min_count=2, min_width=80):
"""Find search result elements with bounding boxes."""
results = []
for selector in selectors:
elements = page.locator(selector).all()
for el in elements[:10]:
try:
box = el.bounding_box(timeout=1000)
title = el.text_content().strip()
if (box and box["width"] >= min_width and box["y"] > 0
and title and len(title) > 5):
if not any(r["title"] == title for r in results):
results.append({"box": box, "title": title})
except Exception:
pass
if len(results) >= min_count:
break
return results
def screenshot(page, name):
"""Save screenshot to SCREENSHOT_DIR."""
path = os.path.join(SCREENSHOT_DIR, name)
page.screenshot(path=path)
print(" Screenshot: " + path, flush=True)
# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
def main():
print("=" * 60, flush=True)
print(" DuckDuckGo Search — PyAutoGUI OS-Level Demo", flush=True)
print("=" * 60, flush=True)
with Stealth().use_sync(sync_playwright()) as p:
browser = p.chromium.launch(
channel="chrome",
headless=False,
args=[
"--no-sandbox",
"--disable-blink-features=AutomationControlled",
"--disable-dev-shm-usage",
"--window-size=1920,1080",
"--window-position=0,0",
],
)
ctx = browser.new_context(
viewport={"width": 1920, "height": 1080},
locale="en-US",
timezone_id="America/New_York",
)
page = ctx.new_page()
# Inject stealth script
stealth_path = "/usr/local/share/yao/stealth-init.js"
if os.path.exists(stealth_path):
page.add_init_script(open(stealth_path).read())
# ---- Step 1: Open DuckDuckGo ----
print("\n[1/7] Opening DuckDuckGo...", flush=True)
page.goto("https://duckduckgo.com", timeout=30000)
page.wait_for_timeout(3000)
print(" Title: " + page.title(), flush=True)
screenshot(page, "ddg-01-homepage.png")
# ---- Step 2: Find & click search box ----
print("\n[2/7] Finding search box...", flush=True)
search_selectors = [
"input[name=q]", "#searchbox_input",
"input[type=text]", "input[placeholder*='Search']",
]
search_el, box = find_element(page, search_selectors, min_width=100)
if not box:
print(" ⚠ Could not find search box!", flush=True)
screenshot(page, "ddg-02-no-searchbox.png")
browser.close()
return
cx = int(box["x"] + box["width"] / 2)
cy = int(box["y"] + box["height"] / 2)
print(" [PyAutoGUI] Clicking search box at ({}, {})".format(cx, cy), flush=True)
human_move(200, 200)
time.sleep(0.3)
human_click(cx, cy)
# ---- Step 3: Type search query ----
print("\n[3/7] Typing search query...", flush=True)
query = "yao app engine github"
smart_type(search_el, query)
time.sleep(0.5)
screenshot(page, "ddg-02-typed.png")
# ---- Step 4: Submit search ----
print("\n[4/7] Submitting search...", flush=True)
# Use Playwright Enter on the element (reliable cross-platform)
search_el.press("Enter")
page.wait_for_timeout(6000)
print(" URL: " + page.url[:120], flush=True)
print(" Title: " + page.title()[:80], flush=True)
screenshot(page, "ddg-03-results.png")
# Check if we actually reached results page
on_results = (
"q=" in page.url
or "/search" in page.url
or page.title() != "DuckDuckGo - Protection. Privacy. Peace of mind."
)
if not on_results:
print(" ⚠ Still on homepage — search may not have submitted", flush=True)
screenshot(page, "ddg-03-still-homepage.png")
browser.close()
return
# ---- Step 5: Find results ----
print("\n[5/7] Finding search results...", flush=True)
result_selectors = [
"article h2 a", "a[data-testid='result-title-a']",
"h2 a[href]", "ol li h2 a", "h2 a",
]
results = find_results(page, result_selectors, min_count=3)
print(" Found {} results".format(len(results)), flush=True)
for i, r in enumerate(results[:5]):
print(" [{}] {}".format(i + 1, r["title"][:70]), flush=True)
if len(results) < 2:
print("\n ⚠ Not enough results to click.", flush=True)
screenshot(page, "ddg-04-no-results.png")
else:
# ---- Step 6: Click first result ----
r1 = results[0]
rx = int(r1["box"]["x"] + r1["box"]["width"] / 2)
ry = int(r1["box"]["y"] + r1["box"]["height"] / 2)
print("\n[6/7] [PyAutoGUI] Clicking result #1 at ({},{})".format(rx, ry), flush=True)
print(" " + r1["title"][:70], flush=True)
pyautogui.scroll(-1)
time.sleep(0.3)
human_click(rx, ry)
page.wait_for_timeout(6000)
print(" Landed: {} | {}".format(page.title()[:50], page.url[:80]), flush=True)
screenshot(page, "ddg-04-page1.png")
# Go back
print(" [PyAutoGUI] Going back (Alt+Left)...", flush=True)
pyautogui.hotkey("alt", "Left")
page.wait_for_timeout(4000)
# ---- Step 7: Click second result ----
results2 = find_results(page, result_selectors, min_count=3)
if len(results2) >= 2:
r2 = results2[1]
rx2 = int(r2["box"]["x"] + r2["box"]["width"] / 2)
ry2 = int(r2["box"]["y"] + r2["box"]["height"] / 2)
print("\n[7/7] [PyAutoGUI] Clicking result #2 at ({},{})".format(rx2, ry2), flush=True)
print(" " + r2["title"][:70], flush=True)
human_click(rx2, ry2)
page.wait_for_timeout(6000)
print(" Landed: {} | {}".format(page.title()[:50], page.url[:80]), flush=True)
screenshot(page, "ddg-05-page2.png")
else:
print("\n[7/7] ⚠ Could not re-find results for second click", flush=True)
# Keep browser open for VNC observation
print("\n" + "=" * 60, flush=True)
print(" Demo complete! Browser stays open 30s for observation.", flush=True)
print(" Connect via VNC: http://localhost:6080", flush=True)
print("=" * 60, flush=True)
page.wait_for_timeout(30000)
browser.close()
print("Done.", flush=True)
if __name__ == "__main__":
main()

860
sandbox/docker/chrome/tests/demo-llm-vision.py Normal file
View file

@ -0,0 +1,860 @@
"""
LLM + DOM + CDP Browser Automation Demo
Architecture (each layer does what it's best at):
- Playwright: Opens pages, extracts HTML, keyboard input
- LLM: Reads HTML structure, returns CSS selectors for target elements
- DOM: Uses LLM's selectors to get precise bounding boxes
- CDP: Chrome DevTools Protocol mouse events (isTrusted=true, anti-detection)
- PyAutoGUI: OS-level mouse fallback (when CDP fails)
Click strategy (layered, most reliable first):
1. CDP Input.dispatchMouseEvent — coordinates match bounding_box() exactly,
generates isTrusted=true events, nearly indistinguishable from real user.
2. PyAutoGUI OS-level mouse — true hardware events, but coordinates may
drift due to window chrome offset.
3. Playwright .click() — last resort, may be detected by anti-bot.
No hardcoded selectors — LLM figures out the page structure dynamically.
Works with any cheap text LLM (no vision needed).
Environment variables (required):
LLM_API_KEY — API key
LLM_API_BASE — OpenAI-compatible endpoint URL
LLM_MODEL — Model name/ID
Usage:
export LLM_API_KEY="your-key"
export LLM_API_BASE="https://api.openai.com/v1/"
export LLM_MODEL="gpt-4o-mini"
DISPLAY=:99 python3 demo-llm-vision.py <search_url> <search_query>
Example:
DISPLAY=:99 python3 demo-llm-vision.py https://www.google.com "Yao App Engine"
DISPLAY=:99 python3 demo-llm-vision.py https://www.baidu.com "Yao App Engine"
"""
from playwright.sync_api import sync_playwright
from playwright_stealth import Stealth
import pyautogui
import time
import random
import json
import os
import re
import urllib.request
import urllib.error
import sys
from concurrent.futures import ThreadPoolExecutor, as_completed
# ---------------------------------------------------------------------------
# Config
# ---------------------------------------------------------------------------
pyautogui.FAILSAFE = False
pyautogui.PAUSE = 0.1
SCREENSHOT_DIR = os.environ.get("SCREENSHOT_DIR", "/workspace")
LLM_API_KEY = os.environ.get("LLM_API_KEY", "").strip().strip('"\'')
LLM_API_BASE = os.environ.get("LLM_API_BASE", "").strip().strip('"\'')
LLM_MODEL = os.environ.get("LLM_MODEL", "").strip().strip('"\'')
if LLM_API_BASE and not LLM_API_BASE.endswith("/"):
LLM_API_BASE += "/"
# Command-line arguments: <search_url> <search_query>
SEARCH_URL = sys.argv[1] if len(sys.argv) > 1 else "https://www.google.com"
SEARCH_QUERY = sys.argv[2] if len(sys.argv) > 2 else "Yao App Engine"
# ---------------------------------------------------------------------------
# LLM API
# ---------------------------------------------------------------------------
def ask_llm(prompt, timeout=60):
"""Send text prompt to LLM, return text response."""
url = LLM_API_BASE + "chat/completions"
payload = {
"model": LLM_MODEL,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 2048,
"temperature": 0.1,
}
headers = {"Content-Type": "application/json", "Authorization": "Bearer " + LLM_API_KEY}
data = json.dumps(payload).encode("utf-8")
req = urllib.request.Request(url, data=data, headers=headers, method="POST")
try:
with urllib.request.urlopen(req, timeout=timeout) as resp:
result = json.loads(resp.read().decode("utf-8"))
return result["choices"][0]["message"]["content"]
except urllib.error.HTTPError as e:
body = e.read().decode("utf-8")[:500] if e.fp else ""
print(" LLM HTTP {}: {}".format(e.code, body), flush=True)
return None
except Exception as e:
print(" LLM error: {}".format(e), flush=True)
return None
def parse_llm_json(text):
"""Extract JSON from LLM response (handles markdown fences, thinking tags)."""
if not text:
return None
cleaned = re.sub(r'<think>[\s\S]*?</think>', '', text).strip()
cleaned = re.sub(r'^```\w*\n?', '', cleaned)
cleaned = re.sub(r'\n?```$', '', cleaned)
cleaned = cleaned.strip()
try:
return json.loads(cleaned)
except json.JSONDecodeError:
for m in re.finditer(r'(\{[\s\S]*?\}|\[[\s\S]*?\])', cleaned):
try:
return json.loads(m.group())
except json.JSONDecodeError:
continue
return None
def ask_llm_race(prompts, timeout=60, validator=None):
"""Send multiple prompts to LLM concurrently, return first valid result.
Each prompt is sent in a separate thread. As soon as one returns a valid
JSON result (passing optional validator), return immediately without
waiting for the remaining threads.
Args:
prompts: list of (label, prompt_text) tuples
timeout: per-request timeout in seconds
validator: optional fn(parsed_json) -> bool, extra check on result
Returns:
(label, parsed_json) of the first valid result, or (None, None)
"""
if not prompts:
return None, None
# Single prompt — no need for concurrency
if len(prompts) == 1:
label, prompt_text = prompts[0]
resp = ask_llm(prompt_text, timeout=timeout)
parsed = parse_llm_json(resp)
if parsed and (validator is None or validator(parsed)):
return label, parsed
return None, None
print(" [Race] Sending {} concurrent LLM requests...".format(len(prompts)), flush=True)
# Don't use `with` — it calls shutdown(wait=True) which blocks until ALL
# threads finish, even after we found a winner. Instead, manage manually
# and call shutdown(wait=False) to return immediately.
pool = ThreadPoolExecutor(max_workers=len(prompts))
future_map = {}
for label, prompt_text in prompts:
f = pool.submit(ask_llm, prompt_text, timeout)
future_map[f] = label
try:
for f in as_completed(future_map):
label = future_map[f]
try:
resp = f.result()
parsed = parse_llm_json(resp)
if parsed and (validator is None or validator(parsed)):
print(" [Race] Winner: '{}' → {}".format(
label, json.dumps(parsed, ensure_ascii=False)[:120]), flush=True)
# Cancel pending futures (won't stop running ones, but prevents queued)
for other in future_map:
if other is not f:
other.cancel()
return label, parsed
else:
print(" [Race] '{}' returned invalid result, waiting...".format(label), flush=True)
except Exception as e:
print(" [Race] '{}' failed: {}".format(label, e), flush=True)
finally:
# shutdown(wait=False) — let daemon threads die on their own,
# don't block the main thread waiting for slow LLM responses.
pool.shutdown(wait=False, cancel_futures=True)
print(" [Race] All requests failed", flush=True)
return None, None
# ---------------------------------------------------------------------------
# DOM extraction — compact page summary for LLM
# ---------------------------------------------------------------------------
def extract_page_dom(page):
"""Extract a compact text summary of all interactive elements on the page.
Includes inputs, buttons, and visible links — generic, no hardcoded selectors.
LLM reads this to decide what to interact with."""
return page.evaluate("""() => {
const lines = [];
lines.push('URL: ' + location.href);
lines.push('Title: ' + document.title);
lines.push('');
// Helper: describe element visibility
function vis(el) {
const r = el.getBoundingClientRect();
return (r.width > 5 && r.height > 5)
? '[VISIBLE ' + Math.round(r.width) + 'x' + Math.round(r.height) + ']'
: '[HIDDEN]';
}
// Helper: build a minimal CSS selector for an element
function sel(el) {
const tag = el.tagName.toLowerCase();
if (el.id) return tag + '#' + el.id;
if (el.className) {
const cls = el.className.toString().trim().split(/\\s+/).slice(0, 2).join('.');
if (cls) return tag + '.' + cls;
}
return tag;
}
// 1. Inputs, textareas, buttons
lines.push('=== Inputs & Buttons ===');
document.querySelectorAll('input, textarea, button, [role="textbox"], [contenteditable="true"]').forEach(el => {
const tag = el.tagName.toLowerCase();
const parts = [sel(el)];
if (el.type && el.type !== 'text') parts.push('type="' + el.type + '"');
if (el.name) parts.push('name="' + el.name + '"');
if (el.placeholder) parts.push('placeholder="' + el.placeholder.substring(0, 40) + '"');
if (el.value) parts.push('value="' + el.value.substring(0, 30) + '"');
const text = (el.innerText || '').trim().substring(0, 30);
if (text && tag === 'button') parts.push('text="' + text + '"');
if (el.getAttribute('aria-label')) parts.push('aria-label="' + el.getAttribute('aria-label').substring(0, 30) + '"');
parts.push(vis(el));
lines.push(' ' + parts.join(' '));
});
// 2. Visible links
lines.push('');
lines.push('=== Links ===');
const seen = new Set();
let count = 0;
document.querySelectorAll('a[href]').forEach(a => {
if (count >= 30) return;
const href = a.href || '';
if (!href || href.startsWith('javascript:')) return;
const rect = a.getBoundingClientRect();
if (rect.width === 0 || rect.height === 0) return;
if (rect.top < 30) return;
const text = (a.innerText || '').trim().replace(/\\s+/g, ' ').substring(0, 80);
if (!text || text.length < 2 || seen.has(text)) return;
seen.add(text);
const parent = a.parentElement;
let ctx = parent ? sel(parent) + ' > ' : '';
const heading = a.querySelector('h1,h2,h3,h4');
const htag = heading ? ' [has <' + heading.tagName.toLowerCase() + '>]' : '';
lines.push(' ' + ctx + sel(a) + htag + ' ' + vis(a) + ' → "' + text + '"');
count++;
});
return lines.join('\\n');
}""")
# ---------------------------------------------------------------------------
# CDP click — Chrome DevTools Protocol mouse events (isTrusted=true)
# ---------------------------------------------------------------------------
_cdp_session = None
_cdp_page_id = None
def get_cdp_session(page):
"""Get or create a CDP session for the given page.
Recreates session if page changed (e.g. after tab close/navigation)."""
global _cdp_session, _cdp_page_id
page_id = id(page)
if _cdp_session is None or _cdp_page_id != page_id:
try:
if _cdp_session:
_cdp_session.detach()
except Exception:
pass
_cdp_session = page.context.new_cdp_session(page)
_cdp_page_id = page_id
return _cdp_session
def cdp_click(page, x, y, ctrl=False):
"""Click at (x, y) via CDP Input.dispatchMouseEvent.
Coordinates are in viewport space (same as bounding_box()).
Generates isTrusted=true events — nearly indistinguishable from real user.
When ctrl=True, holds Ctrl modifier to force new tab (like Ctrl+Click)."""
cdp = get_cdp_session(page)
# Simulate human-like: small random offset (±2px)
x += random.uniform(-2, 2)
y += random.uniform(-2, 2)
modifiers = 2 if ctrl else 0 # 2 = Ctrl modifier in CDP
# mouseMoved — simulate cursor arriving
cdp.send("Input.dispatchMouseEvent", {
"type": "mouseMoved",
"x": x, "y": y,
"button": "none",
"modifiers": modifiers,
"pointerType": "mouse",
})
time.sleep(random.uniform(0.05, 0.15))
# mousePressed
cdp.send("Input.dispatchMouseEvent", {
"type": "mousePressed",
"x": x, "y": y,
"button": "left",
"clickCount": 1,
"modifiers": modifiers,
"pointerType": "mouse",
})
time.sleep(random.uniform(0.03, 0.08))
# mouseReleased
cdp.send("Input.dispatchMouseEvent", {
"type": "mouseReleased",
"x": x, "y": y,
"button": "left",
"clickCount": 1,
"modifiers": modifiers,
"pointerType": "mouse",
})
time.sleep(random.uniform(0.1, 0.3))
def cdp_move(page, x, y, steps=10):
"""Simulate human-like mouse movement via CDP (curved path)."""
cdp = get_cdp_session(page)
# Start from a random nearby position
sx = x + random.uniform(-200, 200)
sy = y + random.uniform(-100, 100)
for i in range(steps + 1):
t = i / steps
# Ease-in-out curve
t = t * t * (3 - 2 * t)
mx = sx + (x - sx) * t + random.uniform(-1, 1)
my = sy + (y - sy) * t + random.uniform(-1, 1)
cdp.send("Input.dispatchMouseEvent", {
"type": "mouseMoved",
"x": mx, "y": my,
"button": "none",
"pointerType": "mouse",
})
time.sleep(random.uniform(0.01, 0.03))
def smart_click(page, x, y, label=""):
"""Click using CDP (primary) with PyAutoGUI fallback.
Returns the method used: 'cdp', 'pyautogui', or None on failure."""
tag = "[CDP]" if label else "[CDP]"
try:
cdp_move(page, x, y)
cdp_click(page, x, y)
print(" {} Click at ({},{}){}".format(
tag, int(x), int(y),
" '{}'".format(label[:40]) if label else ""), flush=True)
return "cdp"
except Exception as e:
print(" {} Failed: {} → PyAutoGUI fallback".format(tag, e), flush=True)
try:
pyautogui.moveTo(x, y, duration=random.uniform(0.4, 0.8))
time.sleep(random.uniform(0.05, 0.15))
pyautogui.click()
time.sleep(random.uniform(0.2, 0.5))
print(" [PyAutoGUI] Click at ({},{})".format(int(x), int(y)), flush=True)
return "pyautogui"
except Exception as e2:
print(" [PyAutoGUI] Also failed: {}".format(e2), flush=True)
return None
# ---------------------------------------------------------------------------
# Interaction helpers
# ---------------------------------------------------------------------------
def box_center(box):
return int(box["x"] + box["width"] / 2), int(box["y"] + box["height"] / 2)
def take_screenshot(page, name):
path = os.path.join(SCREENSHOT_DIR, name)
page.screenshot(path=path)
print(" Screenshot: " + path, flush=True)
def locate_element(page, selector):
"""Use a CSS selector to find element, return (element, bounding_box) or (None, None)."""
try:
loc = page.locator(selector)
count = loc.count()
print(" [locate] '{}' matched {} elements".format(selector, count), flush=True)
if count == 0:
return None, None
el = loc.first
# Try to make it visible first
try:
el.scroll_into_view_if_needed(timeout=2000)
except Exception:
pass
box = el.bounding_box(timeout=5000)
if box:
print(" [locate] box: x={} y={} w={} h={}".format(
int(box["x"]), int(box["y"]), int(box["width"]), int(box["height"])), flush=True)
if box["width"] > 5:
return el, box
else:
print(" [locate] bounding_box returned None (element hidden?)", flush=True)
except Exception as e:
print(" [locate] error: {}".format(e), flush=True)
return None, None
def locate_elements(page, selector, min_y=0):
"""Find all visible elements matching selector, return list of (element, box, text)."""
results = []
try:
els = page.locator(selector).all()
for el in els:
try:
box = el.bounding_box(timeout=500)
if box and box["width"] > 30 and box["y"] > min_y:
text = el.inner_text(timeout=500)[:80]
results.append((el, box, text))
except Exception:
continue
except Exception:
pass
return results
def click_new_tab(ctx, page, el, box, label):
"""Ctrl+Click an element via CDP to force open in new tab.
Keeps the search results page intact. Waits for new tab, screenshots, closes it.
Fallback chain: CDP Ctrl+Click → Playwright Ctrl+Click → JS window.open"""
cx, cy = box_center(box)
# Scroll into view if needed
if cy < 100 or cy > 1000:
try:
el.scroll_into_view_if_needed(timeout=2000)
time.sleep(0.5)
box = el.bounding_box(timeout=1000)
if box:
cx, cy = box_center(box)
except Exception:
pass
pages_before = len(ctx.pages)
# --- Attempt 1: CDP Ctrl+Click (isTrusted=true, new tab) ---
try:
cdp_move(page, cx, cy)
cdp_click(page, cx, cy, ctrl=True)
print(" [CDP Ctrl+Click] '{}' at ({},{})".format(label[:40], int(cx), int(cy)), flush=True)
except Exception as e:
print(" [CDP Ctrl+Click] Failed: {}".format(e), flush=True)
# Wait for new tab
for _ in range(12):
page.wait_for_timeout(500)
if len(ctx.pages) > pages_before:
break
if len(ctx.pages) > pages_before:
target = ctx.pages[-1]
target.wait_for_timeout(6000)
print(" ✓ [New Tab] {} | {}".format(target.title()[:50], target.url[:80]), flush=True)
take_screenshot(target, "llm-detail.png")
target.close()
page.wait_for_timeout(500)
# Bring focus back to search results page
page.bring_to_front()
return True
# --- Attempt 2: Playwright modifier click ---
print(" CDP Ctrl+Click no new tab → Playwright modifier click", flush=True)
try:
el.click(modifiers=["Control"], timeout=3000)
page.wait_for_timeout(3000)
if len(ctx.pages) > pages_before:
target = ctx.pages[-1]
target.wait_for_timeout(6000)
print(" ✓ [New Tab] {} | {}".format(target.title()[:50], target.url[:80]), flush=True)
take_screenshot(target, "llm-detail.png")
target.close()
page.wait_for_timeout(500)
page.bring_to_front()
return True
except Exception:
pass
# --- Attempt 3: JS window.open with href ---
print(" Modifier click failed → JS window.open fallback", flush=True)
try:
href = el.get_attribute("href", timeout=2000)
if href:
new_page = ctx.new_page()
new_page.goto(href, timeout=15000)
new_page.wait_for_timeout(5000)
print(" ✓ [JS Tab] {} | {}".format(new_page.title()[:50], new_page.url[:80]), flush=True)
take_screenshot(new_page, "llm-detail.png")
new_page.close()
page.wait_for_timeout(500)
page.bring_to_front()
return True
except Exception as e:
print(" [JS Tab] Failed: {}".format(e), flush=True)
print(" ✗ All methods failed to open new tab", flush=True)
return False
# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
def main():
print("=" * 60, flush=True)
print(" LLM + DOM + CDP Browser Automation", flush=True)
print("=" * 60, flush=True)
print(" URL: " + SEARCH_URL, flush=True)
print(" Query: " + SEARCH_QUERY, flush=True)
print(" Model: " + LLM_MODEL, flush=True)
print(" Endpoint: " + LLM_API_BASE[:60], flush=True)
print(" Key: " + (LLM_API_KEY[:8] + "..." if LLM_API_KEY else "NOT SET"), flush=True)
print("", flush=True)
print(" Flow: LLM reads HTML → returns CSS selectors →", flush=True)
print(" DOM locates elements → CDP clicks (isTrusted)", flush=True)
if not all([LLM_API_KEY, LLM_API_BASE, LLM_MODEL]):
print("\n⚠ Missing LLM config! Set: LLM_API_KEY, LLM_API_BASE, LLM_MODEL", flush=True)
return
with Stealth().use_sync(sync_playwright()) as p:
browser = p.chromium.launch(
channel="chrome", headless=False,
args=["--no-sandbox", "--disable-blink-features=AutomationControlled",
"--disable-dev-shm-usage", "--window-size=1920,1080", "--window-position=0,0"])
ctx = browser.new_context(
viewport={"width": 1920, "height": 1080},
locale="zh-CN", timezone_id="Asia/Shanghai")
page = ctx.new_page()
stealth_path = "/usr/local/share/yao/stealth-init.js"
if os.path.exists(stealth_path):
page.add_init_script(open(stealth_path).read())
# ============================================================
# Phase 1: Open search engine
# ============================================================
print("\n[Phase 1] Opening {}...".format(SEARCH_URL), flush=True)
page.goto(SEARCH_URL, timeout=30000)
page.wait_for_timeout(3000)
print(" Title: " + page.title(), flush=True)
take_screenshot(page, "llm-01-homepage.png")
# ============================================================
# Phase 2: LLM analyzes homepage HTML → gives selectors
# ============================================================
print("\n[Phase 2] [LLM] Analyzing homepage structure...", flush=True)
elements = extract_page_dom(page)
print(elements[:500], flush=True)
if len(elements) > 500:
print(" ... ({} chars total)".format(len(elements)), flush=True)
hp_prompt_tpl = """Below is the DOM structure of a search engine homepage.
Each element is marked [VISIBLE WxH] or [HIDDEN].
I want to:
1. Type a search query into the search input box
2. Click the search submit button
IMPORTANT: Only pick elements marked [VISIBLE]. Ignore [HIDDEN] elements.
Give me CSS selectors for both elements.
{}
Reply ONLY with JSON (no other text):
{{"input_selector": "<CSS selector for a VISIBLE input>", "button_selector": "<CSS selector for a VISIBLE button>"}}"""
# Split homepage DOM into chunks for concurrent LLM calls
hp_lines = elements.split('\n')
hp_header = []
hp_body = []
for line in hp_lines:
if line.startswith("URL:") or line.startswith("Title:") or line == "":
hp_header.append(line)
else:
hp_body.append(line)
hp_hdr = '\n'.join(hp_header[:3])
hp_chunks = []
cur_chunk = []
cur_len = 0
for line in hp_body:
cur_chunk.append(line)
cur_len += len(line) + 1
if cur_len >= 2000:
hp_chunks.append('\n'.join(cur_chunk))
cur_chunk = []
cur_len = 0
if cur_chunk:
hp_chunks.append('\n'.join(cur_chunk))
hp_prompts = []
if len(hp_chunks) > 1:
for i, chunk in enumerate(hp_chunks):
hp_prompts.append(("chunk-{}".format(i + 1), hp_prompt_tpl.format(hp_hdr + '\n' + chunk)))
hp_prompts.append(("full", hp_prompt_tpl.format(elements)))
print(" [Phase 2] {} concurrent LLM requests".format(len(hp_prompts)), flush=True)
def _valid_homepage(parsed):
return (isinstance(parsed, dict)
and bool(parsed.get("input_selector", "").strip())
and bool(parsed.get("button_selector", "").strip()))
label, selectors = ask_llm_race(hp_prompts, timeout=180, validator=_valid_homepage)
if not selectors or not isinstance(selectors, dict):
print(" ✗ LLM failed to return selectors", flush=True)
browser.close()
return
input_sel = selectors.get("input_selector", "")
button_sel = selectors.get("button_selector", "")
print(" → Input: {}".format(input_sel), flush=True)
print(" → Button: {}".format(button_sel), flush=True)
# ============================================================
# Phase 3: DOM locates elements → CDP clicks + Playwright types
# ============================================================
print("\n[Phase 3] [DOM] Locating input: '{}'".format(input_sel), flush=True)
input_el, input_box = locate_element(page, input_sel)
if not input_el:
print(" ✗ Selector '{}' didn't match! Aborting.".format(input_sel), flush=True)
browser.close()
return
ix, iy = box_center(input_box)
print(" ✓ Input at ({},{}) size={}x{}".format(
ix, iy, int(input_box["width"]), int(input_box["height"])), flush=True)
smart_click(page, ix, iy, "search input")
time.sleep(0.3)
print(" [Playwright] Type '{}'".format(SEARCH_QUERY), flush=True)
input_el.type(SEARCH_QUERY, delay=80)
time.sleep(0.5)
take_screenshot(page, "llm-02-typed.png")
# ============================================================
# Phase 4: DOM locates button → CDP clicks
# ============================================================
print("\n[Phase 4] [DOM] Locating button: '{}'".format(button_sel), flush=True)
btn_el, btn_box = locate_element(page, button_sel)
if btn_el and btn_box:
bx, by = box_center(btn_box)
print(" ✓ Button at ({},{})".format(bx, by), flush=True)
smart_click(page, bx, by, "search button")
else:
print(" Button not found → Enter key via CDP", flush=True)
smart_click(page, ix, iy, "input focus")
time.sleep(0.2)
try:
cdp = get_cdp_session(page)
cdp.send("Input.dispatchKeyEvent", {
"type": "keyDown", "key": "Enter", "code": "Enter",
"windowsVirtualKeyCode": 13, "nativeVirtualKeyCode": 13,
})
cdp.send("Input.dispatchKeyEvent", {
"type": "keyUp", "key": "Enter", "code": "Enter",
"windowsVirtualKeyCode": 13, "nativeVirtualKeyCode": 13,
})
except Exception:
pyautogui.press("enter")
page.wait_for_timeout(5000)
results_url = page.url
print(" URL: " + results_url[:100], flush=True)
print(" Title: " + page.title()[:60], flush=True)
take_screenshot(page, "llm-03-results.png")
# If URL unchanged, fallback: Enter key → Playwright click → form submit
homepage = SEARCH_URL.rstrip("/")
if results_url.rstrip("/") == homepage:
print(" URL unchanged — trying CDP Enter key fallback...", flush=True)
smart_click(page, ix, iy, "input refocus")
time.sleep(0.2)
try:
cdp = get_cdp_session(page)
cdp.send("Input.dispatchKeyEvent", {
"type": "keyDown", "key": "Enter", "code": "Enter",
"windowsVirtualKeyCode": 13, "nativeVirtualKeyCode": 13,
})
cdp.send("Input.dispatchKeyEvent", {
"type": "keyUp", "key": "Enter", "code": "Enter",
"windowsVirtualKeyCode": 13, "nativeVirtualKeyCode": 13,
})
except Exception:
pyautogui.press("enter")
page.wait_for_timeout(5000)
results_url = page.url
if results_url.rstrip("/") == homepage:
print(" Still unchanged — trying Playwright click fallback...", flush=True)
try:
if btn_el:
btn_el.click(timeout=3000)
else:
input_el.press("Enter")
page.wait_for_timeout(5000)
results_url = page.url
except Exception:
pass
if results_url.rstrip("/") == homepage:
print(" Still unchanged — trying form submit fallback...", flush=True)
try:
page.evaluate("document.querySelector('form')?.submit()")
page.wait_for_timeout(5000)
results_url = page.url
except Exception:
pass
print(" Final URL: " + results_url[:100], flush=True)
take_screenshot(page, "llm-03-results.png")
if results_url.rstrip("/") == homepage:
print(" ✗ All submit methods failed", flush=True)
browser.close()
return
# ============================================================
# Phase 5: LLM analyzes results page → gives link selector
# Split DOM into chunks, race concurrent LLM calls
# ============================================================
print("\n[Phase 5] [LLM] Analyzing search results page...", flush=True)
results_dom = extract_page_dom(page)
print(results_dom[:500], flush=True)
if len(results_dom) > 500:
print(" ... ({} chars total)".format(len(results_dom)), flush=True)
# Split DOM into chunks for concurrent LLM calls
dom_lines = results_dom.split('\n')
link_prompt_tpl = """Below is part of the DOM from a search results page.
Each line shows: parent > link_selector [has <h3> if any] → "link text"
I need a CSS selector that matches the organic search result title links.
NOT ads, NOT navigation, NOT pagination — only the main result links.
{}
Reply ONLY JSON: {{"link_selector": "<CSS selector>"}}"""
# Build chunks: split at ~2000 char boundaries, always include URL/Title header
header_lines = []
body_lines = []
for line in dom_lines:
if line.startswith("URL:") or line.startswith("Title:") or line == "":
header_lines.append(line)
else:
body_lines.append(line)
header = '\n'.join(header_lines[:3]) # URL + Title + blank
chunks = []
current_chunk = []
current_len = 0
chunk_limit = 2000
for line in body_lines:
current_chunk.append(line)
current_len += len(line) + 1
if current_len >= chunk_limit:
chunks.append('\n'.join(current_chunk))
current_chunk = []
current_len = 0
if current_chunk:
chunks.append('\n'.join(current_chunk))
# Also send the full DOM as one prompt (in case chunks miss context)
prompts = []
if len(chunks) > 1:
for i, chunk in enumerate(chunks):
chunk_dom = header + '\n' + chunk
prompts.append(("chunk-{}".format(i + 1), link_prompt_tpl.format(chunk_dom)))
# Always include the full DOM as the last prompt
prompts.append(("full", link_prompt_tpl.format(results_dom)))
print(" [Phase 5] {} concurrent LLM requests ({} chunks + full)".format(
len(prompts), len(chunks) if len(chunks) > 1 else 0), flush=True)
def _valid_link_selector(parsed):
return isinstance(parsed, dict) and bool(parsed.get("link_selector", "").strip())
label, link_info = ask_llm_race(prompts, timeout=180, validator=_valid_link_selector)
link_sel = ""
link_results = []
if link_info and isinstance(link_info, dict):
link_sel = link_info.get("link_selector", "")
print(" → Selector: '{}' (from {})".format(link_sel, label), flush=True)
if link_sel:
link_results = locate_elements(page, link_sel, min_y=100)
print(" Found {} clickable links".format(len(link_results)), flush=True)
for i, (el, box, text) in enumerate(link_results[:5]):
cx, cy = box_center(box)
print(" [{}] ({},{}) '{}'".format(i, cx, cy, text[:50]), flush=True)
if len(link_results) < 1:
print(" ✗ No links found", flush=True)
take_screenshot(page, "llm-04-no-links.png")
browser.close()
return
# ============================================================
# Phase 7+: Ctrl+Click results (open in new tab, keep list intact)
# ============================================================
max_clicks = min(len(link_results), 3)
for idx in range(max_clicks):
el_r, box_r, text_r = link_results[idx]
print("\n[Phase {}] Ctrl+Click result #{}: '{}'".format(
7 + idx, idx + 1, text_r[:50]), flush=True)
click_new_tab(ctx, page, el_r, box_r, text_r)
page.wait_for_timeout(1000)
# Done
print("\n" + "=" * 60, flush=True)
print(" ✓ Demo complete!", flush=True)
print(" VNC: http://localhost:6080", flush=True)
print("=" * 60, flush=True)
page.wait_for_timeout(30000)
browser.close()
print("Done.", flush=True)
if __name__ == "__main__":
if len(sys.argv) > 1 and sys.argv[1] in ("-h", "--help"):
print("Usage: python3 demo-llm-vision.py <search_url> <search_query>")
print("")
print("Arguments:")
print(" search_url Search engine URL (default: https://www.google.com)")
print(" search_query What to search for (default: Yao App Engine)")
print("")
print("Environment variables (required):")
print(" LLM_API_KEY API key for the LLM service")
print(" LLM_API_BASE OpenAI-compatible endpoint URL")
print(" LLM_MODEL Model name/ID")
print("")
print("Examples:")
print(' python3 demo-llm-vision.py https://www.google.com "Yao App Engine"')
print(' python3 demo-llm-vision.py https://www.bing.com "Yao App Engine"')
print(' python3 demo-llm-vision.py https://duckduckgo.com "Yao App Engine"')
print(' python3 demo-llm-vision.py https://www.baidu.com "Yao App Engine"')
print(' python3 demo-llm-vision.py https://www.sogou.com "Yao App Engine"')
sys.exit(0)
main()