nicola.leonardi
/
wcag_AI_validation
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
wcag_AI_validation/scripts/browser-use/webagent-openai-full-log.py

318 lines
12 KiB
Python
Raw Permalink Blame History

#pip install browser-use langchain-openai langchain-ollama playwright
#playwright install
## comandi da CLI
"""
browser-use open https://example.com # Navigate to URL
browser-use state # See clickable elements
browser-use click 5 # Click element by index
browser-use type "Hello" # Type text
browser-use screenshot page.png # Take screenshot
browser-use close # Close browser
"""
import asyncio
import aiofiles
import os
import logging
import sys
import re
from datetime import datetime
from browser_use import Agent, ChatAzureOpenAI, Browser
import base64, os, json
from browser_use.browser import BrowserProfile, BrowserSession
from browser_use.browser.profile import ViewportSize
# highlight_elements goes on BrowserProfile, not Browser
browser_profile = BrowserProfile(
headless=False, #Visible window opens on screen
highlight_elements=True, # draws colored bounding boxes on interactive elements
# dom_highlight_elements=True, # alternative: DOM-based highlighting (takes priority if both set)
window_size=ViewportSize(width=900, height=1100), # bigger window to compensate for OS chrome
viewport=ViewportSize(width=768, height=1024), # ✅ forces exact page content size
)
browser_session = BrowserSession(browser_profile=browser_profile)
# ─── LOGGING SETUP ────────────────────────────────────────────────────────────
os.environ["BROWSER_USE_LOGGING_LEVEL"] = "debug"
# ── BBox tracker — captures the count from the log message ──────────────────
class BBoxTracker:
"""Intercepts the BBox filtering log line and stores the latest count."""
last_excluded: int = 0
history: list[dict] = []
def record(self, step: int, count: int):
self.last_excluded = count
self.history.append({"step": step, "excluded_nodes": count})
bbox_tracker = BBoxTracker()
_BBOX_RE = re.compile(r"BBox filtering excluded (\d+) nodes", re.IGNORECASE)
class BBoxInterceptHandler(logging.Handler):
"""Silently watches every log record for the BBox line."""
def emit(self, record: logging.LogRecord):
msg = record.getMessage()
m = _BBOX_RE.search(msg)
if m:
bbox_tracker.last_excluded = int(m.group(1))
# Colored formatter
class ColoredFormatter(logging.Formatter):
COLORS = {
logging.DEBUG: "\033[36m",
logging.INFO: "\033[32m",
logging.WARNING: "\033[33m",
logging.ERROR: "\033[31m",
logging.CRITICAL: "\033[35m",
}
RESET = "\033[0m"
BOLD = "\033[1m"
def format(self, record):
color = self.COLORS.get(record.levelno, self.RESET)
record.levelname = f"{color}{self.BOLD}{record.levelname:<8}{self.RESET}"
record.name = f"\033[90m{record.name}{self.RESET}"
return super().format(record)
console_handler = logging.StreamHandler(sys.stdout)
console_handler.setLevel(logging.DEBUG)
console_handler.setFormatter(ColoredFormatter(
fmt="%(asctime)s │ %(levelname)s │ %(name)s\n └─ %(message)s\n",
datefmt="%H:%M:%S",
))
root_logger = logging.getLogger()
root_logger.setLevel(logging.DEBUG)
root_logger.handlers.clear()
root_logger.addHandler(console_handler)
root_logger.addHandler(BBoxInterceptHandler()) # ← silent tracker
for noisy in ("httpx", "httpcore", "playwright", "urllib3", "asyncio"):
logging.getLogger(noisy).setLevel(logging.WARNING)
logger = logging.getLogger("agent_runner")
# ─── STEP CALLBACKS ───────────────────────────────────────────────────────────
step_count = 0
step_start_time: datetime | None = None
os.makedirs("./screenshots_on_step", exist_ok=True)
import asyncio
import aiofiles
import base64
import io
import os
from PIL import Image, ImageDraw, ImageFont
async def on_step(state, model_output, step_num):
if not state.screenshot:
return
img = Image.open(io.BytesIO(base64.b64decode(state.screenshot)))
draw = ImageDraw.Draw(img)
selector_map = state.dom_state.selector_map
colors = ["#FF0000", "#FF6600", "#0066FF", "#009900", "#9900CC"]
for idx, node in selector_map.items():
# ✅ node IS the EnhancedDOMTreeNode directly — no .original_node
rect = node.absolute_position
if rect is None:
continue
x, y, w, h = rect.x, rect.y, rect.width, rect.height
if w <= 0 or h <= 0:
continue
color = colors[idx % len(colors)]
draw.rectangle([x, y, x + w, y + h], outline=color, width=2)
label = str(idx)
draw.rectangle([x, y - 14, x + len(label) * 8 + 4, y], fill=color)
draw.text((x + 2, y - 13), label, fill="white")
path = f"./screenshots_on_step/step_{step_num:03d}.png"
buf = io.BytesIO()
img.save(buf, format="PNG")
async with aiofiles.open(path, "wb") as f:
await f.write(buf.getvalue())
print(f"[step {step_num}] annotated: {path} ({len(selector_map)} elements) | {state.url}")
async def on_step_start(agent):
global step_count, step_start_time
step_count += 1
step_start_time = datetime.now()
bbox_tracker.last_excluded = 0 # reset for this step
print(f"\n\033[1;34m{'━'*60}\033[0m")
print(f"\033[1;34m 🚀 STEP {step_count} — {step_start_time.strftime('%H:%M:%S')}\033[0m")
print(f"\033[1;34m{'━'*60}\033[0m")
async def on_step_end(agent):
global step_start_time
elapsed = (datetime.now() - step_start_time).total_seconds() if step_start_time else 0
# record bbox stats for this step
bbox_tracker.record(step_count, bbox_tracker.last_excluded)
history = agent.history.history
last = history[-1] if history else None
url = "N/A"
if last and last.state and hasattr(last.state, "url"):
url = last.state.url
actions = "N/A"
if last and last.model_output and hasattr(last.model_output, "action"):
acts = last.model_output.action
if isinstance(acts, list):
actions = ", ".join(
type(a).__name__ if not isinstance(a, dict) else list(a.keys())[0]
for a in acts
)
else:
actions = str(acts)
result_text = "N/A"
if last and last.result:
results = last.result if isinstance(last.result, list) else [last.result]
parts = []
for r in results:
if hasattr(r, "extracted_content") and r.extracted_content:
parts.append(f"📄 {str(r.extracted_content)[:120]}")
elif hasattr(r, "error") and r.error:
parts.append(f"❌ {str(r.error)[:120]}")
result_text = " | ".join(parts) if parts else "✔ done"
# BBox colour: green if low, yellow if medium, red if high
bbox_n = bbox_tracker.last_excluded
if bbox_n < 20:
bbox_color = "\033[32m"
elif bbox_n < 60:
bbox_color = "\033[33m"
else:
bbox_color = "\033[31m"
print(f"\033[32m ✅ STEP {step_count} COMPLETE ({elapsed:.2f}s)\033[0m")
print(f"\033[90m URL :\033[0m {url}")
print(f"\033[90m Actions:\033[0m {actions}")
print(f"\033[90m Result :\033[0m {result_text}")
print(f"\033[90m BBox :\033[0m {bbox_color}{bbox_n} nodes pruned (redundant children of links/buttons)\033[0m")
print(f"\033[1;34m{'━'*60}\033[0m\n")
# ✅ highlight_elements draws bounding boxes from DOM on the live browser
#browser = Browser(
# headless=False,
# highlight_elements=True,
#)
# ─── MAIN ─────────────────────────────────────────────────────────────────────
async def run_azure_agent():
logger.info("Initializing Azure OpenAI LLM...")
llm = ChatAzureOpenAI(
model="gpt-4o",
azure_endpoint="https://hiis-accessibility-fonderia.cognitiveservices.azure.com/",
azure_deployment="gpt-4o",
api_version="2025-01-01-preview",
api_key=""
)
agent = Agent(
#task=(
# "Open the URL https://github.com/rasbt and check if the page offers "
# "an alternative function to the interactive map of content in the year "
# "(contribution heatmap data visualization). 'Equivalent' is to be interpreted "
# "according to the WCAG Success Criterion 2.5.8 Target Size (Minimum): "
# "'The function can be achieved through a different control on the same page "
# "that meets this criterion' - Answer YES or NO and justify your answer."
#)
task=(
"Open the URL https://github.com/rasbt and check if the page offers "
"an 'Equivalent' function to the ”Contribution graph cells”: The dense grid of small squares in "
"the '1,535 contributions in the last year' heatmap calendar."
"'Equivalent' is to be interpreted according to the WCAG Success Criterion 2.5.8 Target Size (Minimum): "
"'The function can be achieved through a different control on the same page "
"that meets this criterion'. - Answer YES or NO and justify your answer."
),
llm=llm,
#on_step_start=on_step_start,
#on_step_end=on_step_end,
save_conversation_path="./logs/session.json",# all screenshots embedded here
save_conversation_path_encoding="utf-8", # default
# Optional: keep full history (don't prune steps)
max_history_items=None, # None = keep ALL steps in memory
use_vision=True,
browser_session=browser_session, #browser # pass session, not Browser()
vision_detail_level="high", # 'low', 'high', or 'auto'
generate_gif="./logs/session.gif",
register_new_step_callback=on_step, # real-time callback
)
print(f"\n\033[1;35m{'═'*60}\033[0m")
print(f"\033[1;35m 🤖 BROWSER-USE AGENT STARTING\033[0m")
print(f"\033[1;35m{'═'*60}\033[0m\n")
result = await agent.run(max_steps=20)
#await browser.close()
"""
#### non va ###
import os
import base64
import json
os.makedirs("screenshots_from_results", exist_ok=True)
for i, shot in enumerate(result.screenshots()):
# screenshots() returns base64 strings
img_bytes = base64.b64decode(shot)
path = f"screenshots_from_results/step_{i+1:02d}.png"
with open(path, "wb") as f:
f.write(img_bytes)
print(f"Saved {path}")
"""
# ── BBox summary across all steps ──────────────────────────────────────
print(f"\n\033[1;33m{'═'*60}\033[0m")
print(f"\033[1;33m 📦 BBOX FILTERING SUMMARY\033[0m")
print(f"\033[1;33m{'═'*60}\033[0m")
total = sum(s["excluded_nodes"] for s in bbox_tracker.history)
for s in bbox_tracker.history:
bar = "█" * min(s["excluded_nodes"] // 5, 30)
print(f" Step {s['step']:>2} │ {s['excluded_nodes']:>4} nodes {bar}")
print(f" {'─'*40}")
print(f" Total pruned across all steps: {total} nodes")
print(f"\033[1;33m{'═'*60}\033[0m\n")
print(f"\n\033[1;35m{'═'*60}\033[0m")
print(f"\033[1;35m 🏁 FINAL RESULT\033[0m")
print(f"\033[1;35m{'═'*60}\033[0m")
print(result)
if __name__ == "__main__":
asyncio.run(run_azure_agent())
Reference in New Issue View Git Blame Copy Permalink