File size: 6,879 Bytes
77c49b2 c5a7516 65de2c3 77c49b2 65de2c3 77c49b2 c5a7516 77c49b2 c5a7516 65de2c3 a0672e2 65de2c3 a0672e2 65de2c3 a0672e2 77c49b2 a0672e2 77c49b2 a0672e2 65de2c3 a0672e2 65de2c3 a0672e2 77c49b2 a0672e2 77c49b2 a0672e2 77c49b2 a0672e2 65de2c3 a0672e2 65de2c3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 | 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
|