"""Único punto del servicio que toca ddgs. Si se cambia el backend (Brave, Google CSE, etc.) sólo se reescribe este módulo; el contrato `search_web(...)` -> list[SearchResult] no cambia. """ from __future__ import annotations import html import re from dataclasses import dataclass from ddgs import DDGS SNIPPET_MAX_LEN = 300 _TAG_RE = re.compile(r"<[^>]+>") _WS_RE = re.compile(r"\s+") class SearchProviderError(RuntimeError): """Error opaco del proveedor de búsqueda. main.py lo traduce a HTTP 502.""" @dataclass(frozen=True) class SearchResult: title: str url: str snippet: str def _clean_snippet(raw: str | None) -> str: if not raw: return "" text = _TAG_RE.sub("", raw) text = html.unescape(text) text = _WS_RE.sub(" ", text).strip() if len(text) > SNIPPET_MAX_LEN: text = text[: SNIPPET_MAX_LEN - 1].rstrip() + "…" return text def search_web(query: str, max_results: int, region: str, timeout: int) -> list[SearchResult]: try: raw = DDGS(timeout=timeout).text( query, region=region, max_results=max_results, ) except Exception as exc: # ddgs lanza varias excepciones distintas; las unificamos raise SearchProviderError(str(exc)) from exc results: list[SearchResult] = [] for item in raw or []: title = (item.get("title") or "").strip() url = (item.get("href") or item.get("url") or "").strip() if not url: continue results.append( SearchResult( title=title or url, url=url, snippet=_clean_snippet(item.get("body") or item.get("snippet")), ) ) return results