In today’s digital age, information is abundant, but finding the right data can be a challenge. A meta search engine aggregates results from multiple search engines, providing a more comprehensive view of available information. In this blog post, we’ll walk through the process of building a simple meta search engine in Python, complete with error handling, rate limiting, and privacy features.
What is a Meta Search Engine?
A meta search engine does not maintain its own database of indexed pages. Instead, it sends user queries to multiple search engines, collects the results, and presents them in a unified format. This approach allows users to access a broader range of information without having to search each engine individually.
Prerequisites
To follow along with this tutorial, you’ll need:
- Python installed on your machine (preferably Python 3.6 or higher).
- Basic knowledge of Python programming.
- An API key for Bing Search (you can sign up for a free tier).
Step 1: Set Up Your Environment
First, ensure you have the necessary libraries installed. We’ll use requests for making HTTP requests and json for handling JSON data.
You can install the requests library using pip:
pip install requests
Step 2: Define Your Search Engines
Create a new Python file named meta_search_engine.py and start by defining the search engines you want to query. For this example, we’ll use DuckDuckGo and Bing.
import requests
import json
import os
import time
# Define your search engines
SEARCH_ENGINES = {
"DuckDuckGo": "https://api.duckduckgo.com/?q={}&format=json",
"Bing": "https://api.bing.microsoft.com/v7.0/search?q={}&count=10",
}
BING_API_KEY = "YOUR_BING_API_KEY" # Replace with your Bing API Key
Step 3: Implement the Query Function
Next, create a function to query the search engines and retrieve results. We’ll also implement error handling to manage network issues gracefully.
def search(query):
results = []
# Query DuckDuckGo
ddg_url = SEARCH_ENGINES["DuckDuckGo"].format(query)
try:
response = requests.get(ddg_url)
response.raise_for_status() # Raise an error for bad responses
data = response.json()
for item in data.get("RelatedTopics", []):
if 'Text' in item and 'FirstURL' in item:
results.append({
'title': item['Text'],
'url': item['FirstURL']
})
except requests.exceptions.RequestException as e:
print(f"Error querying DuckDuckGo: {e}")
# Query Bing
bing_url = SEARCH_ENGINES["Bing"].format(query)
headers = {"Ocp-Apim-Subscription-Key": BING_API_KEY}
try:
response = requests.get(bing_url, headers=headers)
response.raise_for_status() # Raise an error for bad responses
data = response.json()
for item in data.get("webPages", {}).get("value", []):
results.append({
'title': item['name'],
'url': item['url']
})
except requests.exceptions.RequestException as e:
print(f"Error querying Bing: {e}")
return results
Step 4: Implement Rate Limiting
To prevent hitting API rate limits, we’ll implement a simple rate limiter using time.sleep().
# Rate limit settings
RATE_LIMIT = 1 # seconds between requests
def rate_limited_search(query):
time.sleep(RATE_LIMIT) # Wait before making the next request
return search(query)
Step 5: Add Privacy Features
To enhance user privacy, we’ll avoid logging user queries and implement a caching mechanism to temporarily store results.
CACHE_FILE = 'cache.json'
def load_cache():
if os.path.exists(CACHE_FILE):
with open(CACHE_FILE, 'r') as f:
return json.load(f)
return {}
def save_cache(results):
with open(CACHE_FILE, 'w') as f:
json.dump(results, f)
def search_with_cache(query):
cache = load_cache()
if query in cache:
print("Returning cached results.")
return cache[query]
results = rate_limited_search(query)
save_cache({query: results})
return results
Step 6: Remove Duplicates
To ensure the results are unique, we’ll implement a function to remove duplicates based on the URL.
def remove_duplicates(results):
seen = set()
unique_results = []
for result in results:
if result['url'] not in seen:
seen.add(result['url'])
unique_results.append(result)
return unique_results
Step 7: Display Results
Create a function to display the search results in a user-friendly format.
def display_results(results):
for idx, result in enumerate(results, start=1):
print(f"{idx}. {result['title']}\n {result['url']}\n")
Step 8: Main Function
Finally, integrate everything into a main function that runs the meta search engine.
def main():
query = input("Enter your search query: ")
results = search_with_cache(query)
unique_results = remove_duplicates(results)
display_results(unique_results)
if __name__ == "__main__":
main()
Complete Code
Here’s the complete code for your meta search engine:
import requests
import json
import os
import time
# Define your search engines
SEARCH_ENGINES = {
"DuckDuckGo": "https://api.duckduckgo.com/?q={}&format=json",
"Bing": "https://api.bing.microsoft.com/v7.0/search?q={}&count=10",
}
BING_API_KEY = "YOUR_BING_API_KEY" # Replace with your Bing API Key
# Rate limit settings
RATE_LIMIT = 1 # seconds between requests
def search(query):
results = []
# Query DuckDuckGo
ddg_url = SEARCH_ENGINES["DuckDuckGo"].format(query)
try:
response = requests.get(ddg_url)
response.raise_for_status()
data = response.json()
for item in data.get("RelatedTopics", []):
if 'Text' in item and 'FirstURL' in item:
results.append({
'title': item['Text'],
'url': item['FirstURL']
})
except requests.exceptions.RequestException as e:
print(f"Error querying DuckDuckGo: {e}")
# Query Bing
bing_url = SEARCH_ENGINES["Bing"].format(query)
headers = {"Ocp-Apim-Subscription-Key": BING_API_KEY}
try:
response = requests.get(bing_url, headers=headers)
response.raise_for_status()
data = response.json()
for item in data.get("webPages", {}).get("value", []):
results.append({
'title': item['name'],
'url': item['url']
})
except requests.exceptions.RequestException as e:
print(f"Error querying Bing: {e}")
return results
def rate_limited_search(query):
time.sleep(RATE_LIMIT)
return search(query)
CACHE_FILE = 'cache.json'
def load_cache():
if os.path.exists(CACHE_FILE):
with open(CACHE_FILE, 'r') as f:
return json.load(f)
return {}
def save_cache(results):
with open(CACHE_FILE, 'w') as f:
json.dump(results, f)
def search_with_cache(query):
cache = load_cache()
if query in cache:
print("Returning cached results.")
return cache[query]
results = rate_limited_search(query)
save_cache({query: results})
return results
def remove_duplicates(results):
seen = set()
unique_results = []
for result in results:
if result['url'] not in seen:
seen.add(result['url'])
unique_results.append(result)
return unique_results
def display_results(results):
for idx, result in enumerate(results, start=1):
print(f"{idx}. {result['title']}\n {result['url']}\n")
def main():
query = input("Enter your search query: ")
results = search_with_cache(query)
unique_results = remove_duplicates(results)
display_results(unique_results)
if __name__ == "__main__":
main()
Conclusion
Congratulations! You’ve built a simple yet functional meta search engine in Python. This project not only demonstrates how to aggregate search results from multiple sources but also emphasizes the importance of error handling, rate limiting, and user privacy. You can further enhance this engine by adding more search engines, implementing a web interface, or even integrating machine learning for improved result ranking. Happy coding!
Top comments (0)