使用 Python 建立元搜尋引擎：逐步指南

發佈於2024-08-28

Building a Meta Search Engine in Python: A Step-by-Step Guide 在当今的数字时代，信息丰富，但找到正确的数据可能是一个挑战。元搜索引擎聚合来自多个搜索引擎的结果，提供更全面的可用信息视图。在这篇博文中，我们将逐步介绍用 Python 构建一个简单的元搜索引擎的过程，包括错误处理、速率限制和隐私功能。

什么是元搜索引擎？

元搜索引擎不维护自己的索引页面数据库。相反，它将用户查询发送到多个搜索引擎，收集结果，并以统一的格式呈现它们。这种方法允许用户访问更广泛的信息，而无需单独搜索每个引擎。

先决条件

要学习本教程，您需要：

您的计算机上已安装 Python（最好是 Python 3.6 或更高版本）。
Python编程基础知识。
Bing 搜索的 API 密钥（您可以注册免费套餐）。

第 1 步：设置您的环境

首先，确保您安装了必要的库。我们将使用 requests 来发出 HTTP 请求，使用 json 来处理 JSON 数据。

您可以使用pip安装requests库：

pip install requests

第 2 步：定义您的搜索引擎

创建一个名为meta_search_engine.py 的新Python 文件，并首先定义要查询的搜索引擎。在此示例中，我们将使用 DuckDuckGo 和 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

第三步：实现查询功能

接下来，创建一个函数来查询搜索引擎并检索结果。我们还将实施错误处理以优雅地管理网络问题。

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

第 4 步：实施速率限制

为了防止达到 API 速率限制，我们将使用 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)

第 5 步：添加隐私功能

为了增强用户隐私，我们将避免记录用户查询并实施缓存机制来临时存储结果。

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

第 6 步：删除重复项

为了确保结果是唯一的，我们将实现一个根据 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

第 7 步：显示结果

创建一个函数，以用户友好的格式显示搜索结果。

def display_results(results):
    for idx, result in enumerate(results, start=1):
        print(f"{idx}. {result['title']}\n   {result['url']}\n")

第8步：主要功能

最后，将所有内容集成到运行元搜索引擎的主函数中。

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()

完整代码

这是元搜索引擎的完整代码：

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()