""" Google Search API handler. Uses the Serper API to access Google search results. """ import os import json import requests from typing import Dict, List, Any, Optional from .base_handler import BaseSearchHandler from config.config import get_config, get_api_key class GoogleSearchHandler(BaseSearchHandler): """Handler for Google Search using the Serper API.""" def __init__(self): """Initialize the Google search handler.""" self.config = get_config() self.api_key = get_api_key("serper") self.base_url = "https://google.serper.dev/search" self.available = self.api_key is not None def search(self, query: str, num_results: int = 10, **kwargs) -> List[Dict[str, Any]]: """ Execute a Google search query using Serper API. Args: query: The search query to execute num_results: Number of results to return **kwargs: Additional search parameters: - country: Country code (default: "us") - language: Language code (default: "en") - page: Page number (default: 1) Returns: List of search results with standardized format """ if not self.available: raise ValueError("Google Search API is not available. API key is missing.") # Set up the request parameters params = { "q": query, "num": num_results, "type": "search" # Specify search type } # Add optional parameters if "country" in kwargs: params["gl"] = kwargs["country"] if "language" in kwargs: params["hl"] = kwargs["language"] if "page" in kwargs: params["page"] = kwargs["page"] # Set up the headers headers = { "X-API-KEY": self.api_key, "Content-Type": "application/json" } try: # Make the request response = requests.post( self.base_url, headers=headers, json=params ) response.raise_for_status() # Parse the response data = response.json() # Extract and standardize the results results = [] # Process organic results if "organic" in data: for item in data["organic"][:num_results]: result = { "title": item.get("title", ""), "url": item.get("link", ""), "snippet": item.get("snippet", ""), "source": "google", "position": item.get("position", 0), "raw_data": item } results.append(result) return results except requests.exceptions.RequestException as e: print(f"Error executing Google search: {e}") return [] def get_name(self) -> str: """Get the name of the search handler.""" return "google" def is_available(self) -> bool: """Check if the Google Search API is available.""" return self.available def get_rate_limit_info(self) -> Dict[str, Any]: """Get information about the API's rate limits.""" # These are example values - adjust based on your Serper plan return { "requests_per_minute": 60, "requests_per_day": 2500, "current_usage": None # Serper doesn't provide usage info in responses }