""" browser.py — Playwright browser automation. S65 — /screenshot + /navigate stateless (mantenuti per retrocompat) S_NEW — /open + /act + /close con sessioni persistenti + DOM Intelligence S174 — GET /screenshot/{session_id} snapshot sessione attiva senza azioni W-NAV aggiornamenti: - MAX_TEXT 3000→6000: più contesto per l'LLM senza rischiare OOM - _goto_with_networkidle(): networkidle per SPA/React/Vue, fallback domcontentloaded - _dismiss_cookie_banner(): auto-dismiss CSS+JS prima dell'estrazione DOM (W-NAV2) - _extract_text_trafilatura(): estrazione mainbody Readability-quality (W-NAV) - /navigate: usa trafilatura per text_content (da 2000→5000 chars utili) - /open: aggiunto text_content via trafilatura nella risposta Vincoli HF free tier: - Max SESSION_LIMIT sessioni vive (OOM guard: Chromium ~300 MB/sessione) - Timeout sessione: SESSION_TTL_S secondi di inattività - Un solo _browser_lock per aprire nuove sessioni (evita race condition) Problematiche W-NAV anticipate: - networkidle timeout: pagine con polling infinito (ads/analytics/WebSocket) non raggiungono mai networkidle → timeout catturato, flusso continua (domcontentloaded è già avvenuto come prerequisito → DOM accessibile) - Cookie banner loop: _dismiss_cookie_banner() è idempotente e silenziosa — se fallisce il flusso continua normalmente (banner nella DOM, LLM lo vede e può istruire browser_act per gestirlo manualmente) - trafilatura su pagine non-article (login, dashboard, SPA vuota): restituisce None → fallback a DOM innerText evaluation (pre-esistente, sempre funziona) - get_by_text Playwright API: usiamo page.evaluate() JS per text-matching invece di Locator API (più stabile tra versioni playwright, zero versioning issues) """ import os import asyncio, base64, hashlib, os, time, uuid, logging from typing import Optional, Any from fastapi import APIRouter, HTTPException, Request from pydantic import BaseModel router = APIRouter(prefix="/api/browser", tags=["browser"]) _logger = logging.getLogger("browser") # ─── Costanti ───────────────────────────────────────────────────────────────── SESSION_LIMIT = 2 # max sessioni vive (OOM guard HF free) SESSION_TTL_S = 300 # QF-4: 2→5 min inattività — supporta navigazione multi-step più lunga GOTO_TIMEOUT = 25_000 # QF-4: 15→25s goto — SPA pesanti (React/Next.js) servono più tempo ACTION_TIMEOUT = 5_000 # ms per singola azione MAX_LINKS = 25 MAX_INPUTS = 20 MAX_TEXT = 6000 # W-NAV: alzato da 3000→6000 (più contesto per LLM) # ─── Lock + registry sessioni ───────────────────────────────────────────────── _browser_lock = asyncio.Lock() _sessions: dict[str, dict[str, Any]] = {} # ─── Launch args ottimizzati HF Spaces free ─────────────────────────────────── _LAUNCH_ARGS = [ "--no-sandbox", "--disable-setuid-sandbox", "--disable-dev-shm-usage", "--disable-gpu", "--single-process", "--no-zygote", "--disable-extensions", "--disable-background-networking", "--disable-default-apps", "--mute-audio", # S274-SEC5: rimosso --disable-web-security — disabilita Same-Origin Policy → SSRF via siti visitati # "--disable-web-security", # RIMOSSO per sicurezza # ARCH-3: ulteriori flag riduzione memoria (HF free tier ~1GB RAM) "--disable-accelerated-2d-canvas", # disabilita canvas GPU (non usato in headless) "--disable-renderer-backgrounding", # previene throttling renderer in background "--renderer-process-limit=1", # max 1 renderer process (headless, no tabs visibili) "--js-flags=--max-old-space-size=200",# limita heap V8 a 200MB per renderer ] _UA_MOBILE = "Mozilla/5.0 (iPhone; CPU iPhone OS 17_0 like Mac OS X) AppleWebKit/605.1.15" _UA_DESKTOP = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 Chrome/124.0.0.0" _BLOCKED = [ "localhost", "127.0.0.1", "0.0.0.0", "192.168.", "10.0.", "172.16.", "172.17.", "172.18.", "metadata.google", "169.254", ] # ─── W-NAV2: Cookie dismiss — selettori CSS prioritizzati ──────────────────── # Ordine: vendor-specifici (più precisi) → pattern generici (più ampi) # Problematica: selettori troppo generici (es. "button") catturano azioni non-cookie # → usiamo pattern specifici per namespace (id/class con "cookie/consent/gdpr") _COOKIE_DISMISS_SELECTORS = [ "#onetrust-accept-btn-handler", # OneTrust (ubiquo) "#CybotCookiebotDialogBodyButtonAccept", # Cookiebot "#CybotCookiebotDialogBodyLevelButtonLevelOptinAllowAll", "[data-cookiebanner='accept_button']", ".cc-btn.cc-allow", # CookieConsent.js ".cc-accept-all", "#cookie_action_close_header", "#accept-cookies", "#acceptAllCookies", "#acceptCookies", ".cookie-accept-all", ".cookie__btn--accept", "#gdpr-cookie-accept", ".gdpr-accept-all", ".gdpr__btn", "[aria-label='Accept all cookies']", "[aria-label='Accetta tutti i cookie']", "[aria-label='Consenti tutti i cookie']", "[aria-label='Allow all cookies']", "button[id*='cookie'][id*='accept']", "button[id*='accept'][id*='cookie']", "button[class*='cookie'][class*='accept']", ".cookie-consent__accept", ".cookie-notice__accept", ".cookie-banner__accept", "#cookie-accept", ".js-accept-cookies", ] # JS fallback: testo-matching su pulsanti visibili # Problematica: .innerText può essere "" su elementi visibili ma con solo icone # → usiamo .textContent come fallback per innerText # Problematica: false positive su "OK" generico (es. dialog di conferma) # → "ok" solo se il genitore contiene "cookie/consent/gdpr" nel className/id _COOKIE_DISMISS_JS = """() => { const EXACT = [ 'accetta tutto', 'accetta tutti', 'accetta tutti i cookie', 'accept all', 'accept all cookies', 'allow all', 'allow all cookies', 'tout accepter', 'alle akzeptieren', 'aceitar tudo', 'aceptar todo', 'i accept all', 'i agree to all', ]; const PARTIAL = [ 'accetta', 'accept cookies', 'allow cookies', 'consenti tutto', 'ho capito', 'i accept', 'i agree', ]; const isCookieCtx = (el) => { const ctx = (el.id + ' ' + el.className + ' ' + (el.closest('[class*=cookie],[class*=consent],[class*=gdpr],[id*=cookie],[id*=consent],[id*=gdpr]')?.className || '') ).toLowerCase(); return ctx.includes('cookie') || ctx.includes('consent') || ctx.includes('gdpr') || ctx.includes('privacy'); }; const btns = Array.from(document.querySelectorAll( 'button,a,[role=button],[class*=cookie] *,[class*=consent] *,[id*=cookie] *,[id*=gdpr] *' )); for (const el of btns) { const txt = (el.innerText || el.textContent || '').trim().toLowerCase().replace(/\\s+/g, ' '); if (!txt || txt.length > 60) continue; const s = window.getComputedStyle(el); if (s.display === 'none' || s.visibility === 'hidden' || s.opacity === '0') continue; if (EXACT.includes(txt) || (PARTIAL.some(p => txt.includes(p)) && isCookieCtx(el))) { el.click(); return txt.slice(0, 40); } } return null; }""" # ─── DOM Intelligence script ───────────────────────────────────────────────── _DOM_SCRIPT = """() => { const links = [...document.querySelectorAll('a[href]')] .filter(a => a.href.startsWith('http') && a.innerText.trim()) .slice(0, %d) .map(a => ({ text: a.innerText.trim().replace(/\\s+/g, ' ').slice(0, 80), href: a.href, selector: a.id ? '#' + a.id : (a.getAttribute('aria-label') ? '[aria-label="' + a.getAttribute('aria-label') + '"]' : (a.className ? '.' + a.className.split(' ').filter(c=>c&&!c.match(/^[a-z]{1,2}$/)).slice(0,2).join('.') : 'a')) })); const inputs = [...document.querySelectorAll( 'input:not([type=hidden]),textarea,select,button,[role=button],[role=checkbox],[role=radio],[role=switch],[role=combobox]' )] .filter(el => { const s = window.getComputedStyle(el); return s.display !== 'none' && s.visibility !== 'hidden' && s.opacity !== '0'; }) .slice(0, %d) .map(el => { let selector = null; const role = el.getAttribute('role') || el.tagName.toLowerCase(); if (el.id) selector = '#' + el.id; else if (el.getAttribute('name')) selector = '[name="' + el.getAttribute('name') + '"]'; else if (el.getAttribute('aria-label')) selector = '[aria-label="' + el.getAttribute('aria-label') + '"]'; else if (el.placeholder) selector = '[placeholder="' + el.placeholder + '"]'; else if (el.getAttribute('data-testid')) selector = '[data-testid="' + el.getAttribute('data-testid') + '"]'; const tag = el.tagName.toLowerCase(); const isChecked = el.checked !== undefined ? el.checked : null; const currentVal = (tag === 'input' || tag === 'textarea') ? (el.value || '') : null; const isDisabled = el.disabled || el.getAttribute('aria-disabled') === 'true'; let label = el.getAttribute('aria-label') || el.placeholder || el.getAttribute('name') || el.id; if (!label) { const lbl = el.id ? document.querySelector('label[for="'+el.id+'"]') : el.closest('label'); if (lbl) label = lbl.innerText.trim().slice(0, 40); } if (!label) label = el.innerText?.trim().slice(0, 40) || null; return { tag, role, type: el.type || null, label, selector, value: currentVal, checked: isChecked, disabled: isDisabled || false }; }) .filter(el => el.label || el.selector); const headings = [...document.querySelectorAll('h1,h2,h3')] .slice(0, 8) .map(h => ({ level: h.tagName.toLowerCase(), text: h.innerText.trim().slice(0, 80) })); const modals = [...document.querySelectorAll( '[role=dialog],[role=alertdialog],[role=modal],.modal,.dialog,[aria-modal=true]' )] .filter(el => { const s = window.getComputedStyle(el); return s.display !== 'none' && s.visibility !== 'hidden'; }) .slice(0, 3) .map(el => ({ role: el.getAttribute('role') || 'modal', title: el.querySelector('h1,h2,h3,[role=heading]')?.innerText.trim().slice(0,60) || null, selector: el.id ? '#' + el.id : (el.getAttribute('aria-label') ? '[aria-label="'+el.getAttribute('aria-label')+'"]' : '[role="'+(el.getAttribute('role')||'dialog')+'"]'), })); const title = document.title; const desc = document.querySelector('meta[name=description]')?.content?.slice(0, 200) || null; const mainEl = document.querySelector('main,[role=main],article,.content,#content') || document.body; const text = mainEl.innerText.replace(/\\s+/g, ' ').trim().slice(0, %d); return { title, desc, text, links, inputs, headings, modals }; }""" # ─── Helpers ────────────────────────────────────────────────────────────────── async def _get_ax_tree(page: Any, max_depth: int = 5) -> Optional[dict]: """GAP-AX: Playwright Accessibility Tree — 'screenshot testuale' MCP-style. Restituisce una struttura ad albero ARIA che descrive la pagina in modo semantico: ruoli, nomi, stati, relazioni. Usato dall'LLM per interazioni precise senza ambiguità visiva (stile Manus / browser-use). Limitazioni sicure: - max_depth=5: alberi profondi generano payload enormi. 5 livelli coprono il 95% delle pagine senza superare ~8KB di JSON. - Timeout 3s: mai blocca il flusso principale (snapshot è sincrono ma può bloccarsi su pagine con ARIA dinamici in aggiornamento continuo). - Ritorna None su qualsiasi errore: campo opzionale, zero impatto. """ try: # interesting=True: esclude nodi ARIA nascosti (display:none, aria-hidden) snapshot = await asyncio.wait_for( page.accessibility.snapshot(interesting_only=True), timeout=3.0, ) if not snapshot: return None return _trim_ax_tree(snapshot, max_depth) except Exception: return None def _trim_ax_tree(node: dict, depth: int) -> dict: """Riduce ricorsivamente l'albero AX a max_depth livelli. Mantiene: role, name, description, value, checked, expanded, required. Scarta: proprietà interne Playwright (nodeId, backendDOMNodeId, ignoredReasons). """ KEEP = frozenset({role, name, description, value, checked, expanded, required, haspopup, level, pressed, selected, multiselectable, orientation}) result: dict = {k: v for k, v in node.items() if k in KEEP and v not in (None, False, )} if depth > 0 and node.get(children): trimmed = [_trim_ax_tree(c, depth - 1) for c in node[children]] # Filtra nodi completamente vuoti (solo role senza nome né figli) trimmed = [c for c in trimmed if len(c) > 1 or c.get(children)] if trimmed: result[children] = trimmed return result def _safe_url(url: str) -> bool: low = url.lower() return url.startswith(("http://", "https://")) and not any(b in low for b in _BLOCKED) async def _goto_with_networkidle(page: Any, url: str, timeout: int = GOTO_TIMEOUT) -> None: """ Naviga all'URL con wait_until='networkidle' per SPA/React/Vue/Next.js. W-NAV3: never use only domcontentloaded for SPAs — may miss JS-rendered content. Comportamento: - networkidle: aspetta 500ms senza richieste HTTP attive (standard per SPA) - Timeout: se la pagina ha polling infinito (ads, analytics, WebSocket keepalive), il timeout scatta DOPO che domcontentloaded è già avvenuto → DOM accessibile. Catturiamo silenziosamente e continuiamo. - wait_for_load_state fallback: su timeout, forza attesa domcontentloaded (già avvenuto, ritorna subito — è solo un safety net) Nota: timeout di goto con networkidle NON significa pagina non caricata. Significa che ci sono richieste background attive dopo il caricamento visibile. """ try: await page.goto(url, wait_until="networkidle", timeout=timeout) except Exception: # Timeout o navigazione interrotta — il DOM è comunque disponibile try: await page.wait_for_load_state("domcontentloaded", timeout=3000) except Exception: pass # Anche domcontentloaded fallisce? Procediamo — page.content() funziona comunque async def _dismiss_cookie_banner(page: Any) -> bool: """ Auto-dismiss banner cookie prima dell'estrazione DOM. W-NAV2: eseguita dopo ogni _goto_with_networkidle in /open, /navigate, /screenshot. Strategia a 2 fasi: Fase 1: CSS selectors vendor-specifici (precisi, zero false positive) Fase 2: JS text-matching su pulsanti visibili (copre CMP custom e traduzioni) Silenziosa: non lancia mai, timeout brevi (1.5s per selettore) per non bloccare il flusso. Se fallisce, il banner rimane nel DOM e l'LLM lo vede nei dom.modals → può istruire browser_act per gestirlo manualmente. """ # Fase 1: CSS selectors diretti for sel in _COOKIE_DISMISS_SELECTORS: try: el = await page.query_selector(sel) if el: visible = await el.is_visible() if visible: await el.click(timeout=1500) await page.wait_for_timeout(300) _logger.debug("Cookie banner dismissed via CSS: %s", sel) return True except Exception: continue # Fase 2: JS text-matching (CMP custom, traduzioni non standard) try: clicked = await page.evaluate(_COOKIE_DISMISS_JS) if clicked: await page.wait_for_timeout(300) _logger.debug("Cookie banner dismissed via JS text-match: '%s'", clicked) return True except Exception as _exc: _logger.debug("[browser] silenced %s", type(_exc).__name__) # noqa: BLE001 return False async def _extract_text_trafilatura(page: Any, url: str = "", max_chars: int = MAX_TEXT) -> str: """ Estrae il testo mainbody via trafilatura (Readability-quality). Fallback: DOM innerText evaluation se trafilatura non disponibile o restituisce poco. Problematiche: - trafilatura su SPA: l'HTML di page.content() include il DOM post-JS → trafilatura può estrarre più testo rispetto all'HTML statico iniziale - trafilatura su pagine non-article (login, 404): restituisce None → fallback a DOM innerText (sempre disponibile) - max_chars applicato sia a trafilatura che al fallback """ try: import trafilatura # type: ignore[import-untyped] html = await page.content() extracted = trafilatura.extract( html, url=url or None, include_comments=False, include_tables=True, include_images=False, deduplicate=True, favor_recall=True, ) if extracted and len(extracted.strip()) > 200: return extracted[:max_chars] except Exception as _exc: _logger.debug("[browser] silenced %s", type(_exc).__name__) # noqa: BLE001 # Fallback: DOM evaluation (pre-esistente, sempre funziona) try: text = await page.evaluate( "() => (document.querySelector('main,[role=main],article,.content,#content') || document.body)" f".innerText.replace(/\\s+/g,' ').trim().slice(0,{max_chars})" ) return str(text) except Exception: return "" async def _try_persist_screenshot(url: str, png_b64: str, title: str) -> None: try: sb_url = os.getenv("SUPABASE_URL", "") sb_key = os.getenv("SUPABASE_ANON_KEY") or os.getenv("SUPABASE_KEY", "") if not (sb_url and sb_key): return from supabase import create_client sb = create_client(sb_url, sb_key) fid = hashlib.sha256(url.encode()).hexdigest()[:32] now = int(time.time() * 1000) sb.table("vfs_files").upsert({ "id": fid, "path": f"/browser-screenshots/{fid}.png", "language": "image", "content": png_b64, "conversation_id": None, "created_at": now, "updated_at": now, "metadata": {"source_url": url, "title": title}, }).execute() except Exception as _e: _logger.warning("_try_persist_screenshot: Supabase upsert failed: %s", _e) async def _make_context(browser: Any, width: int, height: int, mobile: bool) -> Any: return await browser.new_context( viewport={"width": 390 if mobile else width, "height": height}, is_mobile=mobile, user_agent=_UA_MOBILE if mobile else _UA_DESKTOP, ) async def _execute_actions(page: Any, actions: list) -> None: for action in actions: sel = action.selector or "" try: if action.type == "click" and sel: await page.click(sel, timeout=ACTION_TIMEOUT) await page.wait_for_load_state("domcontentloaded", timeout=5000) elif action.type == "click" and action.x is not None and action.y is not None: # GAP-CLICK: fallback coordinate per elementi senza selector DOM await page.mouse.click(action.x, action.y) await page.wait_for_load_state("domcontentloaded", timeout=5000) elif action.type == "fill" and sel: await page.fill(sel, action.value or "", timeout=ACTION_TIMEOUT) elif action.type == "select" and sel: await page.select_option(sel, action.value or "", timeout=ACTION_TIMEOUT) elif action.type == "press" and sel: await page.press(sel, action.key or "Enter", timeout=ACTION_TIMEOUT) elif action.type == "hover" and sel: await page.hover(sel, timeout=ACTION_TIMEOUT) elif action.type == "wait_for" and sel: await page.wait_for_selector(sel, timeout=ACTION_TIMEOUT) elif action.type == "wait": await page.wait_for_timeout(min(action.ms or 500, 5000)) elif action.type == "scroll": pct = float(action.value or "50") await page.evaluate(f"window.scrollTo(0, document.body.scrollHeight * {pct / 100})") except Exception as e: _logger.debug("Action %s on '%s' failed: %s", action.type, sel, e) # ─── ARCH-7: Browserless.io CDP fallback ───────────────────────────────────── # Se BROWSERLESS_TOKEN è impostato su Railway, usa CDP remoto (zero RAM locale). # Fallback automatico a Chromium locale se token assente o connessione fallisce. _BROWSERLESS_WS = "wss://chrome.browserless.io" async def _get_browser_instance(): """ Ritorna (pw, browser, is_remote). is_remote=True → CDP remoto: non chiamare browser.close() né pw.stop(). is_remote=False → Chromium locale: cleanup normale. """ from playwright.async_api import async_playwright pw = await async_playwright().start() token = os.getenv("BROWSERLESS_TOKEN", "").strip() if token: try: ws = f"{_BROWSERLESS_WS}?token={token}" browser = await pw.chromium.connect_over_cdp(ws, timeout=10_000) _logger.info("Browser: Browserless.io CDP ✓ (zero RAM locale)") return pw, browser, True except Exception as _e: _logger.warning("Browser: CDP non disponibile (%s) — fallback locale", _e) browser = await pw.chromium.launch(headless=True, args=_LAUNCH_ARGS) return pw, browser, False async def _close_session(sid: str, reason: str = "explicit") -> None: sess = _sessions.pop(sid, None) if not sess: return is_remote = sess.get("is_remote", False) try: await sess["context"].close() if not is_remote: await sess["browser"].close() # pw.stop() sempre: chiude la connessione playwright (locale: termina processo; CDP: disconnette) await sess["pw"].stop() _logger.info("Session %s closed (%s, remote=%s)", sid, reason, is_remote) except Exception as e: _logger.warning("Session %s close error: %s", sid, e) async def _session_cleanup_loop() -> None: while True: await asyncio.sleep(30) # ARCH-3: check più frequente (era 60s) now = time.time() expired = [sid for sid, s in list(_sessions.items()) if now - s["last_used"] > SESSION_TTL_S] for sid in expired: await _close_session(sid, "TTL expired") def _start_cleanup() -> None: try: loop = asyncio.get_event_loop() if loop.is_running(): asyncio.create_task(_session_cleanup_loop()) except Exception as _e: _logger.warning("_start_cleanup: create_task failed — cleanup loop not running: %s", _e) # ─── Models ─────────────────────────────────────────────────────────────────── class BrowserAction(BaseModel): type: str selector: Optional[str] = None value: Optional[str] = None key: Optional[str] = None ms: Optional[int] = None x: Optional[float] = None # GAP-CLICK: coordinata X per click-by-position y: Optional[float] = None # GAP-CLICK: coordinata Y per click-by-position class ScreenshotRequest(BaseModel): url: str width: int = 1280 height: int = 800 mobile: bool = False wait_ms: int = 2000 class NavigateRequest(BaseModel): url: str actions: list[BrowserAction] = [] width: int = 1280 height: int = 800 mobile: bool = False wait_ms: int = 2000 class BrowserOpenRequest(BaseModel): url: str actions: list[BrowserAction] = [] mobile: bool = False width: int = 1280 wait_ms: int = 1500 class BrowserActRequest(BaseModel): session_id: str actions: list[BrowserAction] wait_ms: int = 1000 take_screenshot: bool = True class BrowserCloseRequest(BaseModel): session_id: str class DomSnapshot(BaseModel): title: Optional[str] = None desc: Optional[str] = None text: Optional[str] = None links: list[dict] = [] inputs: list[dict] = [] headings: list[dict] = [] modals: list[dict] = [] class BrowserResult(BaseModel): ok: bool session_id: Optional[str] = None screenshot_b64: Optional[str] = None title: Optional[str] = None url: Optional[str] = None text_content: Optional[str] = None dom: Optional[DomSnapshot] = None ax_tree: Optional[dict] = None # GAP-AX: Playwright Accessibility Tree (MCP-style) error: Optional[str] = None warnings: list[str] = [] # ─── verify_goal_browser ────────────────────────────────────────────────────── # Sprint 3b ITEM 8 async def verify_goal_browser( goal: str, url: str, requirements: "list | None" = None, timeout_s: float = 30.0, ) -> dict: if not _safe_url(url): return {"ok": False, "overall": "UNKNOWN", "per_criterion": {}, "error": "URL non consentita"} # S701: se requirements=None, fallback a basic DOM check (era UNKNOWN immediato) # Prima: 90%+ dei casi usciva subito senza verificare nulla. # Ora: check DOM non-empty + zero white screen + JS error interception. if not requirements: try: import playwright # noqa: F401 except ImportError: return {"ok": True, "overall": "UNKNOWN", "per_criterion": {}, "error": "playwright non installato"} try: from playwright.async_api import async_playwright _js_errs: list[str] = [] async with async_playwright() as _pw2: _b2 = await asyncio.wait_for( _pw2.chromium.launch(headless=True, args=_LAUNCH_ARGS), timeout=10.0) _ctx2 = await _b2.new_context(viewport={"width": 1280, "height": 800}, user_agent=_UA_DESKTOP) _pg2 = await _ctx2.new_page() _pg2.on("pageerror", lambda e: _js_errs.append(str(e))) try: await asyncio.wait_for( _goto_with_networkidle(_pg2, url, GOTO_TIMEOUT), timeout=min(timeout_s, 15.0), ) await _pg2.wait_for_timeout(600) _body_txt = await _pg2.evaluate("document.body ? document.body.innerText.trim() : ''") _body_html = await _pg2.evaluate("document.body ? document.body.innerHTML.trim() : ''") _title2 = await _pg2.title() finally: await _ctx2.close() await _b2.close() _is_white = len(_body_txt) < 5 and len(_body_html) < 30 _has_js_err = bool(_js_errs) if _is_white: _overall2 = "FAIL" _crit2 = {"dom_not_empty": "FAIL", "js_errors": "PASS" if not _has_js_err else "FAIL"} elif _has_js_err: _overall2 = "FAIL" _crit2 = {"dom_not_empty": "PASS", "js_errors": "FAIL"} else: _overall2 = "PASS" _crit2 = {"dom_not_empty": "PASS", "js_errors": "PASS"} # S701 R5: telemetria DOM check try: from api.state import increment_stat as _inc_dom _inc_dom("browser_dom_check_pass" if _overall2 == "PASS" else "browser_dom_check_fail") except Exception as _exc: _logger.debug("[browser] silenced %s", type(_exc).__name__) # noqa: BLE001 return {"ok": True, "overall": _overall2, "per_criterion": _crit2, "error": None, "title": _title2, "js_errors": _js_errs[:3]} except asyncio.TimeoutError: return {"ok": True, "overall": "UNKNOWN", "per_criterion": {}, "error": "timeout"} except Exception as _be: return {"ok": True, "overall": "UNKNOWN", "per_criterion": {}, "error": str(_be)[:200]} try: import playwright # noqa: F401 except ImportError: return { "ok": True, "overall": "UNKNOWN", "per_criterion": {}, "error": "playwright non installato — browser verify disabilitato", } per_criterion: dict[str, str] = {} try: from playwright.async_api import async_playwright async with async_playwright() as _pw: _browser = await _pw.chromium.launch(headless=True, args=_LAUNCH_ARGS) _ctx = await _browser.new_context( viewport={"width": 1280, "height": 800}, user_agent=_UA_DESKTOP, ) _page = await _ctx.new_page() try: # W-NAV3: usa networkidle anche per verify_goal_browser await asyncio.wait_for( _goto_with_networkidle(_page, url, GOTO_TIMEOUT), timeout=min(timeout_s, GOTO_TIMEOUT / 1000), ) await _page.wait_for_timeout(1000) _criteria: list[str] = [] for _req in (requirements or [])[:5]: _ac = getattr(_req, "acceptance_criteria", None) or [] _criteria.extend(str(c) for c in _ac[:2]) _criteria = _criteria[:5] for _crit in _criteria: _low = _crit.lower() _verdict = "UNKNOWN" try: if any(k in _low for k in ["login", "autenti", "sessione", "http 200", "200 ok"]): _el = await _page.query_selector("form, input[type=password], input[type=email]") _verdict = "PASS" if _el else "FAIL" elif any(k in _low for k in ["crud", "lista", "record", "endpoint", "array", "json"]): _body = await _page.evaluate("() => document.body.innerText.slice(0, 3000)") _verdict = "PASS" if len(str(_body)) > 100 else "FAIL" elif any(k in _low for k in ["dashboard", "dati", "metriche", "renderizza", "mostra"]): _el = await _page.query_selector("main, [role=main], .dashboard, #app, #root, table, chart") _verdict = "PASS" if _el else "FAIL" elif any(k in _low for k in ["form", "validaz", "campo", "submit", "bottone"]): _el = await _page.query_selector("form, input, textarea, button[type=submit]") _verdict = "PASS" if _el else "FAIL" elif any(k in _low for k in ["errore", "error", "400", "401", "403", "fallisce"]): _verdict = "UNKNOWN" else: _title = await _page.title() _verdict = "PASS" if _title else "FAIL" except Exception: _verdict = "UNKNOWN" per_criterion[_crit[:80]] = _verdict finally: await _ctx.close() await _browser.close() _pass_n = sum(1 for v in per_criterion.values() if v == "PASS") _total = len(per_criterion) if _total == 0: _overall = "UNKNOWN" elif _pass_n / _total >= 0.5: _overall = "PASS" else: _overall = "FAIL" return {"ok": True, "overall": _overall, "per_criterion": per_criterion, "error": None} except ImportError: return {"ok": False, "overall": "UNKNOWN", "per_criterion": {}, "error": "Playwright non installato"} except Exception as _e: return {"ok": False, "overall": "UNKNOWN", "per_criterion": per_criterion, "error": str(_e)[:300]} # S588 # ─── /screenshot ───────────────────────────────────────────────────────────── @router.post("/screenshot", response_model=BrowserResult) async def browser_screenshot(req: ScreenshotRequest): """Screenshot headless di una pagina web (stateless).""" if not _safe_url(req.url): raise HTTPException(400, "URL non consentita") async with _browser_lock: try: from playwright.async_api import async_playwright async with async_playwright() as pw: browser = await pw.chromium.launch(headless=True, args=_LAUNCH_ARGS) ctx = await _make_context(browser, req.width, req.height, req.mobile) page = await ctx.new_page() try: # W-NAV3: networkidle per SPA await _goto_with_networkidle(page, req.url, GOTO_TIMEOUT) # W-NAV2: dismiss cookie banner prima dello screenshot await _dismiss_cookie_banner(page) await page.wait_for_timeout(req.wait_ms) png = await page.screenshot(type="png", full_page=False) title = await page.title() png_b64 = base64.b64encode(png).decode() asyncio.create_task(_try_persist_screenshot(req.url, png_b64, title)) return BrowserResult(ok=True, screenshot_b64=png_b64, title=title, url=page.url) except Exception as e: return BrowserResult(ok=False, error=str(e)[:500]) # S599: 300→500 finally: await ctx.close() await browser.close() except Exception as e: return BrowserResult(ok=False, error=str(e)[:500]) # ─── /navigate ──────────────────────────────────────────────────────────────── @router.post("/navigate", response_model=BrowserResult) async def browser_navigate(req: NavigateRequest): """ Naviga, esegui azioni, restituisce screenshot + testo (stateless). W-NAV: text_content ora estratto via trafilatura (da 2000→5000 chars utili). """ if not _safe_url(req.url): raise HTTPException(400, "URL non consentita") async with _browser_lock: try: from playwright.async_api import async_playwright async with async_playwright() as pw: browser = await pw.chromium.launch(headless=True, args=_LAUNCH_ARGS) ctx = await _make_context(browser, req.width, req.height, req.mobile) page = await ctx.new_page() try: # W-NAV3: networkidle per SPA await _goto_with_networkidle(page, req.url, GOTO_TIMEOUT) # W-NAV2: dismiss cookie prima delle azioni await _dismiss_cookie_banner(page) await page.wait_for_timeout(500) await _execute_actions(page, req.actions) await page.wait_for_timeout(req.wait_ms) png = await page.screenshot(type="png", full_page=False) title = await page.title() # W-NAV: trafilatura estrae mainbody, molto più testo utile text = await _extract_text_trafilatura(page, req.url, max_chars=5000) png_b64 = base64.b64encode(png).decode() asyncio.create_task(_try_persist_screenshot(page.url, png_b64, title)) ax_tree = await _get_ax_tree(page) # GAP-AX return BrowserResult( ok=True, screenshot_b64=png_b64, title=title, url=page.url, text_content=text[:5000] if text else None, ax_tree=ax_tree, ) except Exception as e: return BrowserResult(ok=False, error=str(e)[:500]) finally: await ctx.close() await browser.close() except Exception as e: return BrowserResult(ok=False, error=str(e)[:500]) # ─── /open ──────────────────────────────────────────────────────────────────── @router.post("/open", response_model=BrowserResult) async def browser_open(req: BrowserOpenRequest, request: Request): """ Apre una sessione Playwright persistente, naviga all'URL, restituisce session_id + screenshot + mappa DOM + text_content (trafilatura). """ _internal_token = os.getenv('INTERNAL_TOKEN', '') if _internal_token and request.headers.get('X-Internal-Token') != _internal_token: raise HTTPException(401, 'Unauthorized') if not _safe_url(req.url): raise HTTPException(400, "URL non consentita") if len(_sessions) >= SESSION_LIMIT: oldest = min(_sessions, key=lambda sid: _sessions[sid]["last_used"]) await _close_session(oldest, "OOM guard") async with _browser_lock: try: # ARCH-7: CDP remoto (Browserless) se BROWSERLESS_TOKEN, else locale pw, browser, _is_remote = await _get_browser_instance() ctx = await _make_context(browser, req.width, 800, req.mobile) page = await ctx.new_page() # W-NAV3: networkidle per SPA await _goto_with_networkidle(page, req.url, GOTO_TIMEOUT) # W-NAV2: dismiss cookie prima dell'estrazione DOM await _dismiss_cookie_banner(page) await page.wait_for_timeout(500) await _execute_actions(page, req.actions) await page.wait_for_timeout(req.wait_ms) png = await page.screenshot(type="png", full_page=False) title = await page.title() png_b64 = base64.b64encode(png).decode() dom_raw = await page.evaluate(_DOM_SCRIPT % (MAX_LINKS, MAX_INPUTS, MAX_TEXT)) # W-NAV: text_content via trafilatura (aggiunto — prima non era nella risposta /open) text = await _extract_text_trafilatura(page, req.url, max_chars=MAX_TEXT) sid = uuid.uuid4().hex[:16] _sessions[sid] = { "pw": pw, "browser": browser, "context": ctx, "page": page, "created_at": time.time(), "last_used": time.time(), "url": page.url, "click_history": [], "visited_urls": [page.url], "action_log": [], "is_remote": _is_remote, # ARCH-7: True=CDP, False=locale } asyncio.create_task(_try_persist_screenshot(page.url, png_b64, title)) dom = DomSnapshot(**dom_raw) if isinstance(dom_raw, dict) else None ax_tree = await _get_ax_tree(page) # GAP-AX: Accessibility Tree MCP-style return BrowserResult( ok=True, session_id=sid, screenshot_b64=png_b64, title=title, url=page.url, dom=dom, text_content=text[:MAX_TEXT] if text else None, ax_tree=ax_tree, ) except Exception as e: return BrowserResult(ok=False, error=str(e)[:500]) # S603: 400→500 # ─── /act ───────────────────────────────────────────────────────────────────── @router.post("/act", response_model=BrowserResult) async def browser_act(req: BrowserActRequest, request: Request): """ Esegue azioni su una sessione aperta. Restituisce screenshot + mappa DOM + warnings anti-loop. """ _internal_token = os.getenv('INTERNAL_TOKEN', '') if _internal_token and request.headers.get('X-Internal-Token') != _internal_token: raise HTTPException(401, 'Unauthorized') sess = _sessions.get(req.session_id) if not sess: raise HTTPException(404, f"Sessione {req.session_id} non trovata o scaduta") sess["last_used"] = time.time() page = sess["page"] warnings: list[str] = [] CLICK_HISTORY_MAX = 20 LOOP_THRESHOLD = 3 for action in req.actions: if action.type == "click" and action.selector: history: list[str] = sess.get("click_history", []) repeat_count = history.count(action.selector) if repeat_count >= LOOP_THRESHOLD: warnings.append( f"⚠️ ANTI-LOOP: selettore '{action.selector}' già cliccato " f"{repeat_count}x — cambia strategia (prova altro selettore, scroll, modal, ecc.)" ) history.append(action.selector) sess["click_history"] = history[-CLICK_HISTORY_MAX:] try: url_before = page.url await _execute_actions(page, req.actions) await page.wait_for_timeout(req.wait_ms) url_after = page.url sess["url"] = url_after visited: list[str] = sess.get("visited_urls", []) if url_after not in visited: visited.append(url_after) sess["visited_urls"] = visited[-30:] action_log: list[dict] = sess.get("action_log", []) for a in req.actions: action_log.append({ "type": a.type, "selector": a.selector, "value": a.value, "url_after": url_after, }) sess["action_log"] = action_log[-50:] title = await page.title() dom_raw = await page.evaluate(_DOM_SCRIPT % (MAX_LINKS, MAX_INPUTS, MAX_TEXT)) dom = DomSnapshot(**dom_raw) if isinstance(dom_raw, dict) else None if dom and dom.modals: modal_titles = [m.get("title") or m.get("role", "modal") for m in dom.modals] warnings.append( f"🔔 MODAL RILEVATO: {', '.join(str(t) for t in modal_titles)} " "— potrebbe bloccare le azioni. Chiudilo prima di continuare." ) ax_tree = await _get_ax_tree(page) if req.take_screenshot else None # GAP-AX result = BrowserResult( ok=True, session_id=req.session_id, url=page.url, title=title, dom=dom, warnings=warnings, ax_tree=ax_tree, ) if req.take_screenshot: png = await page.screenshot(type="png", full_page=False) png_b64 = base64.b64encode(png).decode() result.screenshot_b64 = png_b64 asyncio.create_task(_try_persist_screenshot(page.url, png_b64, title)) return result except Exception as e: return BrowserResult(ok=False, session_id=req.session_id, error=str(e)[:500], warnings=warnings) # S603 # ─── /close ─────────────────────────────────────────────────────────────────── @router.post("/close", response_model=BrowserResult) async def browser_close(req: BrowserCloseRequest): """Chiude esplicitamente una sessione persistente e libera risorse.""" if req.session_id not in _sessions: return BrowserResult(ok=True, session_id=req.session_id) await _close_session(req.session_id, "explicit") return BrowserResult(ok=True, session_id=req.session_id) # ─── GET /screenshot/{session_id} ───────────────────────────────────────────── @router.get("/screenshot/{session_id}", response_model=BrowserResult) async def browser_session_screenshot(session_id: str, full_page: bool = False): """Snapshot della pagina corrente senza azioni. Aggiorna last_used.""" sess = _sessions.get(session_id) if not sess: raise HTTPException(404, f"Sessione {session_id} non trovata o scaduta") sess["last_used"] = time.time() page = sess["page"] try: png = await page.screenshot(type="png", full_page=full_page) title = await page.title() png_b64 = base64.b64encode(png).decode() asyncio.create_task(_try_persist_screenshot(page.url, png_b64, title)) return BrowserResult( ok=True, session_id=session_id, screenshot_b64=png_b64, title=title, url=page.url, ) except Exception as e: return BrowserResult(ok=False, session_id=session_id, error=str(e)[:500]) # S603 # ─── /sessions ──────────────────────────────────────────────────────────────── @router.get("/sessions") async def list_sessions(): now = time.time() return { sid: { "url": s["url"], "idle_s": int(now - s["last_used"]), "age_s": int(now - s["created_at"]), } for sid, s in _sessions.items() } _start_cleanup()