taller-wox/app/web_search/api.py

"""Sub-app FastAPI con la web-search tool para watsonx Orchestrate.

Se monta en la app principal bajo `/web-search`, por lo que:
- `GET /web-search/search`      → endpoint de búsqueda (abierto, sin auth)
- `GET /web-search/health`      → health check (fuera del schema)
- `GET /web-search/openapi.json`→ spec OpenAPI 3.0.3 listo para importar en wxO
- `GET /web-search/docs`        → Swagger UI para probar
"""
from __future__ import annotations

from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException, Query, status
from fastapi.openapi.utils import get_openapi
from pydantic import BaseModel, Field

from app.web_search.config import WebSearchSettings, get_web_search_settings
from app.web_search.search import SearchProviderError, search_web


OPERATION_DESCRIPTION = (
    "Busca en la web y devuelve los resultados más relevantes (título, URL y "
    "resumen). Úsala cuando necesites información actual o reciente, noticias, "
    "datos que puedan haber cambiado, o cualquier información que no esté en tu "
    "conocimiento base."
)


class SearchResultModel(BaseModel):
    title: str = Field(..., description="Título del resultado de búsqueda.")
    url: str = Field(..., description="URL del resultado.")
    snippet: str = Field(
        ...,
        description="Fragmento de texto / resumen del resultado (texto plano, sin HTML, máx. 300 caracteres).",
    )


class SearchResponse(BaseModel):
    query: str = Field(..., description="Consulta original recibida.")
    count: int = Field(..., description="Cantidad de resultados devueltos.")
    results: list[SearchResultModel] = Field(
        ..., description="Lista de resultados, ordenados de mayor a menor relevancia."
    )


web_search_app = FastAPI(
    title="FIT Web Search Tool",
    version="1.0.0",
    description=(
        "Herramienta de búsqueda web para agentes de watsonx Orchestrate. "
        "Devuelve resultados normalizados desde DuckDuckGo."
    ),
    docs_url="/docs",
    redoc_url=None,
    openapi_url="/openapi.json",
)


@web_search_app.get(
    "/health",
    include_in_schema=False,
)
def health() -> dict[str, str]:
    return {"status": "ok"}


@web_search_app.get(
    "/search",
    operation_id="webSearch",
    summary="Buscar en la web",
    description=OPERATION_DESCRIPTION,
    response_model=SearchResponse,
    responses={
        422: {"description": "Parámetros de entrada inválidos."},
        502: {"description": "Error en el proveedor de búsqueda."},
    },
)
def search(
    settings: Annotated[WebSearchSettings, Depends(get_web_search_settings)],
    query: Annotated[
        str,
        Query(
            min_length=1,
            max_length=400,
            description="Consulta de búsqueda en lenguaje natural.",
            examples=["últimas noticias IBM watsonx orchestrate"],
        ),
    ],
    max_results: Annotated[
        int,
        Query(
            ge=1,
            le=10,
            description="Cantidad de resultados a devolver (1–10). Por defecto 5.",
        ),
    ] = 5,
    region: Annotated[
        str,
        Query(
            min_length=2,
            max_length=10,
            description=(
                "Región/idioma del motor de búsqueda. Ejemplos: 'wt-wt' (global), "
                "'cl-es' (Chile/español), 'mx-es' (México/español), 'us-en' (EEUU/inglés)."
            ),
            examples=["wt-wt", "cl-es", "mx-es", "us-en"],
        ),
    ] = "wt-wt",
) -> SearchResponse:
    try:
        results = search_web(
            query=query,
            max_results=max_results,
            region=region,
            timeout=settings.request_timeout,
        )
    except SearchProviderError:
        raise HTTPException(
            status_code=status.HTTP_502_BAD_GATEWAY,
            detail="Error en el proveedor de búsqueda",
        )

    return SearchResponse(
        query=query,
        count=len(results),
        results=[
            SearchResultModel(title=r.title, url=r.url, snippet=r.snippet) for r in results
        ],
    )


def _custom_openapi() -> dict:
    if web_search_app.openapi_schema:
        return web_search_app.openapi_schema

    settings = get_web_search_settings()
    schema = get_openapi(
        title=web_search_app.title,
        version=web_search_app.version,
        description=web_search_app.description,
        routes=web_search_app.routes,
    )

    # wxO prefiere 3.0.x (FastAPI emite 3.1.0 por defecto)
    schema["openapi"] = "3.0.3"

    # Bloque servers absoluto — obligatorio para wxO
    schema["servers"] = [{"url": settings.public_base_url}]

    web_search_app.openapi_schema = schema
    return schema


web_search_app.openapi = _custom_openapi  # type: ignore[assignment]