Fix: DuckDuckGo bloccava le richieste automatizzate, sostituito con Bing (testato e funzionante)

scraper.py

@@ -1,11 +1,13 @@
-from urllib.parse import quote
+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://html.duckduckgo.com/html/"
+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
 
@@ -17,23 +19,42 @@ def _launch_browser(p):
     )
 
 
+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 DuckDuckGo (HTML, no JS) e restituisce [{title, url, snippet}]."""
-    url = f"{SEARCH_URL}?q={quote(query)}&kl=it-it"
+    """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(".result").all()[: max_results * 2]
+    rows = page.locator("li.b_algo").all()[: max_results * 2]
     for row in rows:
         try:
-            link = row.locator("a.result__a").first
+            link = row.locator("h2 a").first
+            if not link.count():
+                continue
             href = link.get_attribute("href")
-            title = link.inner_text().strip()
-            if not href or not href.startswith("http"):
+            title = link.text_content() or ""
+            if not href:
                 continue
-            snippet_el = row.locator(".result__snippet").first
-            snippet = snippet_el.inner_text().strip() if snippet_el.count() else ""
-            hits.append({"title": title, "url": href, "snippet": snippet})
+            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
@@ -53,49 +74,36 @@ def _extract_text(page, url: str, snippet: str) -> str | None:
             no_fallback=False,
         )
     except PlaywrightTimeoutError:
-        log.warning("[%s] Timeout caricamento pagina: %s", ErrorCode.WEB_FETCH_FAILED.value, url)
+        log.debug("Timeout caricamento pagina: %s", url)
         text = None
     except Exception as e:
-        log.warning("[%s] Fetch fallito per %s — %s", ErrorCode.WEB_FETCH_FAILED.value, url, fmt_exc(e))
+        log.debug("Fetch fallito per %s — %s", url, fmt_exc(e))
         text = None
 
-    # Fallback allo snippet del motore di ricerca se l'estrazione non basta
-    if not text or len(text) < 100:
+    if not text or len(text) < 80:
         text = snippet
-
-    if not text or len(text) < 50:
+    if not text or len(text) < 20:
         return None
     return text[:2000]
 
 
-def search_and_extract(query: str, max_results: int = 3) -> list[dict]:
-    """Cerca sul web tramite Chromium (Playwright) ed estrae testo pulito.
-
-    Ritorna [] se la ricerca fallisce completamente. Interfaccia invariata
-    rispetto alla precedente implementazione basata su SearXNG.
-    """
-    log.info("Ricerca web (Chromium) per: %r", query)
-
+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)
+            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 DuckDuckGo per: %r", ErrorCode.WEB_SEARCH_FAILED.value, query)
+                log.warning("[%s] Timeout ricerca Bing per: %r", ErrorCode.WEB_SEARCH_FAILED.value, query)
                 hits = []
             except Exception as e:
-                log.warning("[%s] Ricerca DuckDuckGo fallita: %s", ErrorCode.WEB_SEARCH_FAILED.value, fmt_exc(e))
+                log.warning("[%s] Ricerca Bing fallita: %s", ErrorCode.WEB_SEARCH_FAILED.value, fmt_exc(e))
                 hits = []
 
-            if not hits:
-                browser.close()
-                log.warning("[%s] Nessun risultato per: %r", ErrorCode.WEB_NO_RESULTS.value, query)
-                return []
-
-            results = []
             for hit in hits:
                 if len(results) >= max_results:
                     break
@@ -112,8 +120,59 @@ def search_and_extract(query: str, max_results: int = 3) -> list[dict]:
         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 testo utile estratto per: %r", ErrorCode.WEB_NO_RESULTS.value, query)
+        log.warning("[%s] Nessun risultato per: %r", ErrorCode.WEB_NO_RESULTS.value, query)
 
     return results