Source code for aidputils.agents.tools.experimental.websearch.base

from __future__ import annotations

from abc import ABC
from typing import Any

from aidputils.agents.tools.experimental.websearch.web_search_models import NormalizedWebResult


def _as_str(v: Any) -> str | None:
    if v is None:
        return None
    if isinstance(v, str):
        s = v.strip()
        return s or None
    return str(v).strip() or None




def normalize_generic_results(data: Any, source: str) -> list[NormalizedWebResult]:
    """
    Fallback normalization when we don't have an engine-specific parser.

    Accepts:
      - {"results": [ {title|name, url|link, snippet|description|content, ...}, ... ]}
      - [ { ... }, ... ]

    Returns a normalized list of NormalizedWebResult.
    """
    results = None
    if isinstance(data, dict):
        results = data.get("results")
    if results is None and isinstance(data, list):
        results = data

    if not isinstance(results, list):
        return []

    out: list[NormalizedWebResult] = []
    for r in results:
        if not isinstance(r, dict):
            continue
        out.append(
            NormalizedWebResult(
                title=_as_str(r.get("title") or r.get("name")),
                url=_as_str(r.get("url") or r.get("link")),
                snippet=_as_str(r.get("snippet") or r.get("description") or r.get("content")),
                source=_as_str(r.get("source") or source),
                published_time=_as_str(r.get("publishedTime") or r.get("published_time")),
            )
        )
    return out





class BaseWebSearchHandler(ABC):
    """
    Base class for engine-specific web search handling.

    Responsibilities:
      - Provide endpoint resolution (`get_endpoint`)
      - Parse engine-native responses into `NormalizedWebResult` list via `parse`

    Subclasses may override `parse()` to handle engine-native response shapes.
    If they do not, the base implementation normalizes common/generic result shapes.
    """

    def __init__(self, engine: str):
        self.engine = (engine or "").strip().upper()



    def get_endpoint(self, conf: dict) -> str:
        # Default: use endpoint from conf (proxy mode)
        return str(conf.get("endpoint") or "").strip()


    def _fallback(self, data: Any) -> list[NormalizedWebResult]:
        return normalize_generic_results(data, source=self.engine)



    def parse(self, data: Any) -> list[NormalizedWebResult]:
        return self._fallback(data)