Spaces:
Runtime error
Runtime error
File size: 1,830 Bytes
9e93b10 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 | from __future__ import annotations
import logging
from rexpro_ai.retrieval.web.main import SearchResult, get_filtered_results
from rexpro_ai.utils.session_pool import get_session
log = logging.getLogger(__name__)
async def search_google_pse(
api_key: str,
search_engine_id: str,
query: str,
count: int,
filter_list: list[str | None] | None = None,
referer: str | None = None,
) -> list[SearchResult]:
"""Query Google Programmable Search Engine with automatic pagination.
The PSE API returns at most 10 results per request, so this function
issues multiple requests when ``count > 10``.
"""
url = 'https://www.googleapis.com/customsearch/v1'
headers: dict[str, str] = {'Content-Type': 'application/json'}
if referer:
headers['Referer'] = referer
all_items: list[dict] = []
start_index = 1 # PSE uses 1-based pagination
session = await get_session()
remaining = count
while remaining > 0:
page_size = min(remaining, 10)
params = {
'cx': search_engine_id,
'q': query,
'key': api_key,
'num': str(page_size),
'start': str(start_index),
}
async with session.get(url, headers=headers, params=params) as response:
response.raise_for_status()
payload = await response.json()
items = payload.get('items', [])
if not items:
break
all_items.extend(items)
remaining -= len(items)
start_index += 10
if filter_list:
all_items = get_filtered_results(all_items, filter_list)
return [
SearchResult(
link=item.get('link', ''),
title=item.get('title'),
snippet=item.get('snippet'),
)
for item in all_items
]
|