Add Chrome support to sandbox module and update documentation
- Introduce a new Chrome image for the sandbox, enabling VNC access with real Chrome and CDP support (amd64 only). - Update README to reflect the addition of the Chrome image and its build instructions. - Modify build script to include a dedicated build process for the Chrome image. - Enhance .gitignore to exclude the new Chrome plan file. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Max
parent
28363b973e
commit
95cdf7961c
10 changed files with 1878 additions and 8 deletions
1
.gitignore
vendored
1
.gitignore
vendored
|
|
@ -64,3 +64,4 @@ sandbox/docker/claude/claude-proxy-*
|
|||
sandbox/proxy/claude-proxy-linux-*
|
||||
release/*
|
||||
sandbox/TODO-VNC.md
|
||||
sandbox/docker/chrome/PLAN.md
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
166
sandbox/docker/chrome/Dockerfile
Normal 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
34
sandbox/docker/chrome/config/chrome-preferences.json
Normal 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
44
sandbox/docker/chrome/config/stealth-init.js
Normal 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
105
sandbox/docker/chrome/tests/README.md
Normal 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
349
sandbox/docker/chrome/tests/demo-baidu.py
Normal 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
275
sandbox/docker/chrome/tests/demo-duckduckgo.py
Normal 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
860
sandbox/docker/chrome/tests/demo-llm-vision.py
Normal 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()
|
||||
Loading…
Add table
Reference in a new issue