api/browser.py

"""
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()