api/api_utils/api_handler.py - master - outthevoid/python_telegram_bot_project

api/api_utils/api_handler.py
199 строк7 KB
VladimirChukavin
Добавил функцию _search_movies_by_budget, метод search_movies_by_budget, описание этого метода.
10 сен 2025, 17:09
10 сен 2025, 17:0981f9a42
from typing import Callable, Any

import requests
from loguru import logger


def _get_response(
    method: str, url: str, headers: dict, params: dict, timeout: int
) -> int | requests.Response:
    """
    Функция для получения ответа от сервера
    после отправки запроса
    :param method: GET
    :param url: Адрес сайта
    :param headers: Заголовок запроса
    :param params: Параметры запроса
    :param timeout: Время ожидания ответа сервера
    :return: Код ответа сервера, если запрос завершился
    неудачно, или ответ сервера (объект Response)
    """
    response = requests.request(
        method, url, headers=headers, params=params, timeout=timeout
    )
    status_code = response.status_code

    if status_code != 200:
        return status_code

    return response


def _search_movies_by_name(
    method: str,
    url: str,
    headers: dict,
    name: str,
    limit: int = 250,
    timeout: int = 5,
    func=_get_response,
) -> list[Any, Any] | Any:
    """
    Функция для поиска фильма по названию
    :param method: GET
    :param url: Адрес сайта
    :param headers: Заголовок запроса
    :param name: Название фильма
    :param limit: Количество вариантов в ответе
    :param timeout: Время ожидания ответа сервера
    :param func: Функция для получения ответа сервера
    :return: Список фильмов
    """
    try:
        url = "{}/{}".format(url, "search")
        params = {"limit": limit, "query": name}
        response = func(method, url, headers=headers, params=params, timeout=timeout)
        response.raise_for_status()
        data = response.json()
        return data.get("docs", [])
    except Exception as e:
        logger.error("Ошибка запроса поиска фильма по названию: {}".format(e))
        return []


def _search_movies_by_rating(
    method: str,
    url: str,
    headers: dict,
    min_rating: float,
    max_rating: float,
    limit: int = 250,
    timeout: int = 5,
    func=_get_response,
) -> list[Any, Any] | Any:
    """
    Функция для поиска фильма по рейтингу
    :param method: GET
    :param url: Адрес сайта
    :param headers: Заголовок запроса
    :param min_rating: Минимальный рейтинг
    :param max_rating: Максимальный рейтинг
    :param limit: Количество вариантов в ответе
    :param timeout: Время ожидания ответа сервера
    :param func: Функция для получения ответа сервера
    :return: Список фильмов
    """
    try:
        params = {
            "limit": limit,
            "selectFields": [
                "name",
                "alternativeName",
                "description",
                "year",
                "rating",
                "ageRating",
                "genres",
                "poster",
            ],
            "rating.kp": [f"{min_rating}-{max_rating}"],
        }
        response = func(method, url, headers=headers, params=params, timeout=timeout)
        response.raise_for_status()
        data = response.json()
        return data.get("docs", [])
    except Exception as e:
        logger.error("Ошибка запроса поиска фильма по рейтингу: {}".format(e))
        return []


def _search_movies_by_budget(
    method: str,
    url: str,
    headers: dict,
    min_budget: int = None,
    max_budget: int = None,
    limit: int = 250,
    timeout: int = 5,
    func=_get_response,
) -> list[Any, Any] | Any:
    """
    Функция для поиска фильма по величине бюджета
    :param method: GET
    :param url: Адрес сайта
    :param headers: Заголовок запроса
    :param min_budget: Минимальный бюджет
    :param max_budget: Максимальный бюджет
    :param limit: Количество вариантов в ответе
    :param timeout: Время ожидания ответа сервера
    :param func: Функция для получения ответа сервера
    :return: Список фильмов
    """
    try:
        budget = ""
        if min_budget is None and max_budget:
            budget = "0-{}".format(max_budget)
        elif min_budget and max_budget is None:
            budget = "{}-50000000000".format(min_budget)

        params = {
            "limit": limit,
            "selectFields": [
                "name",
                "alternativeName",
                "description",
                "year",
                "rating",
                "ageRating",
                "genres",
                "poster",
            ],
            # "budget.value": [f"{min_budget}-{max_budget}"],
            "budget.value": [budget],
        }
        response = func(method, url, headers=headers, params=params, timeout=timeout)
        response.raise_for_status()
        data = response.json()
        return data.get("docs", [])
    except Exception as e:
        logger.error("Ошибка запроса поиска фильма по бюджету: {}".format(e))
        return []


class APISearchInterface:
    """
    Класс для организации методов поиска фильмов.

    Методы
    ------
    search_movies_by_name():
        поиск фильмов по названию
    search_movies_by_rating():
        поиск фильмов по рейтингу
    search_movies_by_budget():
        поиск фильмов по бюджету
    """

    @staticmethod
    def search_movies_by_name() -> Callable:
        """
        Метод для поиска фильмов по названию
        :return: Функция для поиска фильмов
        """
        return _search_movies_by_name

    @staticmethod
    def search_movies_by_rating() -> Callable:
        """
        Метод для поиска фильмов по рейтингу
        :return: Функция для поиска фильмов
        """
        return _search_movies_by_rating

    @staticmethod
    def search_movies_by_budget() -> Callable:
        """
        Метод для поиска фильмов по бюджету
        :return: Функция для поиска фильмов
        """
        return _search_movies_by_budget