| """Strumenti web di Neuronal: cercare su internet e leggere una pagina. |
| Lo lancia il motore come programma a parte, con i parametri scritti in esadecimale (niente problemi di virgolette): |
| python web_tool.py cerca <hex della domanda> [risultati] |
| python web_tool.py leggi <hex dell'indirizzo> [caratteri] |
| Scrive su stdout il testo da dare al modello e esce con 0 se ha trovato qualcosa, 1 altrimenti. Solo libreria standard. |
| |
| Sicurezza: legge solo http e https, mai indirizzi interni (localhost, rete di casa, servizi dei provider di cloud): |
| si controlla l'indirizzo vero a cui ci si collega e ogni passaggio di un reindirizzamento. Pagine e risposte hanno un tetto |
| di dimensione e di tempo. Il testo delle pagine è materiale da leggere, non istruzioni: il motore lo dice al modello. |
| |
| Motori di ricerca, nell'ordine (il primo che risponde): un SearXNG tuo (NEURONAL_SEARXNG=https://...), DuckDuckGo, |
| Bing, Wikipedia. Nessuna chiave o registrazione.""" |
| import base64, html, http.client, ipaddress, json, os, re, socket, ssl, sys, time, urllib.parse |
| from html.parser import HTMLParser |
|
|
| sys.stdout.reconfigure(encoding="utf-8") |
| UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/140.0 Safari/537.36 Neuronal" |
| MAX_BYTES = 2_500_000 |
| TOTALE = 24.0 |
| T0 = time.time() |
|
|
|
|
| def tempo(minimo=2.0): |
| return max(minimo, TOTALE - (time.time() - T0)) |
|
|
|
|
| |
|
|
| class Vietato(Exception): |
| pass |
|
|
|
|
| def indirizzo_ok(ip): |
| a = ipaddress.ip_address(ip) |
| if a.version == 6 and a.ipv4_mapped: |
| a = a.ipv4_mapped |
| return not (a.is_private or a.is_loopback or a.is_link_local or a.is_multicast or a.is_reserved or a.is_unspecified |
| or str(a).startswith("100.64.") or str(a) == "169.254.169.254") |
|
|
|
|
| def risolvi(host, porta): |
| try: |
| info = socket.getaddrinfo(host, porta, type=socket.SOCK_STREAM) |
| except socket.gaierror: |
| raise Vietato("indirizzo non trovato") |
| ok = [i for i in info if indirizzo_ok(i[4][0])] |
| if not ok or len(ok) != len(info): |
| raise Vietato("indirizzo interno o non pubblico: non lo apro") |
| return ok[0] |
|
|
|
|
| def apri(url, dati=None, intestazioni=None, passaggi=4): |
| """Scarica un indirizzo (GET, o POST se ci sono dati) seguendo i reindirizzamenti con gli stessi controlli. Ritorna (stato, intestazioni, byte, url finale).""" |
| for _ in range(passaggi + 1): |
| p = urllib.parse.urlsplit(url) |
| if p.scheme not in ("http", "https") or not p.hostname: |
| raise Vietato("solo indirizzi http e https") |
| porta = p.port or (443 if p.scheme == "https" else 80) |
| fam, tipo, proto, _, sa = risolvi(p.hostname, porta) |
| s = socket.socket(fam, tipo, proto) |
| s.settimeout(min(10, tempo())) |
| s.connect(sa) |
| if p.scheme == "https": |
| s = ssl.create_default_context().wrap_socket(s, server_hostname=p.hostname) |
| conn = http.client.HTTPConnection(p.hostname, porta, timeout=min(15, tempo())) |
| conn.sock = s |
| path = (p.path or "/") + (("?" + p.query) if p.query else "") |
| h = {"User-Agent": UA, "Accept": "text/html,application/xhtml+xml,application/json,text/plain;q=0.9,*/*;q=0.5", |
| "Accept-Language": "it-IT,it;q=0.9,en;q=0.6", "Accept-Encoding": "identity", "Connection": "close"} |
| h.update(intestazioni or {}) |
| corpo = None |
| if dati is not None: |
| corpo = urllib.parse.urlencode(dati).encode() |
| h["Content-Type"] = "application/x-www-form-urlencoded" |
| conn.request("POST" if corpo else "GET", path, body=corpo, headers=h) |
| r = conn.getresponse() |
| if r.status in (301, 302, 303, 307, 308) and r.getheader("Location"): |
| url = urllib.parse.urljoin(url, r.getheader("Location")) |
| dati = None if r.status in (301, 302, 303) else dati |
| conn.close() |
| continue |
| buf = bytearray() |
| while len(buf) < MAX_BYTES and time.time() - T0 < TOTALE: |
| pezzo = r.read(65536) |
| if not pezzo: |
| break |
| buf += pezzo |
| intest = {k.lower(): v for k, v in r.getheaders()} |
| conn.close() |
| return r.status, intest, bytes(buf), url |
| raise Vietato("troppi reindirizzamenti") |
|
|
|
|
| def decodifica(byte, intest): |
| cs = re.search(r"charset=([\w-]+)", intest.get("content-type", ""), re.I) |
| if not cs: |
| m = re.search(rb"<meta[^>]+charset=[\"']?([\w-]+)", byte[:4096], re.I) |
| cs = m and m.group(1).decode("ascii", "ignore") |
| else: |
| cs = cs.group(1) |
| for enc in (cs, "utf-8", "latin-1"): |
| try: |
| return byte.decode(enc or "utf-8") |
| except (UnicodeDecodeError, LookupError): |
| continue |
| return byte.decode("utf-8", "replace") |
|
|
|
|
| |
|
|
| class Testo(HTMLParser): |
| SALTA = {"script", "style", "noscript", "svg", "template", "iframe", "head", "nav", "footer", "form", "aside", "button"} |
| A_CAPO = {"p", "div", "br", "li", "tr", "h1", "h2", "h3", "h4", "h5", "h6", "section", "article", "blockquote", "pre", "table", "ul", "ol", "dt", "dd"} |
|
|
| def __init__(self): |
| super().__init__(convert_charrefs=True) |
| self.out, self.salta, self.titolo, self.in_titolo = [], 0, "", False |
|
|
| def handle_starttag(self, tag, attrs): |
| if tag == "title": |
| self.in_titolo = True |
| if tag in self.SALTA: |
| self.salta += 1 |
| if tag in self.A_CAPO and not self.salta: |
| self.out.append("\n") |
| if tag in ("h1", "h2", "h3") and not self.salta: |
| self.out.append("## ") |
| if tag == "li" and not self.salta: |
| self.out.append("- ") |
|
|
| def handle_endtag(self, tag): |
| if tag == "title": |
| self.in_titolo = False |
| if tag in self.SALTA and self.salta: |
| self.salta -= 1 |
| if tag in self.A_CAPO and not self.salta: |
| self.out.append("\n") |
|
|
| def handle_data(self, d): |
| if self.in_titolo: |
| self.titolo += d |
| elif not self.salta: |
| self.out.append(d) |
|
|
|
|
| def html_a_testo(testo): |
| p = Testo() |
| try: |
| p.feed(testo) |
| except Exception: |
| pass |
| t = "".join(p.out) |
| t = re.sub(r"[ \t\r\f\v]+", " ", t) |
| t = re.sub(r" ?\n ?", "\n", t) |
| t = re.sub(r"\n{3,}", "\n\n", t).strip() |
| return p.titolo.strip(), t |
|
|
|
|
| |
|
|
| def leggi(url, massimo): |
| if not re.match(r"^[a-z][a-z0-9+.-]*://", url, re.I): |
| url = "https://" + url |
| stato, intest, byte, finale = apri(url) |
| tipo = intest.get("content-type", "").split(";")[0].strip().lower() |
| if stato >= 400: |
| print(f"La pagina ha risposto con errore {stato}.") |
| return 1 |
| if tipo == "application/pdf" or byte[:5] == b"%PDF-": |
| print("È un PDF: da qui non lo leggo. Scarica il file e allegalo alla chat, poi lo leggo da lì.") |
| return 1 |
| if tipo.startswith("image/") or tipo.startswith(("audio/", "video/")) or tipo == "application/octet-stream": |
| print(f"È un file di tipo {tipo or 'sconosciuto'}, non una pagina di testo.") |
| return 1 |
| testo = decodifica(byte, intest) |
| if "html" in tipo or "xml" in tipo or testo.lstrip()[:15].lower().startswith(("<!doctype", "<html")): |
| titolo, corpo = html_a_testo(testo) |
| elif "json" in tipo: |
| titolo = "" |
| try: |
| corpo = json.dumps(json.loads(testo), ensure_ascii=False, indent=1) |
| except ValueError: |
| corpo = testo |
| else: |
| titolo, corpo = "", testo |
| if len(corpo) < 40: |
| print("La pagina non ha testo leggibile (forse si costruisce con JavaScript).") |
| return 1 |
| taglio = len(corpo) > massimo |
| corpo = corpo[:massimo] |
| print(f"Pagina: {finale}") |
| if titolo: |
| print(f"Titolo: {titolo}") |
| print("---") |
| print(corpo) |
| if taglio: |
| print(f"\n… (pagina più lunga: letti i primi {massimo} caratteri)") |
| return 0 |
|
|
|
|
| |
|
|
| def pulisci(t): |
| return re.sub(r"\s+", " ", html.unescape(re.sub(r"<[^>]+>", "", t or ""))).strip() |
|
|
|
|
| def cerca_searxng(q, n): |
| base = os.environ.get("NEURONAL_SEARXNG", "").rstrip("/") |
| if not base: |
| return [] |
| st, _, b, _ = apri(base + "/search?" + urllib.parse.urlencode({"q": q, "format": "json", "language": "it"})) |
| if st != 200: |
| return [] |
| return [(r.get("title", ""), r.get("url", ""), r.get("content", "")) for r in json.loads(b).get("results", [])[:n]] |
|
|
|
|
| def cerca_ddg(q, n): |
| st, _, b, _ = apri("https://html.duckduckgo.com/html/", dati={"q": q, "kl": "it-it"}) |
| t = b.decode("utf-8", "replace") |
| if st != 200 or "result__a" not in t: |
| return [] |
| out = [] |
| for blocco in re.split(r'<div class="result ', t)[1:]: |
| m = re.search(r'class="result__a"[^>]*href="([^"]+)"[^>]*>(.*?)</a>', blocco, re.S) |
| if not m: |
| continue |
| url = html.unescape(m.group(1)) |
| if "uddg=" in url: |
| url = urllib.parse.unquote(urllib.parse.parse_qs(urllib.parse.urlsplit(url.replace("//duckduckgo.com", "https://duckduckgo.com", 1) if url.startswith("//") else url).query).get("uddg", [url])[0]) |
| s = re.search(r'class="result__snippet"[^>]*>(.*?)</a>', blocco, re.S) |
| if "duckduckgo.com/y.js" in url: |
| continue |
| out.append((pulisci(m.group(2)), url, pulisci(s.group(1) if s else ""))) |
| if len(out) >= n: |
| break |
| return out |
|
|
|
|
| def cerca_bing(q, n): |
| st, _, b, _ = apri("https://www.bing.com/search?" + urllib.parse.urlencode({"q": q, "setlang": "it", "cc": "IT"})) |
| t = b.decode("utf-8", "replace") |
| if st != 200: |
| return [] |
| out = [] |
| for blocco in re.findall(r'<li class="b_algo".*?</li>', t, re.S): |
| m = re.search(r'<h2[^>]*><a[^>]+href="([^"]+)"[^>]*>(.*?)</a>', blocco, re.S) |
| if not m: |
| continue |
| url = html.unescape(m.group(1)) |
| if "bing.com/ck/a" in url: |
| u = urllib.parse.parse_qs(urllib.parse.urlsplit(url).query).get("u", [""])[0] |
| if u.startswith("a1"): |
| try: |
| url = base64.urlsafe_b64decode(u[2:] + "=" * (-len(u[2:]) % 4)).decode() |
| except Exception: |
| pass |
| s = re.search(r"<p[^>]*>(.*?)</p>", blocco, re.S) |
| out.append((pulisci(m.group(2)), url, pulisci(s.group(1) if s else ""))) |
| if len(out) >= n: |
| break |
| return out |
|
|
|
|
| def cerca_wikipedia(q, n): |
| out = [] |
| for lingua in ("it", "en"): |
| st, _, b, _ = apri(f"https://{lingua}.wikipedia.org/w/api.php?" + urllib.parse.urlencode( |
| {"action": "query", "list": "search", "srsearch": q, "format": "json", "srlimit": n, "utf8": 1})) |
| if st != 200: |
| continue |
| for r in json.loads(b)["query"]["search"]: |
| titolo = r["title"] |
| out.append((titolo + f" (Wikipedia {lingua})", f"https://{lingua}.wikipedia.org/wiki/" + urllib.parse.quote(titolo.replace(" ", "_")), |
| pulisci(r.get("snippet", "")))) |
| if out: |
| break |
| return out[:n] |
|
|
|
|
| def cerca(q, n): |
| motori = [("SearXNG", cerca_searxng), ("DuckDuckGo", cerca_ddg), ("Bing", cerca_bing), ("Wikipedia", cerca_wikipedia)] |
| errori = [] |
| for nome, fn in motori: |
| if time.time() - T0 > TOTALE - 3: |
| break |
| try: |
| r = fn(q, n) |
| except Exception as e: |
| errori.append(f"{nome}: {type(e).__name__}") |
| continue |
| r = [x for x in r if x[1].startswith("http")] |
| if r: |
| print(f"Risultati per «{q}» ({nome}):") |
| for i, (t, u, s) in enumerate(r, 1): |
| print(f"{i}. {t}\n {u}\n {s[:300]}") |
| print("\nPer leggere una pagina per intero usa leggi_pagina con il suo indirizzo.") |
| return 0 |
| print(f"Nessun risultato per «{q}»." + (f" (motori non raggiungibili: {', '.join(errori)})" if errori else "")) |
| return 1 |
|
|
|
|
| def main(): |
| if len(sys.argv) < 3 or sys.argv[1] not in ("cerca", "leggi"): |
| print("uso: web_tool.py cerca|leggi <hex> [numero]") |
| return 2 |
| try: |
| arg = bytes.fromhex(sys.argv[2]).decode("utf-8").strip() |
| except ValueError: |
| print("parametro non valido") |
| return 2 |
| extra = int(sys.argv[3]) if len(sys.argv) > 3 and sys.argv[3].isdigit() else None |
| if not arg: |
| print("manca il testo da cercare" if sys.argv[1] == "cerca" else "manca l'indirizzo") |
| return 2 |
| try: |
| if sys.argv[1] == "cerca": |
| return cerca(arg[:300], max(1, min(extra or 6, 10))) |
| return leggi(arg[:2000], max(500, min(extra or 7000, 20000))) |
| except Vietato as e: |
| print(f"Non posso aprirlo: {e}.") |
| return 1 |
| except (socket.timeout, TimeoutError): |
| print("La pagina non ha risposto in tempo.") |
| return 1 |
| except (OSError, http.client.HTTPException, ssl.SSLError) as e: |
| print(f"Non riesco a collegarmi ({type(e).__name__}).") |
| return 1 |
|
|
|
|
| if __name__ == "__main__": |
| sys.exit(main()) |
|
|
