scraper.py

import base64
from urllib.parse import quote, urlparse, parse_qs
import requests
import trafilatura
from playwright.sync_api import sync_playwright, TimeoutError as PlaywrightTimeoutError
from errors import get_logger, GenerAIError, ErrorCode, fmt_exc

log = get_logger("scraper")

SEARCH_URL = "https://www.bing.com/search"
USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36"
NAV_TIMEOUT_MS = 15000


def _launch_browser(p):
    return p.chromium.launch(
        headless=True,
        args=["--no-sandbox", "--disable-setuid-sandbox", "--disable-dev-shm-usage"],
    )


def _decode_bing_redirect(href: str) -> str:
    """Bing avvolge i link organici in redirect di tracking (bing.com/ck/a?...&u=<base64>)."""
    if "bing.com/ck/a" not in href:
        return href
    try:
        u = parse_qs(urlparse(href).query).get("u", [""])[0]
        if u.startswith("a1"):
            u = u[2:]
        pad = "=" * (-len(u) % 4)
        return base64.urlsafe_b64decode(u + pad).decode("utf-8", errors="ignore")
    except Exception:
        return href


def _search_links(page, query: str, max_results: int) -> list[dict]:
    """Cerca su Bing e restituisce [{title, url, snippet}]."""
    url = f"{SEARCH_URL}?q={quote(query)}&setlang=it"
    page.goto(url, timeout=NAV_TIMEOUT_MS, wait_until="domcontentloaded")

    hits = []
    rows = page.locator("li.b_algo").all()[: max_results * 2]
    for row in rows:
        try:
            link = row.locator("h2 a").first
            if not link.count():
                continue
            href = link.get_attribute("href")
            title = link.text_content() or ""
            if not href:
                continue
            snippet_el = row.locator("p, .b_lineclamp2, .b_lineclamp3, .b_lineclamp4").first
            snippet = snippet_el.text_content() if snippet_el.count() else ""
            real_url = _decode_bing_redirect(href)
            if not real_url.startswith("http"):
                continue
            hits.append({"title": title.strip(), "url": real_url, "snippet": (snippet or "").strip()})
        except Exception:
            continue
    return hits


def _extract_text(page, url: str, snippet: str) -> str | None:
    """Naviga alla pagina e ne estrae il testo pulito con trafilatura."""
    try:
        page.goto(url, timeout=NAV_TIMEOUT_MS, wait_until="domcontentloaded")
        html = page.content()
        text = trafilatura.extract(
            html,
            url=url,
            include_links=False,
            include_images=False,
            include_tables=False,
            no_fallback=False,
        )
    except PlaywrightTimeoutError:
        log.debug("Timeout caricamento pagina: %s", url)
        text = None
    except Exception as e:
        log.debug("Fetch fallito per %s — %s", url, fmt_exc(e))
        text = None

    if not text or len(text) < 80:
        text = snippet
    if not text or len(text) < 20:
        return None
    return text[:2000]


def _chromium_search(query: str, max_results: int) -> list[dict]:
    """Cerca ed estrae testo usando Chromium headless (Playwright) via Bing."""
    results: list[dict] = []
    try:
        with sync_playwright() as p:
            browser = _launch_browser(p)
            context = browser.new_context(user_agent=USER_AGENT, locale="it-IT")
            page = context.new_page()
            try:
                hits = _search_links(page, query, max_results)
            except PlaywrightTimeoutError:
                log.warning("[%s] Timeout ricerca Bing per: %r", ErrorCode.WEB_SEARCH_FAILED.value, query)
                hits = []
            except Exception as e:
                log.warning("[%s] Ricerca Bing fallita: %s", ErrorCode.WEB_SEARCH_FAILED.value, fmt_exc(e))
                hits = []

            for hit in hits:
                if len(results) >= max_results:
                    break
                log.debug("Fetching: %s", hit["url"])
                text = _extract_text(page, hit["url"], hit["snippet"])
                if not text:
                    log.debug("Testo troppo corto per: %s", hit["url"])
                    continue
                log.info("Estratti %d chars da: %s", len(text), hit["url"])
                results.append({"title": hit["title"], "url": hit["url"], "text": text})

            browser.close()
    except Exception as e:
        err = GenerAIError(ErrorCode.WEB_SEARCH_FAILED, f"Browser Chromium non avviato: {fmt_exc(e)}", cause=e)
        err.log(log)
        return []
    return results


def _wikipedia_search(query: str) -> list[dict]:
    """Fallback diretto su Wikipedia italiana + inglese."""
    results = []
    for lang in ("it", "en"):
        if len(results) >= 2:
            break
        try:
            api = f"https://{lang}.wikipedia.org/w/api.php"
            s = requests.get(api, params={
                "action": "query", "list": "search",
                "srsearch": query, "format": "json", "srlimit": 2,
            }, timeout=8, headers={"User-Agent": "GenerAI/3.0"})
            if s.status_code != 200:
                continue
            for hit in s.json().get("query", {}).get("search", []):
                title = hit["title"]
                p = requests.get(api, params={
                    "action": "query", "prop": "extracts",
                    "exintro": "1", "explaintext": "1",
                    "titles": title, "format": "json",
                }, timeout=8, headers={"User-Agent": "GenerAI/3.0"})
                if p.status_code != 200:
                    continue
                for page in p.json().get("query", {}).get("pages", {}).values():
                    extract = page.get("extract", "").strip()
                    if len(extract) > 50:
                        results.append({
                            "title": page["title"],
                            "url": f"https://{lang}.wikipedia.org/wiki/{page['title'].replace(' ', '_')}",
                            "text": extract[:2000],
                        })
                        break
        except Exception as e:
            log.debug("Wikipedia %s fallita: %s", lang, fmt_exc(e))
    if results:
        log.info("Wikipedia fallback → %d risultati", len(results))
    return results


def search_and_extract(query: str, max_results: int = 3) -> list[dict]:
    """Cerca sul web con Chromium (Playwright, Bing) + fallback Wikipedia. Estrae testo pulito."""
    log.info("Ricerca web (Chromium) per: %r", query)

    results = _chromium_search(query, max_results)

    if not results:
        log.info("Chromium non ha prodotto risultati — provo Wikipedia...")
        results = _wikipedia_search(query)

    if not results:
        log.warning("[%s] Nessun risultato per: %r", ErrorCode.WEB_NO_RESULTS.value, query)

    return results