Sub-app FastAPI montada en /web-search con búsqueda vía ddgs. OpenAPI 3.0.3 + servers absoluto + operationId webSearch listo para importar como external tool en wxO. Sin auth (uso de taller/demo). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
66 lines
1.7 KiB
Python
66 lines
1.7 KiB
Python
"""Ú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
|