zaclys/searxng
1
0
Fork 1
mirror of https://github.com/searxng/searxng synced 2024-01-01 19:24:07 +01:00
Code Issues Projects Releases Packages Wiki Activity

[mod] mudularize & document searx.results

Browse source 
The intention of this patch is to improve modularization & documentation of the
implementations about the *result* items.

  This patch does not contain any functional change!

Signed-off-by: Markus Heiser <markus.heiser@darmarit.de>
This commit is contained in:
Markus Heiser 2022-06-21 13:14:11 +02:00
parent cee586029c
commit 069e1d7fb4
15 changed files with 935 additions and 262 deletions
Download patch file Download diff file Expand all files Collapse all files

3
searx/results/__init__.py Normal file
View file

@ -0,0 +1,3 @@
# SPDX-License-Identifier: AGPL-3.0-or-later
# lint: pylint
"""Implementation of the result container and the result types."""

37
searx/results/answer.py Normal file
View file

@ -0,0 +1,37 @@
# SPDX-License-Identifier: AGPL-3.0-or-later
# lint: pylint
"""Answer item in the result list. The answer result item is used in
the :origin:`results.html <searx/templates/simple/results.html>` template.
A answer item is a dictionary type with dedicated keys and values.
.. code:: python
results.append({
'answer' : str,
'url' : str,
})
answer : ``str``
The answer string append by the engine.
url : ``str``
A link that is related to the answer (e.g. the origin of the answer).
"""
def is_answer(result):
"""Returns ``True`` if result type is :py:obj:`.answer`, otherwise ``False``
In the result list a answer item is identified by the existence of the key
``answer``.
"""
return 'answer' in result
class Answers(dict):
"""Dictionary of answers in the :py:obj:`.container.ResultContainer`"""
def add(self, result):
self[result['answer']] = result

348
searx/results/container.py Normal file
View file

@ -0,0 +1,348 @@
# SPDX-License-Identifier: AGPL-3.0-or-later
# lint: pylint
"""ResultContainer
"""
from collections import defaultdict
from operator import itemgetter
from threading import RLock
from typing import List, Set
from urllib.parse import urlparse
from searx import logger
from searx.engines import engines
from searx.metrics import histogram_observe, counter_add, count_error
from .core import (
WHITESPACE_REGEX,
Timing,
UnresponsiveEngine,
result_content_len,
result_score,
compare_urls,
)
from .infobox import Infoboxes, merge_two_infoboxes, is_infobox
from .suggestion import Suggestions, is_suggestion
from .answer import Answers, is_answer
from .correction import Corrections, is_correction
def is_number_of_results(result):
"""Returns ``True`` if result type is ``number_of_results``, otherwise
``False``"""
return 'number_of_results' in result
def is_engine_data(result):
"""Returns ``True`` if result type is :ref:`engine_data`, otherwise ``False``"""
return 'engine_data' in result
def is_standard_result(result):
"""Returns ``True`` if result type is a :ref:`standard result <standard
result>`, otherwise ``False``"""
return 'url' in result
class ResultContainer:
"""A container to organize the result items and the various result types. New
results can be added by :py:obj:`ResultContainer.extend`.
To be clear, a result-type is not a special python data-type, a result is
always a python dicticonary. The result-type is determined by the presence
of one of the following keys (in that order, first match wins)
1. suggestion: :py:obj:`.suggestion`
2. answer: :py:obj:`.answer`
3. correction: :py:obj:`.correction`
4. infobox: :py:obj:`.infobox`
5. number_of_results: Number of results origin engine has.
.. code:: python
results.append({
'number_of_results' : int,
})
6. engine_data: used to pass :ref:`engine_data <engine_data>` to next request.
7. url: :ref:`standard result <standard result>`
"""
__slots__ = (
'_merged_results',
'infoboxes',
'suggestions',
'answers',
'corrections',
'_number_of_results',
'_closed',
'paging',
'unresponsive_engines',
'timings',
'redirect_url',
'engine_data',
'on_result',
'_lock',
)
def __init__(self):
super().__init__()
self._merged_results = []
self.infoboxes = Infoboxes()
self.suggestions = Suggestions()
self.answers = Answers()
self.corrections = Corrections()
self._number_of_results = []
self.engine_data = defaultdict(dict)
self._closed = False
self.paging = False
self.unresponsive_engines: Set[UnresponsiveEngine] = set()
self.timings: List[Timing] = []
self.redirect_url = None
self.on_result = lambda _: True
self._lock = RLock()
def extend(self, engine_name, results): # pylint: disable=too-many-branches
"""Add a result item to the container."""
if self._closed:
return
standard_result_count = 0
error_msgs = set()
for result in list(results):
result['engine'] = engine_name
if is_suggestion(result) and self.on_result(result):
self.suggestions.add(result['suggestion'])
elif is_answer(result) and self.on_result(result):
self.answers.add(result)
elif is_correction(result) and self.on_result(result):
self.corrections.add(result['correction'])
elif is_infobox(result) and self.on_result(result):
self._merge_infobox(result)
elif is_number_of_results(result) and self.on_result(result):
self._number_of_results.append(result['number_of_results'])
elif is_engine_data(result) and self.on_result(result):
self.engine_data[engine_name][result['key']] = result['engine_data']
elif is_standard_result(result):
# standard result (url, title, content)
if not self._is_valid_url_result(result, error_msgs):
continue
# normalize the result
self._normalize_url_result(result)
# call on_result call searx.search.SearchWithPlugins._on_result
# which calls the plugins
if not self.on_result(result):
continue
self.__merge_url_result(result, standard_result_count + 1)
standard_result_count += 1
elif self.on_result(result):
self.__merge_result_no_url(result, standard_result_count + 1)
standard_result_count += 1
if len(error_msgs) > 0:
for msg in error_msgs:
count_error(engine_name, 'some results are invalids: ' + msg, secondary=True)
if engine_name in engines:
histogram_observe(standard_result_count, 'engine', engine_name, 'result', 'count')
if not self.paging and standard_result_count > 0 and engine_name in engines and engines[engine_name].paging:
self.paging = True
def _merge_infobox(self, infobox):
add_infobox = True
infobox_id = infobox.get('id', None)
infobox['engines'] = set([infobox['engine']])
if infobox_id is not None:
parsed_url_infobox_id = urlparse(infobox_id)
with self._lock:
for existingIndex in self.infoboxes:
if compare_urls(urlparse(existingIndex.get('id', '')), parsed_url_infobox_id):
merge_two_infoboxes(existingIndex, infobox)
add_infobox = False
if add_infobox:
self.infoboxes.append(infobox)
def _is_valid_url_result(self, result, error_msgs):
if 'url' in result:
if not isinstance(result['url'], str):
logger.debug('result: invalid URL: %s', str(result))
error_msgs.add('invalid URL')
return False
if 'title' in result and not isinstance(result['title'], str):
logger.debug('result: invalid title: %s', str(result))
error_msgs.add('invalid title')
return False
if 'content' in result:
if not isinstance(result['content'], str):
logger.debug('result: invalid content: %s', str(result))
error_msgs.add('invalid content')
return False
return True
def _normalize_url_result(self, result):
"""Return True if the result is valid"""
result['parsed_url'] = urlparse(result['url'])
# if the result has no scheme, use http as default
if not result['parsed_url'].scheme:
result['parsed_url'] = result['parsed_url']._replace(scheme="http")
result['url'] = result['parsed_url'].geturl()
# avoid duplicate content between the content and title fields
if result.get('content') == result.get('title'):
del result['content']
# make sure there is a template
if 'template' not in result:
result['template'] = 'default.html'
# strip multiple spaces and cariage returns from content
if result.get('content'):
result['content'] = WHITESPACE_REGEX.sub(' ', result['content'])
def __merge_url_result(self, result, position):
result['engines'] = set([result['engine']])
with self._lock:
duplicated = self.__find_duplicated_http_result(result)
if duplicated:
self.__merge_duplicated_http_result(duplicated, result, position)
return
# if there is no duplicate found, append result
result['positions'] = [position]
self._merged_results.append(result)
def __find_duplicated_http_result(self, result):
result_template = result.get('template')
for merged_result in self._merged_results:
if 'parsed_url' not in merged_result:
continue
if compare_urls(result['parsed_url'], merged_result['parsed_url']) and result_template == merged_result.get(
'template'
):
if result_template != 'images.html':
# not an image, same template, same url : it's a duplicate
return merged_result
# it's an image
# it's a duplicate if the parsed_url, template and img_src are differents
if result.get('img_src', '') == merged_result.get('img_src', ''):
return merged_result
return None
def __merge_duplicated_http_result(self, duplicated, result, position):
# using content with more text
if result_content_len(result.get('content', '')) > result_content_len(duplicated.get('content', '')):
duplicated['content'] = result['content']
# merge all result's parameters not found in duplicate
for key in result.keys():
if not duplicated.get(key):
duplicated[key] = result.get(key)
# add the new position
duplicated['positions'].append(position)
# add engine to list of result-engines
duplicated['engines'].add(result['engine'])
# using https if possible
if duplicated['parsed_url'].scheme != 'https' and result['parsed_url'].scheme == 'https':
duplicated['url'] = result['parsed_url'].geturl()
duplicated['parsed_url'] = result['parsed_url']
def __merge_result_no_url(self, result, position):
result['engines'] = set([result['engine']])
result['positions'] = [position]
with self._lock:
self._merged_results.append(result)
def close(self):
self._closed = True
for result in self._merged_results:
score = result_score(result)
result['score'] = score
for result_engine in result['engines']:
counter_add(score, 'engine', result_engine, 'score')
results = sorted(self._merged_results, key=itemgetter('score'), reverse=True)
# pass 2 : group results by category and template
gresults = []
categoryPositions = {}
for res in results:
# pylint: disable=fixme
# FIXME : handle more than one category per engine
engine = engines[res['engine']]
res['category'] = engine.categories[0] if len(engine.categories) > 0 else ''
# FIXME : handle more than one category per engine
category = (
res['category']
+ ':'
+ res.get('template', '')
+ ':'
+ ('img_src' if 'img_src' in res or 'thumbnail' in res else '')
)
current = None if category not in categoryPositions else categoryPositions[category]
# group with previous results using the same category
# if the group can accept more result and is not too far
# from the current position
if current is not None and (current['count'] > 0) and (len(gresults) - current['index'] < 20):
# group with the previous results using
# the same category with this one
index = current['index']
gresults.insert(index, res)
# update every index after the current one
# (including the current one)
for k in categoryPositions: # pylint: disable=consider-using-dict-items
v = categoryPositions[k]['index']
if v >= index:
categoryPositions[k]['index'] = v + 1
# update this category
current['count'] -= 1
else:
# same category
gresults.append(res)
# update categoryIndex
categoryPositions[category] = {'index': len(gresults), 'count': 8}
# update _merged_results
self._merged_results = gresults
def get_ordered_results(self):
if not self._closed:
self.close()
return self._merged_results
def results_length(self):
return len(self._merged_results)
def results_number(self):
resultnum_sum = sum(self._number_of_results)
if not resultnum_sum or not self._number_of_results:
return 0
return resultnum_sum / len(self._number_of_results)
def add_unresponsive_engine(self, engine_name: str, error_type: str, suspended: bool = False):
if engines[engine_name].display_error_messages:
self.unresponsive_engines.add(UnresponsiveEngine(engine_name, error_type, suspended))
def add_timing(self, engine_name: str, engine_time: float, page_load_time: float):
self.timings.append(Timing(engine_name, total=engine_time, load=page_load_time))
def get_timings(self):
return self.timings

80
searx/results/core.py Normal file
View file

@ -0,0 +1,80 @@
# SPDX-License-Identifier: AGPL-3.0-or-later
# lint: pylint
"""Core methods
"""
# pylint: disable=too-few-public-methods
import re
from urllib.parse import unquote
from typing import NamedTuple
from searx.engines import engines
CONTENT_LEN_IGNORED_CHARS_REGEX = re.compile(r'[,;:!?\./\\\\ ()-_]', re.M | re.U)
WHITESPACE_REGEX = re.compile('( |\t|\n)+', re.M | re.U)
class Timing(NamedTuple): # pylint: disable=missing-class-docstring
engine: str
total: float
load: float
class UnresponsiveEngine(NamedTuple): # pylint: disable=missing-class-docstring
engine: str
error_type: str
suspended: bool
def result_content_len(content):
"""Return the meaningful length of the content for a result."""
if isinstance(content, str):
return len(CONTENT_LEN_IGNORED_CHARS_REGEX.sub('', content))
return 0
def result_score(result):
weight = 1.0
for result_engine in result['engines']:
if hasattr(engines[result_engine], 'weight'):
weight *= float(engines[result_engine].weight)
occurences = len(result['positions'])
return sum((occurences * weight) / position for position in result['positions'])
def compare_urls(url_a, url_b):
"""Lazy compare between two URL.
"www.example.com" and "example.com" are equals.
"www.example.com/path/" and "www.example.com/path" are equals.
"https://www.example.com/" and "http://www.example.com/" are equals.
Args:
url_a (ParseResult): first URL
url_b (ParseResult): second URL
Returns:
bool: True if url_a and url_b are equals
"""
# ignore www. in comparison
if url_a.netloc.startswith('www.'):
host_a = url_a.netloc.replace('www.', '', 1)
else:
host_a = url_a.netloc
if url_b.netloc.startswith('www.'):
host_b = url_b.netloc.replace('www.', '', 1)
else:
host_b = url_b.netloc
if host_a != host_b or url_a.query != url_b.query or url_a.fragment != url_b.fragment:
return False
# remove / from the end of the url if required
path_a = url_a.path[:-1] if url_a.path.endswith('/') else url_a.path
path_b = url_b.path[:-1] if url_b.path.endswith('/') else url_b.path
return unquote(path_a) == unquote(path_b)

47
searx/results/correction.py Normal file
View file

@ -0,0 +1,47 @@
# SPDX-License-Identifier: AGPL-3.0-or-later
# lint: pylint
"""Correction item in the result list. The correction result item is used in
the :origin:`results.html <searx/templates/simple/results.html>` template.
A correction item is a dictionary type with dedicated keys and values.
.. code:: python
results.append({
'correction' : str,
})
The context ``corrections`` of the HTML template is a set of dictionaries:
.. code:: python
corrections = [
{
'url' : str,
'title' : str,
},
{...},
...
]
url : ``str``
The SearXNG search URL for the correction term.
title : ``str``
The 'correction' string append by the engine.
"""
def is_correction(result):
"""Returns ``True`` if result type is :py:obj:`.correction`, otherwise
``False``
In the result list a answer item is identified by the existence of the key
``correction``.
"""
return 'correction' in result
class Corrections(set):
"""Set of corrections in the :py:obj:`.container.ResultContainer`"""

165
searx/results/infobox.py Normal file
View file

@ -0,0 +1,165 @@
# SPDX-License-Identifier: AGPL-3.0-or-later
# lint: pylint
"""Infobox item in the result list. The infobox result item is used in the
:origin:`infobox.html <searx/templates/simple/infobox.html>` template.
A infobox item is a dictionary type with dedicated keys and values.
.. code:: python
results.append({
'infobox' : str,
'id' : str,
'content' : str,
'img_src' : str,
'urls' : [url, ...],
'attributes' : [attribute, ...],
'relatedTopics' : [topic, ...],
'engine' : engine,
})
infobox : ``str``
Name of the infobox (mandatory).
id : ``str``
URL of the infobox. Will be used to merge infoboxes.
content : ``str``
Content of the infobox (the description)
img_src :
URL of the image to show in the infobox
urls : ``[url, ...]``
A list of dictionaries with links shown in the infobox. A **url** item in the
``infobox.urls`` list is a dicticonary:
.. code:: python
url = {
'title' : str,
'url' : str,
'entity' : str, # set by some engines but unused
'official' : bool, # set by some engines but unused (oscar)
}
attributes : ``[attribute, ...]``
A **attribute** item in the ``infobox.attributes`` list is a dictionary:
.. code:: python
attribute = {
'label' : str,
'value' : str,
'image' : {
'src': str,
'alt': str,
},
'entity' : str, # set by some engines but unused
}
relatedTopics : ``[topic, ...]``
A **topic** item in the ``infobox.relatedTopics`` list is a dictionary:
.. code:: python
topic = {
'suggestion' : str,
'name' : str, # set by some engines but unused
}
"""
from urllib.parse import urlparse
from searx.engines import engines
from .core import (
result_content_len,
compare_urls,
)
def is_infobox(result):
"""Returns ``True`` if result type is :py:obj:`.infobox`, otherwise ``False``
In the result list a infobox item is identified by the existence of the key
``infobox``.
"""
return 'infobox' in result
class Infoboxes(list):
"""List of infobox items in the :py:obj:`.container.ResultContainer`"""
def merge_two_infoboxes(infobox1, infobox2):
# pylint: disable=too-many-branches, too-many-statements
# get engines weights
if hasattr(engines[infobox1['engine']], 'weight'):
weight1 = engines[infobox1['engine']].weight
else:
weight1 = 1
if hasattr(engines[infobox2['engine']], 'weight'):
weight2 = engines[infobox2['engine']].weight
else:
weight2 = 1
if weight2 > weight1:
infobox1['engine'] = infobox2['engine']
infobox1['engines'] |= infobox2['engines']
if 'urls' in infobox2:
urls1 = infobox1.get('urls', None)
if urls1 is None:
urls1 = []
for url2 in infobox2.get('urls', []):
unique_url = True
parsed_url2 = urlparse(url2.get('url', ''))
entity_url2 = url2.get('entity')
for url1 in urls1:
if (entity_url2 is not None and url1.get('entity') == entity_url2) or compare_urls(
urlparse(url1.get('url', '')), parsed_url2
):
unique_url = False
break
if unique_url:
urls1.append(url2)
infobox1['urls'] = urls1
if 'img_src' in infobox2:
img1 = infobox1.get('img_src', None)
img2 = infobox2.get('img_src')
if img1 is None:
infobox1['img_src'] = img2
elif weight2 > weight1:
infobox1['img_src'] = img2
if 'attributes' in infobox2:
attributes1 = infobox1.get('attributes')
if attributes1 is None:
infobox1['attributes'] = attributes1 = []
attributeSet = set()
for attribute in attributes1:
label = attribute.get('label')
if label not in attributeSet:
attributeSet.add(label)
entity = attribute.get('entity')
if entity not in attributeSet:
attributeSet.add(entity)
for attribute in infobox2.get('attributes', []):
if attribute.get('label') not in attributeSet and attribute.get('entity') not in attributeSet:
attributes1.append(attribute)
if 'content' in infobox2:
content1 = infobox1.get('content', None)
content2 = infobox2.get('content', '')
if content1 is not None:
if result_content_len(content2) > result_content_len(content1):
infobox1['content'] = content2
else:
infobox1['content'] = content2

49
searx/results/suggestion.py Normal file
View file

@ -0,0 +1,49 @@
# SPDX-License-Identifier: AGPL-3.0-or-later
# lint: pylint
"""Suggestion item in the result list. The suggestion result item is used in
the :origin:`infobox.html <searx/templates/simple/results.html>` template.
A sugestion item is a dictionary type with dedicated keys and values.
.. code:: python
results.append({
'suggestion' : str,
})
The context ``suggestions`` of the HTML template is a set of dictionaries:
.. code:: python
suggestions = [
{
'url' : str,
'title' : str,
},
{...},
...
]
url : ``str``
The SearXNG search URL for the suggestion term.
title : ``str``
The 'suggestion' string append by the engine.
"""
from typing import Set
def is_suggestion(result):
"""Returns ``True`` if result type is :py:obj:`.suggestion`, otherwise
``False``
In the result list a suggestion item is identified by the existence of the
key ``suggestion``.
"""
return 'suggestion' in result
class Suggestions(Set):
"""Set of suggestions in the :py:obj:`.container.ResultContainer`"""