adam/web_search.py

from __future__ import annotations

"""Small, keyless web-search adapter for conversational context.

Search results are deliberately treated as untrusted reference material.  This
module never opens result pages, downloads files, or performs actions other
than requesting the public search-results page.
"""

from dataclasses import dataclass
from html import unescape
from html.parser import HTMLParser
import ipaddress
import re
import socket
import urllib.error
import urllib.parse
import urllib.request
import xml.etree.ElementTree as ET


@dataclass(frozen=True, slots=True)
class SearchResult:
    title: str
    url: str
    snippet: str = ""


class WebSearchError(RuntimeError):
    pass


class WebReadError(RuntimeError):
    pass


class WebSearchClient:
    """Fetch a short list of public DuckDuckGo HTML search results."""

    SEARCH_URL = "https://www.bing.com/search?format=rss&q={query}"

    def search(self, query: str, *, limit: int = 5, timeout: float = 8.0) -> list[SearchResult]:
        query = " ".join(query.split())
        if not query:
            return []
        url = self.SEARCH_URL.format(query=urllib.parse.quote_plus(query))
        request = urllib.request.Request(
            url,
            headers={"User-Agent": "ADAM-WebSearch/1.0 (+local assistant)"},
        )
        try:
            with urllib.request.urlopen(request, timeout=timeout) as response:
                page = response.read().decode("utf-8", errors="replace")
        except (OSError, urllib.error.URLError, TimeoutError) as exc:
            raise WebSearchError(f"Web search is unavailable: {exc}") from exc

        results: list[SearchResult] = self._parse_bing_rss(page, limit)
        if results:
            return results
        return self._parse_duckduckgo_html(page, limit)

    @staticmethod
    def _parse_bing_rss(page: str, limit: int) -> list[SearchResult]:
        try:
            root = ET.fromstring(page)
        except ET.ParseError:
            return []
        results: list[SearchResult] = []
        for item in root.findall("./channel/item"):
            title = " ".join((item.findtext("title") or "").split())
            result_url = (item.findtext("link") or "").strip()
            snippet = " ".join((item.findtext("description") or "").split())
            if title and result_url.startswith(("https://", "http://")):
                results.append(SearchResult(title, result_url, snippet))
            if len(results) >= max(1, min(limit, 10)):
                break
        return results

    def _parse_duckduckgo_html(self, page: str, limit: int) -> list[SearchResult]:
        results: list[SearchResult] = []
        pattern = re.compile(
            r'<a[^>]+class="result__a"[^>]+href="(?P<url>[^"]+)"[^>]*>(?P<title>.*?)</a>'
            r'.{0,2500}?(?:<a[^>]+class="result__snippet"[^>]*>|<div[^>]+class="result__snippet"[^>]*>)(?P<snippet>.*?)</(?:a|div)>',
            re.IGNORECASE | re.DOTALL,
        )
        for match in pattern.finditer(page):
            title = self._clean_html(match.group("title"))
            result_url = unescape(match.group("url"))
            parsed = urllib.parse.urlparse(result_url)
            if parsed.path.startswith("/l/"):
                target = urllib.parse.parse_qs(parsed.query).get("uddg", [""])[0]
                result_url = urllib.parse.unquote(target)
            if title and result_url.startswith(("https://", "http://")):
                results.append(SearchResult(title, result_url, self._clean_html(match.group("snippet"))))
            if len(results) >= max(1, min(limit, 10)):
                break
        return results

    @staticmethod
    def _clean_html(value: str) -> str:
        return " ".join(unescape(re.sub(r"<[^>]+>", " ", value)).split())


@dataclass(frozen=True, slots=True)
class WebPage:
    url: str
    title: str
    text: str


class _ReadableTextParser(HTMLParser):
    SKIP_TAGS = {"script", "style", "noscript", "svg", "nav", "footer", "header"}

    def __init__(self) -> None:
        super().__init__()
        self.title = ""
        self._in_title = False
        self._skip_depth = 0
        self.parts: list[str] = []

    def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
        if tag in self.SKIP_TAGS:
            self._skip_depth += 1
        if tag == "title":
            self._in_title = True

    def handle_endtag(self, tag: str) -> None:
        if tag in self.SKIP_TAGS and self._skip_depth:
            self._skip_depth -= 1
        if tag == "title":
            self._in_title = False

    def handle_data(self, data: str) -> None:
        if self._skip_depth:
            return
        if self._in_title:
            self.title += data
        self.parts.append(data)


class _NoRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, request, fp, code, msg, headers, newurl):  # type: ignore[no-untyped-def]
        return None


class WebPageReader:
    """Fetch small readable extracts from explicitly requested public pages."""

    MAX_BYTES = 1_500_000
    MAX_REDIRECTS = 3

    def read(self, url: str, *, timeout: float = 10.0, max_characters: int = 6_000) -> WebPage:
        current_url = url
        opener = urllib.request.build_opener(_NoRedirect())
        for _ in range(self.MAX_REDIRECTS + 1):
            self._validate_public_url(current_url)
            request = urllib.request.Request(
                current_url,
                headers={"User-Agent": "ADAM-WebReader/1.0 (+local assistant)"},
            )
            try:
                response = opener.open(request, timeout=timeout)
            except urllib.error.HTTPError as exc:
                if exc.code in {301, 302, 303, 307, 308} and exc.headers.get("Location"):
                    current_url = urllib.parse.urljoin(current_url, exc.headers["Location"])
                    continue
                raise WebReadError(f"Could not read page (HTTP {exc.code}).") from exc
            except (OSError, urllib.error.URLError, TimeoutError) as exc:
                raise WebReadError(f"Could not read page: {exc}") from exc
            with response:
                content_type = response.headers.get_content_type()
                if content_type not in {"text/html", "text/plain"}:
                    raise WebReadError(f"This page is {content_type}, not readable web text.")
                raw = response.read(self.MAX_BYTES + 1)
                if len(raw) > self.MAX_BYTES:
                    raise WebReadError("This page is too large to read safely.")
                body = raw.decode(response.headers.get_content_charset() or "utf-8", errors="replace")
                if content_type == "text/plain":
                    text = " ".join(body.split())
                    return WebPage(current_url, current_url, text[:max_characters])
                parser = _ReadableTextParser()
                parser.feed(body)
                text = " ".join(parser.parts).strip()
                title = " ".join(parser.title.split()) or current_url
                return WebPage(current_url, title, text[:max_characters])
        raise WebReadError("Page redirected too many times.")

    @staticmethod
    def _validate_public_url(url: str) -> None:
        parsed = urllib.parse.urlparse(url)
        if parsed.scheme not in {"http", "https"} or not parsed.hostname:
            raise WebReadError("Only public http(s) links can be read.")
        try:
            addresses = socket.getaddrinfo(parsed.hostname, None, type=socket.SOCK_STREAM)
        except socket.gaierror as exc:
            raise WebReadError("Could not resolve that website.") from exc
        for item in addresses:
            address = ipaddress.ip_address(item[4][0])
            if not address.is_global:
                raise WebReadError("Private, local, and reserved network addresses are blocked.")


def should_search(request: str) -> bool:
    """Whether a chat request explicitly asks ADAM for fresh web information."""
    return bool(re.search(
        r"\b(?:search(?:\s+the)?\s+(?:web|internet|online)|web\s+search|"
        r"look\s+up|google|browse\s+(?:the\s+)?(?:web|internet)|"
        r"latest|current|today'?s|recent\s+news)\b",
        request,
        re.IGNORECASE,
    ))


def search_query_from_request(request: str) -> str:
    """Remove a conversational search command before sending the actual topic."""
    query = re.sub(
        r"^\s*(?:please\s+)?(?:search(?:\s+the)?\s+(?:web|internet|online)|"
        r"web\s+search|look\s+up|google|browse\s+(?:the\s+)?(?:web|internet))"
        r"\s+(?:for\s+)?",
        "",
        request,
        flags=re.IGNORECASE,
    )
    return query.strip(" \t\r\n?!.") or request


def should_read_links(request: str) -> bool:
    return bool(re.search(
        r"\b(?:read|open|research|study|summari[sz]e|go\s+further|learn\s+more|"
        r"look\s+into)\b.*\b(?:link|links|result|results|page|pages|web|website|site)\b"
        r"|\b(?:read|open)\s+https?://",
        request,
        re.IGNORECASE,
    ))


def urls_in_request(request: str) -> list[str]:
    return re.findall(r"https?://[^\s<>\]\[\)\}\"']+", request)


def format_search_context(results: list[SearchResult]) -> str:
    if not results:
        return "No usable web results were returned. Be transparent about that."
    lines = []
    for index, result in enumerate(results, 1):
        snippet = f" — {result.snippet}" if result.snippet else ""
        lines.append(f"[{index}] {result.title}{snippet}\nURL: {result.url}")
    return "\n".join(lines)


def format_page_context(pages: list[WebPage]) -> str:
    if not pages:
        return "No linked pages could be read."
    return "\n\n".join(
        f"PAGE: {page.title}\nURL: {page.url}\nEXTRACT: {page.text}"
        for page in pages
    )