| from typing import List | |
| import re | |
| import httpx | |
| from .base import fetch_json, normalize_result | |
| async def search_serpapi(query: str, limit: int = 20, api_key: str = "", **kwargs) -> List[dict]: | |
| if not api_key: | |
| return [] | |
| try: | |
| params = {"engine": "google_scholar", "q": query, "num": min(limit, 20), "api_key": api_key} | |
| data = await fetch_json("https://serpapi.com/search.json", params=params) | |
| if "error" in data: | |
| return [] | |
| results = [] | |
| for item in data.get("organic_results", []): | |
| title = item.get("title", "") | |
| snippet = item.get("snippet", "") | |
| link = item.get("link", "") | |
| year = None | |
| if item.get("publication_info", {}).get("summary"): | |
| year_match = item["publication_info"]["summary"] | |
| y = re.search(r'(\d{4})', year_match) | |
| if y: | |
| year = int(y.group(1)) | |
| results.append(normalize_result( | |
| title=title, | |
| authors=[], | |
| year=year, | |
| abstract=snippet, | |
| pdf_url=link if "pdf" in link.lower() else "", | |
| source="Google Scholar", | |
| )) | |
| return results[:limit] | |
| except Exception: | |
| return [] | |