В сегодняшнюю цифровую эпоху информации много, но найти нужные данные может оказаться непростой задачей. Метапоисковая система объединяет результаты нескольких поисковых систем, обеспечивая более полное представление доступной информации. В этом сообщении блога мы рассмотрим процесс создания простой метапоисковой системы на Python, дополненной обработкой ошибок, ограничением скорости и функциями конфиденциальности.

Что такое метапоисковая система?

Метапоисковая система не ведет собственную базу данных проиндексированных страниц. Вместо этого он отправляет пользовательские запросы в несколько поисковых систем, собирает результаты и представляет их в едином формате. Такой подход позволяет пользователям получить доступ к более широкому спектру информации без необходимости искать в каждой системе индивидуально.

Предварительные условия

Чтобы следовать этому руководству, вам понадобится:

• На вашем компьютере установлен Python (предпочтительно Python 3.6 или выше).
• Базовые знания программирования на Python.
• Ключ API для поиска Bing (вы можете подписаться на бесплатный уровень).

Шаг 1. Настройте среду

Во-первых, убедитесь, что у вас установлены необходимые библиотеки. Мы будем использовать запросы для выполнения HTTP-запросов и json для обработки данных JSON.

Вы можете установить библиотеку запросов с помощью pip:

pip install requests

Шаг 2. Определите свои поисковые системы

Создайте новый файл Python с именем Meta_search_engine.py и начните с определения поисковых систем, к которым вы хотите выполнить запрос. В этом примере мы будем использовать 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: Основная функция

Наконец, интегрируйте все в основную функцию, которая запускает метапоисковую систему.

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. Этот проект не только демонстрирует, как агрегировать результаты поиска из нескольких источников, но также подчеркивает важность обработки ошибок, ограничения скорости и конфиденциальности пользователей. Вы можете еще больше улучшить эту систему, добавив больше поисковых систем, внедрив веб-интерфейс или даже интегрировав машинное обучение для улучшения ранжирования результатов. Приятного кодирования!