今日のデジタル時代には情報が豊富ですが、適切なデータを見つけるのが難しい場合があります。メタ検索エンジンは、複数の検索エンジンからの結果を集約し、利用可能な情報のより包括的なビューを提供します。このブログ投稿では、エラー処理、レート制限、プライバシー機能を備えたシンプルなメタ検索エンジンを Python で構築するプロセスを順を追って説明します。

メタ検索エンジンとは何ですか?

メタ検索エンジンは、インデックス付きページの独自のデータベースを維持しません。代わりに、ユーザーのクエリを複数の検索エンジンに送信し、結果を収集し、統一された形式で表示します。このアプローチにより、ユーザーは各エンジンを個別に検索することなく、より広範囲の情報にアクセスできるようになります。

前提条件

このチュートリアルを進めるには、次のものが必要です:

• マシンにインストールされている Python (できれば Python 3.6 以降)。
• Python プログラミングの基礎知識。
• Bing Search の API キー (無料枠にサインアップできます)。

ステップ 1: 環境をセットアップする

まず、必要なライブラリがインストールされていることを確認してください。 HTTP リクエストの作成にはリクエストを使用し、JSON データの処理には json を使用します。

pip:
を使用してリクエスト ライブラリをインストールできます。

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

ステップ 3: クエリ関数を実装する

次に、検索エンジンにクエリを実行して結果を取得する関数を作成します。また、ネットワークの問題を適切に管理するためのエラー処理も実装します。

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: メイン関数

最後に、メタ検索エンジンを実行する main 関数にすべてを統合します。

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

結論

おめでとう！シンプルでありながら機能的なメタ検索エンジンを Python で構築しました。このプロジェクトは、複数のソースからの検索結果を集約する方法を示すだけでなく、エラー処理、レート制限、およびユーザーのプライバシーの重要性も強調します。検索エンジンを追加したり、Web インターフェイスを実装したり、結果ランキングを向上させるために機械学習を統合したりすることで、このエンジンをさらに強化できます。コーディングを楽しんでください!