tools/web_fetch.py

"""
web_fetch.py — Scarica + pulisce una pagina web. NON restituisce HTML grezzo.

W-NAV: pipeline aggiornata con 3 livelli:
  1. httpx (statico) + trafilatura come primary parser
  2. Se content < 300 chars dopo pulizia → Playwright fallback (SPA/React/Vue/Next.js)
  3. Se Playwright non disponibile → ritorna il poco testo estratto con warning

Problematiche anticipate:
  - SPA JS-heavy: httpx riceve HTML quasi vuoto (~50 chars). Soglia 300 chars
    copre questi casi senza attivare Playwright su pagine statiche normali.
  - Playwright in web_fetch.py è indipendente da browser.py (zero import circolari).
    Crea una sessione Playwright usa-e-getta, non persiste in _sessions.
  - OOM guard: _fetch_with_playwright usa asyncio.Lock per serializzare i launch
    e ha un timeout aggressivo (12s) per non occupare memoria a lungo.
  - trafilatura fallisce silenziosamente su pagine login/404/SPA vuote →
    _clean_html regex come secondo fallback.
  - TIMEOUT_PLAYWRIGHT < GOTO_TIMEOUT di browser.py (20s vs 25s) — intenzionale:
    web_fetch è un tool "leggero", browser.py è il tool "pesante" per sessioni
    interattive.
"""
import asyncio
import re
import httpx
import ipaddress
import socket
from urllib.parse import urlparse
from typing import Optional

import logging
_logger = logging.getLogger("tools.web_fetch")


# SAS (P45-SAS): Semantic Anchor Scoring — lazy import, fail-safe fallback
_sas_available = False
try:
    from tools import semantic_nav as _semantic_nav  # noqa: PLC0415
    _sas_available = True
except ImportError:
    pass

USER_AGENT = "Mozilla/5.0 (compatible; AgenteAI/3.0)"
MAX_CONTENT = 6000          # W-NAV: alzato da 5000 → 6000 (allineato a MAX_TEXT browser.py)
_PLAYWRIGHT_THRESHOLD = 300 # chars: se content < soglia → prova Playwright
_TIMEOUT_PLAYWRIGHT = 20_000  # QF-3: 12→20s — SPA complesse hanno bisogno di più tempo (LCP/hydration)

# Lock globale: un solo Playwright instance alla volta in web_fetch
# (browser.py ha i suoi; vogliamo max 3 Chromium totali: 2 browser.py + 1 web_fetch)
_playwright_lock = asyncio.Lock()


# ─── HTML cleaners ────────────────────────────────────────────────────────────

def _clean_html(html: str) -> str:
    """Regex-based cleaner — fallback quando trafilatura non è disponibile."""
    html = re.sub(
        r"<(script|style|nav|footer|header|noscript|aside)[^>]*>.*?</\1>",
        " ", html, flags=re.S | re.I,
    )
    text = re.sub(r"<[^>]+>", " ", html)
    text = re.sub(r"[ \t]+", " ", text)
    text = re.sub(r"\n{3,}", "\n\n", text)
    return text.strip()


def _extract_meta(html: str) -> dict:
    t = re.search(r"<title[^>]*>([^<]+)</title>", html, re.I)
    d = re.search(
        r'<meta[^>]+name=["\']description["\'][^>]+content=["\']([^"\']+)["\']',
        html, re.I,
    )
    return {
        "title":       t.group(1).strip() if t else "",
        "description": d.group(1).strip() if d else "",
    }



def _extract_relevant_links(html: str, base_url: str, max_links: int = 10) -> list[dict]:
    """
    ARL: estrae link rilevanti dall'HTML grezzo per navigazione ricorsiva (deepResearchTool).
    Filtra: social, CDN, tracking, ancore interne, mailto, javascript.
    Ordina: stesso dominio in testa (più probabile contenuto rilevante).
    """
    from urllib.parse import urljoin, urlparse as _up
    base_domain = _up(base_url).netloc
    skip = re.compile(
        r'(twitter\.com|x\.com|facebook\.com|instagram\.com|linkedin\.com|'
        r'cdn\.|analytics|pixel|tracking|googletagmanager|doubleclick|'
        r'javascript:|mailto:|tel:)',
        re.I
    )
    links: list[dict] = []
    for m in re.finditer(
        r'<a[^>]+href=["\'\']([^\"\'\'#][^\"\'\']*)["\'\'][^>]*>([^<]{3,80})</a>',
        html, re.I
    ):
        raw_href, text = m.group(1).strip(), m.group(2).strip()
        if not raw_href or skip.search(raw_href):
            continue
        url = urljoin(base_url, raw_href)
        if not url.startswith(("http://", "https://")):
            continue
        entry = {"url": url, "text": text[:60]}
        if _up(url).netloc == base_domain:
            links.insert(0, entry)  # stesso dominio in testa
        else:
            links.append(entry)
    return links[:max_links]


def _parse_content(html: str, url: str = "", max_chars: int = MAX_CONTENT) -> str:
    """
    Parser livello 1: prova trafilatura, fallback a _clean_html.
    Separato da fetch_page per essere usabile anche da _fetch_with_playwright.
    """
    try:
        import trafilatura  # type: ignore[import-untyped]
        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:
            if len(extracted) > max_chars:
                extracted = extracted[:max_chars] + "\n[troncato]"
            return extracted
    except Exception as _exc:
        _logger.debug("[web_fetch] silenced %s", type(_exc).__name__)  # noqa: BLE001
    # Fallback regex
    text = _clean_html(html)
    if len(text) > max_chars:
        text = text[:max_chars] + "\n[...troncato...]"
    return text


# ─── Playwright fallback (usa-e-getta, nessun import da browser.py) ───────────

async def _fetch_with_playwright(url: str, max_chars: int = MAX_CONTENT) -> Optional[str]:
    """
    Fetch Playwright usa-e-getta per SPA/React/Vue/Next.js.
    W-NAV1: attivato da fetch_page() quando content < _PLAYWRIGHT_THRESHOLD.

    Design:
      - asyncio.Lock globale: serializza i launch (max 1 alla volta da web_fetch)
      - Timeout aggressivo: _TIMEOUT_PLAYWRIGHT = 20s (usa-e-getta ≠ sessione persistente)
      - networkidle: aspetta 500ms senza richieste di rete (copre SPA con lazy loading)
        fallback a domcontentloaded se timeout
      - trafilatura: estrae il mainbody dall'HTML grezzo di Playwright
      - Sempre close() nel finally: zero memory leak
      - Restituisce None se playwright non installato o fallisce (caller gestisce)
    """
    try:
        import playwright  # noqa: F401 — verifica disponibilità prima del lock
    except ImportError:
        return None  # playwright non installato — caller usa il poco testo statico

    async with _playwright_lock:
        try:
            from playwright.async_api import async_playwright
            async with async_playwright() as pw:
                browser = await pw.chromium.launch(
                    headless=True,
                    args=[
                        "--no-sandbox", "--disable-setuid-sandbox",
                        "--disable-dev-shm-usage", "--disable-gpu",
                        "--single-process", "--no-zygote",
                        "--disable-extensions", "--mute-audio",
                        "--disable-background-networking",
                    ],
                )
                ctx = await browser.new_context(
                    viewport={"width": 1280, "height": 800},
                    user_agent=USER_AGENT,
                )
                page = await ctx.new_page()
                try:
                    # networkidle per SPA — fallback su timeout (polling infinito)
                    try:
                        await page.goto(url, wait_until="networkidle", timeout=_TIMEOUT_PLAYWRIGHT)
                    except Exception:
                        try:
                            await page.wait_for_load_state("domcontentloaded", timeout=3000)
                        except Exception as _exc:
                            _logger.debug("[web_fetch] silenced %s", type(_exc).__name__)  # noqa: BLE001

                    # Breve attesa per rendering JS post-load
                    await page.wait_for_timeout(500)

                    # Estrai HTML e parsa con trafilatura
                    html = await page.content()
                    text = _parse_content(html, url=url, max_chars=max_chars)
                    return text if text and len(text.strip()) > 100 else None

                finally:
                    await ctx.close()
                    await browser.close()

        except Exception:
            return None


# ─── Endpoint principale ──────────────────────────────────────────────────────

async def fetch_page(url: str, max_chars: int = MAX_CONTENT, navigation_intent_keywords: list[str] | None = None) -> dict:
    """
    Fetch + estrazione testo con pipeline a 3 livelli:
      L1: httpx statico + trafilatura (veloce, nessun overhead)
      L2: Playwright usa-e-getta se content < 300 chars (SPA/React/Vue)
      L3: warning nel result se entrambi falliscono (mai ritornare stringa vuota)

    Invariante W-NAV1: mai ritornare content vuoto senza aver tentato il browser.
    """
    try:
        parsed = urlparse(url)
        if parsed.scheme not in ("http", "https"):
            return {"url": url, "error": "Schema non supportato", "content": ""}

        # S780-SSRF: blocca IP privati/loopback prima della fetch per prevenire SSRF
        _host = parsed.hostname or ""
        try:
            _ip = ipaddress.ip_address(socket.gethostbyname(_host))
            if _ip.is_private or _ip.is_loopback or _ip.is_link_local or _ip.is_reserved:
                return {"url": url, "error": "SSRF bloccato: IP privato/loopback non permesso", "content": ""}
        except Exception:
            pass  # hostname non risolvibile o parsing fallito — httpx gestirà l'errore

        async with httpx.AsyncClient(
            timeout=15,
            follow_redirects=True,
            headers={"User-Agent": USER_AGENT},
        ) as c:
            r = await c.get(url)
            html = r.text
            meta = _extract_meta(html)

            # L1: trafilatura / regex parse
            text = _parse_content(html, url=url, max_chars=max_chars)
            used_playwright = False

            # L2: Playwright fallback per SPA/JS-heavy (W-NAV1)
            if len(text.strip()) < _PLAYWRIGHT_THRESHOLD:
                pw_text = await _fetch_with_playwright(url, max_chars=max_chars)
                if pw_text and len(pw_text.strip()) >= _PLAYWRIGHT_THRESHOLD:
                    text = pw_text
                    used_playwright = True

            # L3: se ancora poco testo, aggiungi warning nel result
            # (mai ritornare errore — il poco testo è comunque utile)
            warning = None
            if len(text.strip()) < _PLAYWRIGHT_THRESHOLD:
                warning = (
                    f"Contenuto scarso ({len(text.strip())} chars) — "
                    "pagina potrebbe richiedere autenticazione o JavaScript avanzato."
                )

            result: dict = {
                "url":         url,
                "title":       meta["title"],
                "description": meta["description"],
                "content":     text,
                "status":      r.status_code,
                "chars":       len(text),
            }
            if used_playwright:
                result["extractor"] = "playwright"
            if warning:
                result["warning"] = warning
            # SAS: prioritizza link per intent semantico (P45-SAS)
            if _sas_available:
                try:
                    result["relevant_links"] = _semantic_nav.extract_and_score_links(
                        html_content=html,
                        base_url=url,
                        navigation_intent_keywords=navigation_intent_keywords,
                    )
                except Exception:
                    result["relevant_links"] = _extract_relevant_links(html, url)
            else:
                result["relevant_links"] = _extract_relevant_links(html, url)
            return result

    except httpx.TimeoutException:
        return {"url": url, "error": "Timeout", "content": ""}
    except Exception as e:
        return {"url": url, "error": str(e), "content": ""}