BonusLockSMith/web-research-agent
0
1#!/usr/bin/env python2"""Robust page fetch + readable-text extraction for the Web Research Agent (#13).3 4The real web is messy, so this handles what a demo skips:5 - resolves redirects (Google News RSS hands back news.google.com redirect links, not the article)6 - SSRF guard (refuses internal/loopback/private hosts — safe for a public demo)7 - trafilatura main-content extraction, with a BeautifulSoup fallback8 - citation-marker / whitespace cleanup9 10Reused patterns: text-summarizer's trafilatura→bs4 fetch, companion's SSRF guard.11"""12import ipaddress13import re14import socket15import urllib.parse16 17import requests18from bs4 import BeautifulSoup19 20_UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "21 "(KHTML, like Gecko) Chrome/124.0 Safari/537.36")22TIMEOUT = 2023_CITE = re.compile(r"\[\d+\]|\[citation needed\]|\[edit\]", re.I)24 25 26def _blocked_host(url: str) -> bool:27 """True if the URL points at a non-public address (SSRF guard)."""28 try:29 host = urllib.parse.urlparse(url).hostname30 if not host:31 return True32 if host.lower() in ("localhost",) or host.endswith(".local"):33 return True34 # resolve and check every A record35 for fam, _, _, _, sockaddr in socket.getaddrinfo(host, None):36 ip = ipaddress.ip_address(sockaddr[0])37 if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved:38 return True39 return False40 except Exception:41 return True # fail closed42 43 44def resolve_url(url: str) -> str:45 """Follow redirects to the real destination. Google News RSS links (news.google.com/rss/...)46 bounce to the actual article — we want that final URL for clean extraction + honest citations."""47 try:48 r = requests.get(url, headers={"User-Agent": _UA}, timeout=TIMEOUT, allow_redirects=True)49 final = r.url50 # Google News sometimes lands on an interstitial with the real link in a <a> or data-attr.51 if "news.google.com" in final:52 m = re.search(r'data-n-au="(https?://[^"]+)"', r.text) or \53 re.search(r'<a[^>]+href="(https?://(?!news\.google\.com)[^"]+)"', r.text)54 if m:55 return m.group(1)56 return final57 except Exception:58 return url59 60 61def _clean(text: str) -> str:62 text = _CITE.sub("", text)63 lines = [ln.strip() for ln in text.splitlines()]64 return "\n".join(ln for ln in lines if ln)65 66 67def fetch(url: str, max_chars: int = 6000) -> dict:68 """Fetch a URL and return {url, final_url, title, text, error}. text is readable main content."""69 final = resolve_url(url)70 if _blocked_host(final):71 return {"url": url, "final_url": final, "title": "", "text": "", "error": "blocked (non-public host)"}72 try:73 r = requests.get(final, headers={"User-Agent": _UA}, timeout=TIMEOUT)74 doc = r.text75 except Exception as e:76 return {"url": url, "final_url": final, "title": "", "text": "", "error": f"fetch failed ({type(e).__name__})"}77 78 title, text = "", ""79 try:80 import trafilatura81 text = trafilatura.extract(doc, include_comments=False, include_tables=False,82 favor_precision=True) or ""83 md = trafilatura.extract_metadata(doc)84 if md and md.title:85 title = md.title86 except Exception:87 text = ""88 89 soup = None90 if len(text) < 200: # fallback: strip chrome, take main/article91 try:92 soup = BeautifulSoup(doc, "html.parser")93 for tag in soup(["script", "style", "nav", "header", "footer", "aside", "noscript", "form"]):94 tag.decompose()95 main = (soup.find("main") or soup.find("article")96 or soup.find(id="mw-content-text") or soup.body or soup)97 text = main.get_text("\n")98 except Exception:99 pass100 if not title:101 try:102 soup = soup or BeautifulSoup(doc, "html.parser")103 if soup.title and soup.title.string:104 title = soup.title.string.strip()105 except Exception:106 pass107 108 text = _clean(text)109 if len(text) < 120:110 return {"url": url, "final_url": final, "title": title, "text": "",111 "error": "could not extract readable text (JS-only / paywall / PDF)"}112 return {"url": url, "final_url": final, "title": title[:200], "text": text[:max_chars], "error": None}113 114 115if __name__ == "__main__":116 import sys117 from search import search118 q = " ".join(sys.argv[1:]) or "retrieval augmented generation"119 print(f"search: {q}")120 for res in search(q, 3):121 f = fetch(res["url"], max_chars=500)122 status = f["error"] or f"OK · {len(f['text'])} chars"123 print(f"\n[{res['source']}] {res['title'][:70]}")124 print(f" → final: {f['final_url'][:90]}")125 print(f" → {status}")126 if f["text"]:127 print(f" → {f['text'][:200].replace(chr(10),' ')}...")128 