Spaces:
Running
Running
| """Web search via DuckDuckGo HTML endpoint (no API key, no new deps). | |
| Used as optional grounding for chat ("use web") and the local-deals feature. | |
| Degrades gracefully: any network/parse failure returns []. | |
| """ | |
| import html | |
| import logging | |
| import re | |
| from urllib.parse import parse_qs, unquote, urlparse | |
| logger = logging.getLogger(__name__) | |
| _SEARCH_URL = "https://html.duckduckgo.com/html/" | |
| _UA = ( | |
| "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " | |
| "(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36" | |
| ) | |
| # <a class="result__a" href="...">Title</a> | |
| _RESULT_RE = re.compile( | |
| r'<a[^>]*class="[^"]*result__a[^"]*"[^>]*href="([^"]+)"[^>]*>(.*?)</a>', | |
| re.IGNORECASE | re.DOTALL, | |
| ) | |
| # <a class="result__snippet" ...>snippet</a> (sometimes a <td>/<div>) | |
| _SNIPPET_RE = re.compile( | |
| r'<(?:a|td|div)[^>]*class="[^"]*result__snippet[^"]*"[^>]*>(.*?)</(?:a|td|div)>', | |
| re.IGNORECASE | re.DOTALL, | |
| ) | |
| _TAG_RE = re.compile(r"<[^>]+>") | |
| def _strip_html(s: str) -> str: | |
| return html.unescape(_TAG_RE.sub("", s)).strip() | |
| def _clean_url(href: str) -> str: | |
| """Resolve DDG redirect links //duckduckgo.com/l/?uddg=<urlencoded>.""" | |
| href = html.unescape(href) | |
| if href.startswith("//"): | |
| href = "https:" + href | |
| parsed = urlparse(href) | |
| if "duckduckgo.com" in parsed.netloc and parsed.path.startswith("/l"): | |
| uddg = parse_qs(parsed.query).get("uddg", [""])[0] | |
| if uddg: | |
| return unquote(uddg) | |
| return href | |
| def search_web(query: str, max_results: int = 5) -> list[dict]: | |
| """Search DuckDuckGo. Returns [{"title", "snippet", "url"}]; [] on failure.""" | |
| try: | |
| import httpx | |
| resp = httpx.get( | |
| _SEARCH_URL, | |
| params={"q": query}, | |
| headers={"User-Agent": _UA}, | |
| timeout=10.0, | |
| follow_redirects=True, | |
| ) | |
| resp.raise_for_status() | |
| page = resp.text | |
| links = _RESULT_RE.findall(page) | |
| snippets = [_strip_html(s) for s in _SNIPPET_RE.findall(page)] | |
| results = [] | |
| for i, (href, title_html) in enumerate(links[:max_results]): | |
| results.append( | |
| { | |
| "title": _strip_html(title_html), | |
| "snippet": snippets[i] if i < len(snippets) else "", | |
| "url": _clean_url(href), | |
| } | |
| ) | |
| return results | |
| except Exception as e: | |
| logger.warning("web search failed for %r: %s", query, e) | |
| return [] | |
| def format_results(results: list[dict]) -> str: | |
| """Readable text block for prompt injection.""" | |
| if not results: | |
| return "" | |
| lines = ["WEB SEARCH RESULTS:"] | |
| for i, r in enumerate(results, 1): | |
| lines.append(f"{i}. {r.get('title', '')}") | |
| if r.get("snippet"): | |
| lines.append(f" {r['snippet']}") | |
| if r.get("url"): | |
| lines.append(f" Source: {r['url']}") | |
| return "\n".join(lines) | |