searxpy/searxpy/searx.py

########################################################################
#  searxpy - Provides Python modules for interacting with searx and
#       searx-stats2 instances.
#  Copyright (C) 2020  CYBERDEViL
#
#  This file is part of searxpy.
#
#  searxpy 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.
#
#  searxpy 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
from copy import deepcopy
import json


class SearchErrorType:
    HttpError = 0
    ConnectionError = 1
    Timeout = 3
    WrongStatus = 4
    DecodeError = 5
    NoResults = 6


class SearchResult:
    def __init__(self, response=None, err="", errType=None):
        self._response = response
        self._err = err
        self._errType = errType

        if errType is None and response.status_code != 200:
            self._errType = SearchErrorType.WrongStatus
            self._err = "WrongStatus: {0}".format(self._response.status_code)
        else:
            try:
                self.json()
            except json.JSONDecodeError as err:
                self._errType = SearchErrorType.DecodeError
                self._err = "DecodeError: `{0}`".format(err)

        if not len(self.json().get('results', {})):
            self._errType = SearchErrorType.NoResults
            self._err = "NoResults: got: `{0}`".format(self.json())

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

    def errorType(self):
        """
        @return: Error type (see SearchErrorType), None if no error.
        @rtype: int or None
        """
        return self._errType

    def error(self):
        """
        @return: Error message if a error occured, else empty string.
        @rtype: str
        """
        return self._err

    def failed(self):
        """ Use this to check if there occured a error first!

        @return: True if a error occured, False if all is good.
        @rtype: bool
        """
        if self._errType is not None:
            return True
        return False

    def json(self):
        """ Response data as json object.

        @rtype: json
        """
        return json.loads(self._response.content)

    def data(self):
        """
        @return: Response data as a dict object.
        @rtype: dict
        """
        if self.errorType() is not None:
            return {}
        return dict(self.json())


class Categories:
    types = {
        'general': ('General', 'category_general'),
        'files': ('Files', 'category_files'),
        'images': ('Images', 'category_images'),
        'videos': ('Videos', 'category_videos'),
        'it': ('IT', 'category_it'),
        'map': ('Location', 'category_map'),
        'music': ('Music', 'category_music'),
        'news': ('News', 'category_news'),
        'science': ('Science', 'category_science'),
        'social': ('Social', 'category_social media')
        }

    def __init__(self):
        """ Enabled / disable certain search catagories.
        """
        self._options = {}
        self.__makeOptions()

    def __makeOptions(self):
        self._options.clear()
        for key, t in self.types.items():
            self._options.update({key: False})

    def get(self, key):
        """
        @param key: One of the keys in Categories.types
        @type key: str

        @return: Returns a tuple with two values.
                [0] str Fancy name
                [1] str Searx category var name
        @rtype: tuple
        """
        return self._options[key]

    def set(self, key, state):
        """
        @param key: One of the keys in Categories.types
        @type key: str

        @param state: Enabled / disabled state
        @type state: bool
        """
        self._options[key] = state

    def dict(self):
        """Returns data on what categories to use, this can be posted or
        added as get vars for a request to a Searx instance.

        @rtype: dict
        """
        newDict = {}
        for key, state in self._options.items():
            if state:
                newDict.update({self.types[key][1]: 'on'})
        return newDict


class Searx:
    def __init__(self, url=""):
        """ Preform searches on Searx instances.

        @param url: Url of the Searx instance
        @type url: str
        """
        self._url = url
        self._categories = Categories()
        self._kwargs = {
            'data': {
                'q': '',
                'format': 'json',
                'lang': 'all',
                'pageno': '1'
            },
        }

    @property
    def categories(self):
        """
        @return: Enable / disable search categories with the returned
                object.
        @rtype: Categories
        """
        return self._categories

    @property
    def url(self):
        """
        @return: Instance url
        @rtype: str
        """
        return self._url

    @url.setter
    def url(self, url):
        """
        @param url: Instance url
        @type url: str
        """
        self._url = url

    @property
    def query(self):
        """
        @return: Search query
        @rtype: str
        """
        return self._kwargs['data']['q']

    @query.setter
    def query(self, q):
        """
        @param q: Search query
        @type q: str
        """
        self._kwargs['data']['q'] = q

    @property
    def lang(self):
        """
        @return: Language code
        @rtype: str
        """
        return self._kwargs['data']['lang']

    @lang.setter
    def lang(self, lang):
        """
        @param lang: Language code
        @type lang: str
        """
        self._kwargs['data']['lang'] = lang

    @property
    def pageno(self):
        """
        @return: Page number
        @rtype: int
        """
        return int(self._kwargs['data']['pageno'])

    @pageno.setter
    def pageno(self, i):
        """
        @param i: Page number
        @type i: int
        """
        self._kwargs['data']['pageno'] = str(i)

    def search(self, requestKwargs={}):
        """ Preform search

        @param requestKwargs: Keyworded arguments passed to requests.post
                for example setting a proxy or setting ssl behaviour.
        @type requestKwargs: dict

        @return: Search results
        @rtype: SearchResult
        """
        requestKwargs.update(deepcopy(self._kwargs))
        requestKwargs['data'].update(self.categories.dict())

        response = None
        err = ""
        errType = None

        try:
            response = requests.post(self._url, **requestKwargs)
        except HTTPError as e:
            print("Search failed! HTTPError: {0}".format(e))
            errType = SearchErrorType.HttpError
            err = e
        except ConnectionError as e:
            print("Search failed! ConnectionError: {0}".format(e))
            errType = SearchErrorType.ConnectionError
            err = e
        except Timeout as e:
            print("Search failed! Timeout: {0}".format(e))
            errType = SearchErrorType.Timeout
            err = e

        return SearchResult(response=response, err=err, errType=errType)