Upload scraper.py with huggingface_hub

scraper.py

@@ -0,0 +1,74 @@
+import trafilatura
+from duckduckgo_search import DDGS
+from errors import get_logger, GenerAIError, ErrorCode, fmt_exc
+
+log = get_logger("scraper")
+
+
+def search_and_extract(query: str, max_results: int = 3) -> list[dict]:
+    """Search DuckDuckGo and extract clean text. Returns [] on total failure."""
+    log.info("Ricerca web per: %r", query)
+
+    # ── 1. DuckDuckGo search ───────────────────────────────────────────────────
+    try:
+        with DDGS() as ddgs:
+            hits = list(ddgs.text(query, max_results=max_results * 2))
+        log.debug("DuckDuckGo ha restituito %d risultati grezzi", len(hits))
+    except Exception as e:
+        err = GenerAIError(ErrorCode.WEB_SEARCH_FAILED, f"DuckDuckGo non raggiungibile: {fmt_exc(e)}", cause=e)
+        err.log(log)
+        return []
+
+    if not hits:
+        log.warning("[%s] Nessun risultato DuckDuckGo per: %r", ErrorCode.WEB_NO_RESULTS.value, query)
+        return []
+
+    # ── 2. Fetch + extract each URL ────────────────────────────────────────────
+    results = []
+    for hit in hits:
+        if len(results) >= max_results:
+            break
+
+        url = hit.get("href", "")
+        if not url:
+            log.debug("Hit senza URL, saltato: %s", hit)
+            continue
+
+        log.debug("Fetching: %s", url)
+        try:
+            downloaded = trafilatura.fetch_url(url, timeout=10)
+        except Exception as e:
+            log.warning("[%s] fetch_url fallito per %s — %s", ErrorCode.WEB_FETCH_FAILED.value, url, fmt_exc(e))
+            continue
+
+        if not downloaded:
+            log.debug("fetch_url ha restituito None per: %s", url)
+            continue
+
+        try:
+            text = trafilatura.extract(
+                downloaded,
+                include_links=False,
+                include_images=False,
+                include_tables=False,
+                no_fallback=False,
+            )
+        except Exception as e:
+            log.warning("[%s] estrazione testo fallita per %s — %s", ErrorCode.WEB_EXTRACT_FAILED.value, url, fmt_exc(e))
+            continue
+
+        if not text or len(text) < 150:
+            log.debug("Testo troppo corto (%d chars) per: %s", len(text) if text else 0, url)
+            continue
+
+        log.info("Estratti %d chars da: %s", len(text), url)
+        results.append({
+            "title": hit.get("title", ""),
+            "url": url,
+            "text": text[:2000],
+        })
+
+    if not results:
+        log.warning("[%s] Nessun testo utile estratto per: %r", ErrorCode.WEB_NO_RESULTS.value, query)
+
+    return results