Class WebSearcher

scripts/fetch_nmpa_fulltext.py:319–413

Unified search via Vertex AI Search (preferred) or Google CSE (legacy).

# ---------------------------------------------------------------------------

class WebSearcher:
    """Unified search via Vertex AI Search (preferred) or Google CSE (legacy)."""

    VERTEX_URL = (
        "https://discoveryengine.googleapis.com/v1/projects/{project_id}"
        "/locations/global/collections/default_collection"
        "/engines/{app_id}/servingConfigs/default_search:searchLite"
    )
    GOOGLE_CSE_URL = "https://www.googleapis.com/customsearch/v1"

    def __init__(self):
        self.use_vertex = bool(
            VERTEX_AI_PROJECT_ID and VERTEX_AI_SEARCH_APP_ID
            and VERTEX_AI_SEARCH_API_KEY
        )
        self.use_google = bool(GOOGLE_SEARCH_API_KEY and GOOGLE_SEARCH_ENGINE_ID)
        if self.use_vertex:
            logger.info("Using Vertex AI Search (searchLite)")
        elif self.use_google:
            logger.info("Using Google CSE (legacy)")
        else:
            logger.warning(
                "No search API configured. Set VERTEX_AI_* or GOOGLE_SEARCH_* env vars."
            )

    @property
    def available(self) -> bool:
        return self.use_vertex or self.use_google

    def search(self, query: str, num: int = 10) -> list[dict]:
        if self.use_vertex:
            return self._vertex_search(query, num)
        if self.use_google:
            return self._google_search(query, num)
        return []

    def _vertex_search(self, query: str, num: int) -> list[dict]:
        query = re.sub(r"site:\S+\s*", "", query).strip()
        query = re.sub(r"filetype:\S+\s*", "", query).strip()
        url = self.VERTEX_URL.format(
            project_id=VERTEX_AI_PROJECT_ID,
            app_id=VERTEX_AI_SEARCH_APP_ID,
        )
        try:
            resp = requests.post(
                url,
                params={"key": VERTEX_AI_SEARCH_API_KEY},
                json={"query": query, "pageSize": min(num, 25)},
                headers={"Content-Type": "application/json"},
                timeout=20,
            )
            if resp.status_code != 200:
                logger.warning(f"Vertex AI Search HTTP {resp.status_code}: "
                               f"{resp.text[:200]}")
                return []
            results = []
            for r in resp.json().get("results", []):
                doc = r.get("document", {})