searx-qt/searxqt/core/requests.py

########################################################################
#  Searx-Qt - Lightweight desktop application for Searx.
#  Copyright (C) 2020-2022  CYBERDEViL
#
#  This file is part of Searx-Qt.
#
#  Searx-Qt is free software: you can redistribute it and/or modify
#  it under the terms of the GNU General Public License as published by
#  the Free Software Foundation, either version 3 of the License, or
#  (at your option) any later version.
#
#  Searx-Qt is distributed in the hope that it will be useful,
#  but WITHOUT ANY WARRANTY; without even the implied warranty of
#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#  GNU General Public License for more details.
#
#  You should have received a copy of the GNU General Public License
#  along with this program.  If not, see <https://www.gnu.org/licenses/>.
#
########################################################################


import requests
from requests.exceptions import (
    HTTPError,
    ConnectionError,
    Timeout,
    ProxyError,
    SSLError,
    RequestException
)

import json

from searxqt.core import jsonVerify, log


class ErrorType:
    Success = 0
    HttpError = 1
    ConnectionError = 2
    Timeout = 3
    WrongStatus = 4
    DecodeError = 5
    NoResults = 6
    ProxyError = 7
    SSLError = 8
    WrongJsonStruct = 9
    Other = 10


ErrorTypeStr = {
    ErrorType.Success: "Success",
    ErrorType.HttpError: "HttpError",
    ErrorType.ConnectionError: "ConnectionError",
    ErrorType.Timeout: "Timeout",
    ErrorType.WrongStatus: "WrongStatus",
    ErrorType.DecodeError: "DecodeError",
    ErrorType.NoResults: "NoResults",
    ErrorType.ProxyError: "ProxyError",
    ErrorType.SSLError: "SSLError",
    ErrorType.WrongJsonStruct: "WrongJsonStruct",
    ErrorType.Other: "Other"
}


class Result:
    def __init__(self, url, response, err="", errType=ErrorType.Success, acceptCodes=None):
        self._url = url  # url used for request.
        self._response = response
        self._err = err
        self._errType = errType

        acceptCodes = acceptCodes
        if not acceptCodes:
            acceptCodes = [200]

        if errType == ErrorType.Success and response.status_code not in acceptCodes:
            self._errType = ErrorType.WrongStatus
            self._err = "WrongStatus: {0}".format(self._response.status_code)
        else:
            self.verifyFurther()

    def __bool__(self):
        return not self.failed()

    def url(self):
        return self._url

    def errorType(self): return self._errType

    def error(self): return self._err

    def content(self):
        """ In case json.loads failed and we want to debug.
        """
        if self._response is None:
            return b''
        return self._response.content

    def text(self):
        if self._response is None:
            return ''
        return self._response.text

    def failed(self):
        if self._errType is not ErrorType.Success:
            return True
        return False

    def statusCode(self):
        if self._response is not None:
            return self._response.status_code
        return 0

    def verifyFurther(self):
        pass


class JsonResult(Result):
    ExpectedStructure = {}

    def __init__(self, url, response, err="", errType=ErrorType.Success, acceptCodes=None):
        Result.__init__(
            self,
            url,
            response,
            err=err,
            errType=errType,
            acceptCodes=acceptCodes
        )

    def verifyFurther(self):
        try:
            self.json()
        except json.JSONDecodeError as err:
            self._errType = ErrorType.DecodeError
            self._err = "DecodeError: `{0}`".format(err)
        except UnicodeDecodeError as err:
            # This could happen when the response encoding isn't plain ? (gzip)
            # Or we just have malformed data/crap.
            self._errType = ErrorType.DecodeError
            self._err = "DecodeError: `{0}`".format(err)

        # Verify the json structure itself.
        verified, error = jsonVerify.verifyStructure(
            self.ExpectedStructure,
            self.json(),
            path=str(self.__class__.__name__) + ".json()"
        )
        if not verified:
            self._errType = ErrorType.WrongJsonStruct
            self._err = "WrongJsonStruct: `{0}`".format(error)

    def json(self):
        if self.errorType() != ErrorType.Success:
            return {}
        return json.loads(self._response.content)


class RequestsHandler:
    def __init__(self, settings):
        """
        @param settings: TODO don't use Qt stuff in core.
        @type settings: searxqt.models.RequestSettingsModel
        """
        self._settings = settings

    def failSafeRequestFactory(func):
        def failSafeRequest(self, url, data=None, ResultType=None):
            response = None
            err = ""
            errType = ErrorType.Success

            if not ResultType:
                # When 'ResultType' isn't specified, set 'JsonResult' as
                # default.
                ResultType = JsonResult

            log.debug("<NEW Request>", self)
            log.debug("# ------------------------", self)
            log.debug("# ResultType : {0}".format(ResultType), self)

            """
            Request exceptions
            https://docs.python-requests.org/en/master/_modules/requests/exceptions/
            """

            try:
                response = func(self, url, data=data, **self._settings.kwargs)
            except HTTPError as e:
                # HTTPError is subclass of RequestException
                log.debug("Request failed! HTTPError: {0}".format(e), self)
                errType = ErrorType.HttpError
                err = str(e)
            except Timeout as e:
                # Timeout is subclass of RequestException
                log.debug("Request failed! Timeout: {0}".format(e), self)
                errType = ErrorType.Timeout
                err = str(e)
            except ProxyError as e:
                # ProxyError is subclass of ConnectionError
                log.debug("Request failed! ProxyError: {0}".format(e), self)
                errType = ErrorType.ProxyError
                err = str(e)
            except SSLError as e:
                # SSLError is subclass of ConnectionError
                log.debug("Request failed! SSLError: {0}".format(e), self)
                errType = ErrorType.SSLError
                err = str(e)
            except ConnectionError as e:
                # ConnectionError is subclass of RequestException
                log.debug(
                    "Request failed! ConnectionError: {0}".format(e),
                    self
                )
                errType = ErrorType.ConnectionError
                err = str(e)
            except RequestException as e:
                # This should catch all other
                log.debug(
                    "Request failed! RequestException: {0}".format(e),
                    self
                )
                errType = ErrorType.Other
                err = str(e)

            log.debug("# ------------------------\n", self)

            return ResultType(url, response, err=err, errType=errType)

        return failSafeRequest

    @failSafeRequestFactory
    def get(self, url, data=None, ResultType=None, **settingsKwargs):
        log.debug("# Type       : GET", self)
        log.debug("# URL        : {0}".format(url), self)
        log.debug("# Data       : {0}".format(data), self)
        log.debug("# Kwargs     : {0}".format(settingsKwargs), self)
        return requests.get(url, data=data, **settingsKwargs)

    @failSafeRequestFactory
    def post(self, url, data=None, ResultType=None, **settingsKwargs):
        log.debug("# Type       : POST", self)
        log.debug("# URL        : {0}".format(url), self)
        log.debug("# Data       : {0}".format(data), self)
        log.debug("# Kwargs     : {0}".format(settingsKwargs), self)
        return requests.post(url, data=data, **settingsKwargs)