my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

#1	"""SearchAPI Client for Real-Time Web Search"""
#2
#3	import httpx
#4	from typing import Optional, Dict, Any, List
#5	from dataclasses import dataclass
#6
#7
#8	SEARCHAPI_BASE_URL = "https://www.searchapi.io/api/v1"
#9
#10
#11	@dataclass
#12	class SearchResult:
#13	"""Single search result"""
#14	position: int
#15	title: str
#16	link: str
#17	snippet: str
#18	source: Optional[str] = None
#19	date: Optional[str] = None
#20
#21
#22	@dataclass
#23	class SearchResponse:
#24	"""Complete search response"""
#25	query: str
#26	total_results: int
#27	results: List[SearchResult]
#28	ai_overview: Optional[str] = None
#29	knowledge_graph: Optional[Dict[str, Any]] = None
#30	answer_box: Optional[Dict[str, Any]] = None
#31	related_searches: Optional[List[str]] = None
#32
#33
#34	class SearchAPIClient:
#35	"""Client for SearchAPI real-time web search"""
#36
#37	def __init__(self, api_key: str):
#38	"""
#39	Initialize SearchAPI client.
#40
#41	Args:
#42	api_key: SearchAPI API key
#43	"""
#44	self.api_key = api_key
#45	self.base_url = SEARCHAPI_BASE_URL
#46
#47	self._client = httpx.AsyncClient(
#48	timeout=30.0
#49	)
#50
#51	async def search(
#52	self,
#53	query: str,
#54	location: str = "United States",
#55	num_results: int = 10,
#56	language: str = "en",
#57	) -> SearchResponse:
#58	"""
#59	Perform a Google search via SearchAPI.
#60
#61	Args:
#62	query: Search query
#63	location: Geographic location for search
#64	num_results: Number of results to return
#65	language: Interface language
#66
#67	Returns:
#68	SearchResponse with results and metadata
#69	"""
#70	params = {
#71	"engine": "google",
#72	"q": query,
#73	"api_key": self.api_key,
#74	"location": location,
#75	"hl": language,
#76	"num": num_results,
#77	}
#78
#79	response = await self._client.get(f"{self.base_url}/search", params=params)
#80	response.raise_for_status()
#81	data = response.json()
#82
#83	# Parse organic results
#84	results = []
#85	for item in data.get("organic_results", [])[:num_results]:
#86	results.append(SearchResult(
#87	position=item.get("position", 0),
#88	title=item.get("title", ""),
#89	link=item.get("link", ""),
#90	snippet=item.get("snippet", ""),
#91	source=item.get("source"),
#92	date=item.get("date"),
#93	))
#94
#95	# Extract AI Overview if present
#96	ai_overview = None
#97	if "ai_overview" in data:
#98	ai_overview = data["ai_overview"].get("markdown") or data["ai_overview"].get("answer")
#99
#100	# Extract related searches
#101	related = None
#102	if "related_searches" in data:
#103	related = [rs.get("query") for rs in data["related_searches"][:5]]
#104
#105	return SearchResponse(
#106	query=query,
#107	total_results=data.get("search_information", {}).get("total_results", 0),
#108	results=results,
#109	ai_overview=ai_overview,
#110	knowledge_graph=data.get("knowledge_graph"),
#111	answer_box=data.get("answer_box"),
#112	related_searches=related,
#113	)
#114
#115	async def search_news(
#116	self,
#117	query: str,
#118	time_period: str = "last_day",
#119	) -> SearchResponse:
#120	"""
#121	Search for recent news articles.
#122
#123	Args:
#124	query: Search query
#125	time_period: Time period (last_hour, last_day, last_week, last_month)
#126
#127	Returns:
#128	SearchResponse with news results
#129	"""
#130	params = {
#131	"engine": "google",
#132	"q": query,
#133	"api_key": self.api_key,
#134	"time_period": time_period,
#135	"tbm": "nws", # News search
#136	}
#137
#138	response = await self._client.get(f"{self.base_url}/search", params=params)
#139	response.raise_for_status()
#140	data = response.json()
#141
#142	results = []
#143	for item in data.get("organic_results", []):
#144	results.append(SearchResult(
#145	position=item.get("position", 0),
#146	title=item.get("title", ""),
#147	link=item.get("link", ""),
#148	snippet=item.get("snippet", ""),
#149	source=item.get("source"),
#150	date=item.get("date"),
#151	))
#152
#153	return SearchResponse(
#154	query=query,
#155	total_results=len(results),
#156	results=results,
#157	)
#158
#159	async def close(self):
#160	"""Close the HTTP client"""
#161	await self._client.aclose()
#162
#163	async def __aenter__(self):
#164	return self
#165
#166	async def __aexit__(self, exc_type, exc_val, exc_tb):
#167	await self.close()
#168

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public