"""Small, dependency-free web search helper for EIM Chat. Uses public HTML search endpoints (DuckDuckGo first, Bing RSS as a fallback). Search snippets are untrusted evidence; callers should never execute or follow instructions embedded in them. Network requests are bounded by timeout/size. """ from __future__ import annotations import html import re import urllib.error import urllib.parse import urllib.request import xml.etree.ElementTree as ET from html.parser import HTMLParser from typing import Any _USER_AGENT = "Mozilla/5.0 (compatible; EIMChat/1.0; +https://huggingface.co/)" _MAX_BYTES = 1_500_000 class _DuckDuckGoParser(HTMLParser): def __init__(self) -> None: super().__init__(convert_charrefs=True) self.results: list[dict[str, str]] = [] self.current: dict[str, str] | None = None self.mode = "" @staticmethod def _classes(attrs: list[tuple[str, str | None]]) -> set[str]: values = dict(attrs) return set((values.get("class") or "").split()) def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: classes = self._classes(attrs) attr_map = dict(attrs) if tag == "a" and "result__a" in classes: self._finish_current() url = _unwrap_search_url(attr_map.get("href", "")) if url: self.current = {"title": "", "url": url, "snippet": ""} self.mode = "title" elif tag in {"div", "a", "td"} and "result__snippet" in classes and self.current: self.mode = "snippet" def handle_endtag(self, tag: str) -> None: if tag == "a" and self.mode in {"title", "snippet"}: self.mode = "" elif tag in {"div", "td"} and self.mode == "snippet": self.mode = "" def handle_data(self, data: str) -> None: if not self.current or not self.mode: return value = re.sub(r"\s+", " ", data).strip() if value: self.current[self.mode] = (self.current.get(self.mode, "") + " " + value).strip() def _finish_current(self) -> None: if self.current and self.current.get("title"): self.results.append(self.current) self.current = None self.mode = "" def close(self) -> None: super().close() self._finish_current() def _unwrap_search_url(raw_url: str) -> str: raw_url = html.unescape((raw_url or "").strip()) if raw_url.startswith("//"): raw_url = "https:" + raw_url parsed = urllib.parse.urlsplit(raw_url) if parsed.netloc.endswith("duckduckgo.com") and parsed.path.startswith("/l/"): query = urllib.parse.parse_qs(parsed.query) raw_url = (query.get("uddg") or [""])[0] parsed = urllib.parse.urlsplit(raw_url) if parsed.scheme not in {"http", "https"} or not parsed.netloc: return "" if parsed.username or parsed.password: return "" return urllib.parse.urlunsplit((parsed.scheme, parsed.netloc, parsed.path, parsed.query, "")) def _clean(value: Any, limit: int = 600) -> str: value = html.unescape(re.sub(r"<[^>]*>", " ", str(value or ""))) return re.sub(r"\s+", " ", value).strip()[:limit] def parse_duckduckgo_html(source: str, limit: int = 5) -> list[dict[str, str]]: parser = _DuckDuckGoParser() parser.feed(source or "") unique: list[dict[str, str]] = [] seen: set[str] = set() for row in parser.results: url = _unwrap_search_url(row.get("url", "")) title = _clean(row.get("title"), 240) if not url or not title or url in seen: continue seen.add(url) unique.append({"title": title, "url": url, "snippet": _clean(row.get("snippet"), 700)}) if len(unique) >= max(1, min(10, int(limit))): break return unique def _fetch(url: str, timeout: float) -> bytes: request = urllib.request.Request(url, headers={"User-Agent": _USER_AGENT, "Accept": "text/html,application/rss+xml,application/xml;q=0.9,*/*;q=0.8"}) with urllib.request.urlopen(request, timeout=max(2.0, min(12.0, float(timeout)))) as response: data = response.read(_MAX_BYTES + 1) if len(data) > _MAX_BYTES: data = data[:_MAX_BYTES] return data def _bing_rss(query: str, limit: int, timeout: float) -> list[dict[str, str]]: url = "https://www.bing.com/search?format=rss&q=" + urllib.parse.quote_plus(query) data = _fetch(url, timeout) root = ET.fromstring(data) results: list[dict[str, str]] = [] seen: set[str] = set() for item in root.findall(".//item"): title = _clean(item.findtext("title"), 240) link = _unwrap_search_url(item.findtext("link", "")) snippet = _clean(item.findtext("description"), 700) if title and link and link not in seen: seen.add(link) results.append({"title": title, "url": link, "snippet": snippet}) if len(results) >= max(1, min(10, int(limit))): break return results def search_web(query: str, limit: int = 5, timeout: float | None = None) -> list[dict[str, str]]: """Search the public web without third-party Python dependencies. Raises RuntimeError only when both public endpoints fail; a valid empty result list means the endpoints responded but did not return readable hits. """ query = re.sub(r"\s+", " ", (query or "")).strip() if not query: return [] timeout = timeout if timeout is not None else float(__import__("os").environ.get("EIM_WEB_TIMEOUT", "6")) limit = max(1, min(10, int(limit))) errors: list[str] = [] try: url = "https://html.duckduckgo.com/html/?q=" + urllib.parse.quote_plus(query) results = parse_duckduckgo_html(_fetch(url, timeout).decode("utf-8", "replace"), limit) if results: return results errors.append("DuckDuckGo returned no readable results") except Exception as exc: errors.append(f"DuckDuckGo {type(exc).__name__}") try: results = _bing_rss(query, limit, timeout) if results: return results errors.append("Bing returned no readable results") except Exception as exc: errors.append(f"Bing {type(exc).__name__}") if errors and all("no readable results" not in e for e in errors): raise RuntimeError("Web search unavailable: " + "; ".join(errors)) return [] def should_search_web(query: str) -> bool: """Only trigger automatic lookup for explicit search/current-information intent.""" text = (query or "").casefold() patterns = ( r"\b(search the web|web search|browse the web|look up online|search online|from the internet|latest|current|today|yesterday|this week|breaking news|news today|weather|exchange rate|stock price)\b", r"(ابحث|ابحثلي|دورلي|دور لي|تصفح الويب|بحث الويب|بحث بالنت|من النت|على الانترنت|عبر الانترنت|آخر الأخبار|اخر الاخبار|الأخبار اليوم|الاخبار اليوم|اليوم|الآن|حالياً|حاليا|أحدث|احدث|السعر الحالي|سعر اليوم)", ) return any(re.search(pattern, text, re.I) for pattern in patterns)