impl.py 16 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415
  1. # SPDX-License-Identifier: AGPL-3.0-or-later
  2. import typing
  3. import types
  4. import functools
  5. import itertools
  6. import threading
  7. from time import time
  8. from urllib.parse import urlparse
  9. import re
  10. from langdetect import detect_langs
  11. from langdetect.lang_detect_exception import LangDetectException
  12. import requests.exceptions
  13. from searx import poolrequests, logger
  14. from searx.results import ResultContainer
  15. from searx.search.models import SearchQuery, EngineRef
  16. from searx.search.processors import EngineProcessor
  17. logger = logger.getChild('searx.search.checker')
  18. HTML_TAGS = [
  19. 'embed', 'iframe', 'object', 'param', 'picture', 'source', 'svg', 'math', 'canvas', 'noscript', 'script',
  20. 'del', 'ins', 'area', 'audio', 'img', 'map', 'track', 'video', 'a', 'abbr', 'b', 'bdi', 'bdo', 'br', 'cite',
  21. 'code', 'data', 'dfn', 'em', 'i', 'kdb', 'mark', 'q', 'rb', 'rp', 'rt', 'rtc', 'ruby', 's', 'samp', 'small',
  22. 'span', 'strong', 'sub', 'sup', 'time', 'u', 'var', 'wbr', 'style', 'blockquote', 'dd', 'div', 'dl', 'dt',
  23. 'figcaption', 'figure', 'hr', 'li', 'ol', 'p', 'pre', 'ul', 'button', 'datalist', 'fieldset', 'form', 'input',
  24. 'label', 'legend', 'meter', 'optgroup', 'option', 'output', 'progress', 'select', 'textarea', 'applet',
  25. 'frame', 'frameset'
  26. ]
  27. def get_check_no_html():
  28. rep = ['<' + tag + '[^\>]*>' for tag in HTML_TAGS]
  29. rep += ['</' + tag + '>' for tag in HTML_TAGS]
  30. pattern = re.compile('|'.join(rep))
  31. def f(text):
  32. return pattern.search(text.lower()) is None
  33. return f
  34. _check_no_html = get_check_no_html()
  35. def _is_url(url):
  36. try:
  37. result = urlparse(url)
  38. except ValueError:
  39. return False
  40. if result.scheme not in ('http', 'https'):
  41. return False
  42. return True
  43. @functools.lru_cache(maxsize=8192)
  44. def _is_url_image(image_url):
  45. if not isinstance(image_url, str):
  46. return False
  47. if image_url.startswith('//'):
  48. image_url = 'https:' + image_url
  49. if image_url.startswith('data:'):
  50. return image_url.startswith('data:image/')
  51. if not _is_url(image_url):
  52. return False
  53. retry = 2
  54. while retry > 0:
  55. a = time()
  56. try:
  57. poolrequests.set_timeout_for_thread(10.0, time())
  58. r = poolrequests.get(image_url, timeout=10.0, allow_redirects=True, headers={
  59. 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:84.0) Gecko/20100101 Firefox/84.0',
  60. 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
  61. 'Accept-Language': 'en-US;q=0.5,en;q=0.3',
  62. 'Accept-Encoding': 'gzip, deflate, br',
  63. 'DNT': '1',
  64. 'Connection': 'keep-alive',
  65. 'Upgrade-Insecure-Requests': '1',
  66. 'Sec-GPC': '1',
  67. 'Cache-Control': 'max-age=0'
  68. })
  69. if r.headers["content-type"].startswith('image/'):
  70. return True
  71. return False
  72. except requests.exceptions.Timeout:
  73. logger.error('Timeout for %s: %i', image_url, int(time() - a))
  74. retry -= 1
  75. except requests.exceptions.RequestException:
  76. logger.exception('Exception for %s', image_url)
  77. return False
  78. def _search_query_to_dict(search_query: SearchQuery) -> typing.Dict[str, typing.Any]:
  79. return {
  80. 'query': search_query.query,
  81. 'lang': search_query.lang,
  82. 'pageno': search_query.pageno,
  83. 'safesearch': search_query.safesearch,
  84. 'time_range': search_query.time_range,
  85. }
  86. def _search_query_diff(sq1: SearchQuery, sq2: SearchQuery)\
  87. -> typing.Tuple[typing.Dict[str, typing.Any], typing.Dict[str, typing.Any]]:
  88. param1 = _search_query_to_dict(sq1)
  89. param2 = _search_query_to_dict(sq2)
  90. common = {}
  91. diff = {}
  92. for k, value1 in param1.items():
  93. value2 = param2[k]
  94. if value1 == value2:
  95. common[k] = value1
  96. else:
  97. diff[k] = (value1, value2)
  98. return (common, diff)
  99. class TestResults:
  100. __slots__ = 'errors', 'logs', 'languages'
  101. def __init__(self):
  102. self.errors: typing.Dict[str, typing.List[str]] = {}
  103. self.logs: typing.Dict[str, typing.List[typing.Any]] = {}
  104. self.languages: typing.Set[str] = set()
  105. def add_error(self, test, message, *args):
  106. # message to self.errors
  107. errors_for_test = self.errors.setdefault(test, [])
  108. if message not in errors_for_test:
  109. errors_for_test.append(message)
  110. # (message, *args) to self.logs
  111. logs_for_test = self.logs.setdefault(test, [])
  112. if (message, *args) not in logs_for_test:
  113. logs_for_test.append((message, *args))
  114. def add_language(self, language):
  115. self.languages.add(language)
  116. @property
  117. def succesfull(self):
  118. return len(self.errors) == 0
  119. def __iter__(self):
  120. for test_name, errors in self.errors.items():
  121. for error in sorted(errors):
  122. yield (test_name, error)
  123. class ResultContainerTests:
  124. __slots__ = 'test_name', 'search_query', 'result_container', 'languages', 'stop_test', 'test_results'
  125. def __init__(self,
  126. test_results: TestResults,
  127. test_name: str,
  128. search_query: SearchQuery,
  129. result_container: ResultContainer):
  130. self.test_name = test_name
  131. self.search_query = search_query
  132. self.result_container = result_container
  133. self.languages: typing.Set[str] = set()
  134. self.test_results = test_results
  135. self.stop_test = False
  136. @property
  137. def result_urls(self):
  138. results = self.result_container.get_ordered_results()
  139. return [result['url'] for result in results]
  140. def _record_error(self, message: str, *args) -> None:
  141. sq = _search_query_to_dict(self.search_query)
  142. sqstr = ' '.join(['{}={!r}'.format(k, v) for k, v in sq.items()])
  143. self.test_results.add_error(self.test_name, message, *args, '(' + sqstr + ')')
  144. def _add_language(self, text: str) -> typing.Optional[str]:
  145. try:
  146. r = detect_langs(str(text)) # pylint: disable=E1101
  147. except LangDetectException:
  148. return None
  149. if len(r) > 0 and r[0].prob > 0.95:
  150. self.languages.add(r[0].lang)
  151. self.test_results.add_language(r[0].lang)
  152. return None
  153. def _check_result(self, result):
  154. if not _check_no_html(result.get('title', '')):
  155. self._record_error('HTML in title', repr(result.get('title', '')))
  156. if not _check_no_html(result.get('content', '')):
  157. self._record_error('HTML in content', repr(result.get('content', '')))
  158. self._add_language(result.get('title', ''))
  159. self._add_language(result.get('content', ''))
  160. template = result.get('template', 'default.html')
  161. if template == 'default.html':
  162. return
  163. if template == 'code.html':
  164. return
  165. if template == 'torrent.html':
  166. return
  167. if template == 'map.html':
  168. return
  169. if template == 'images.html':
  170. thumbnail_src = result.get('thumbnail_src')
  171. if thumbnail_src is not None:
  172. if not _is_url_image(thumbnail_src):
  173. self._record_error('thumbnail_src URL is invalid', thumbnail_src)
  174. elif not _is_url_image(result.get('img_src')):
  175. self._record_error('img_src URL is invalid', result.get('img_src'))
  176. if template == 'videos.html' and not _is_url_image(result.get('thumbnail')):
  177. self._record_error('thumbnail URL is invalid', result.get('img_src'))
  178. def _check_results(self, results: list):
  179. for result in results:
  180. self._check_result(result)
  181. def _check_answers(self, answers):
  182. for answer in answers:
  183. if not _check_no_html(answer):
  184. self._record_error('HTML in answer', answer)
  185. def _check_infoboxes(self, infoboxes):
  186. for infobox in infoboxes:
  187. if not _check_no_html(infobox.get('content', '')):
  188. self._record_error('HTML in infobox content', infobox.get('content', ''))
  189. self._add_language(infobox.get('content', ''))
  190. for attribute in infobox.get('attributes', {}):
  191. if not _check_no_html(attribute.get('value', '')):
  192. self._record_error('HTML in infobox attribute value', attribute.get('value', ''))
  193. def check_basic(self):
  194. if len(self.result_container.unresponsive_engines) > 0:
  195. for message in self.result_container.unresponsive_engines:
  196. self._record_error(message[1] + ' ' + (message[2] or ''))
  197. self.stop_test = True
  198. return
  199. results = self.result_container.get_ordered_results()
  200. if len(results) > 0:
  201. self._check_results(results)
  202. if len(self.result_container.answers) > 0:
  203. self._check_answers(self.result_container.answers)
  204. if len(self.result_container.infoboxes) > 0:
  205. self._check_infoboxes(self.result_container.infoboxes)
  206. def has_infobox(self):
  207. """Check the ResultContainer has at least one infobox"""
  208. if len(self.result_container.infoboxes) == 0:
  209. self._record_error('No infobox')
  210. def has_answer(self):
  211. """Check the ResultContainer has at least one answer"""
  212. if len(self.result_container.answers) == 0:
  213. self._record_error('No answer')
  214. def has_language(self, lang):
  215. """Check at least one title or content of the results is written in the `lang`.
  216. Detected using pycld3, may be not accurate"""
  217. if lang not in self.languages:
  218. self._record_error(lang + ' not found')
  219. def not_empty(self):
  220. """Check the ResultContainer has at least one answer or infobox or result"""
  221. result_types = set()
  222. results = self.result_container.get_ordered_results()
  223. if len(results) > 0:
  224. result_types.add('results')
  225. if len(self.result_container.answers) > 0:
  226. result_types.add('answers')
  227. if len(self.result_container.infoboxes) > 0:
  228. result_types.add('infoboxes')
  229. if len(result_types) == 0:
  230. self._record_error('No result')
  231. def one_title_contains(self, title: str):
  232. """Check one of the title contains `title` (case insensitive comparaison)"""
  233. title = title.lower()
  234. for result in self.result_container.get_ordered_results():
  235. if title in result['title'].lower():
  236. return
  237. self._record_error(('{!r} not found in the title'.format(title)))
  238. class CheckerTests:
  239. __slots__ = 'test_results', 'test_name', 'result_container_tests_list'
  240. def __init__(self,
  241. test_results: TestResults,
  242. test_name: str,
  243. result_container_tests_list: typing.List[ResultContainerTests]):
  244. self.test_results = test_results
  245. self.test_name = test_name
  246. self.result_container_tests_list = result_container_tests_list
  247. def unique_results(self):
  248. """Check the results of each ResultContain is unique"""
  249. urls_list = [rct.result_urls for rct in self.result_container_tests_list]
  250. if len(urls_list[0]) > 0:
  251. # results on the first page
  252. for i, urls_i in enumerate(urls_list):
  253. for j, urls_j in enumerate(urls_list):
  254. if i < j and urls_i == urls_j:
  255. common, diff = _search_query_diff(self.result_container_tests_list[i].search_query,
  256. self.result_container_tests_list[j].search_query)
  257. common_str = ' '.join(['{}={!r}'.format(k, v) for k, v in common.items()])
  258. diff1_str = ', ' .join(['{}={!r}'.format(k, v1) for (k, (v1, v2)) in diff.items()])
  259. diff2_str = ', ' .join(['{}={!r}'.format(k, v2) for (k, (v1, v2)) in diff.items()])
  260. self.test_results.add_error(self.test_name,
  261. 'results are identitical for {} and {} ({})'
  262. .format(diff1_str, diff2_str, common_str))
  263. class Checker:
  264. __slots__ = 'processor', 'tests', 'test_results'
  265. def __init__(self, processor: EngineProcessor):
  266. self.processor = processor
  267. self.tests = self.processor.get_tests()
  268. self.test_results = TestResults()
  269. @property
  270. def engineref_list(self):
  271. engine_name = self.processor.engine_name
  272. engine_category = self.processor.engine.categories[0]
  273. return [EngineRef(engine_name, engine_category)]
  274. @staticmethod
  275. def search_query_matrix_iterator(engineref_list, matrix):
  276. p = []
  277. for name, values in matrix.items():
  278. if isinstance(values, (tuple, list)):
  279. l = [(name, value) for value in values]
  280. else:
  281. l = [(name, values)]
  282. p.append(l)
  283. for kwargs in itertools.product(*p):
  284. kwargs = {k: v for k, v in kwargs}
  285. query = kwargs['query']
  286. params = dict(kwargs)
  287. del params['query']
  288. yield SearchQuery(query, engineref_list, **params)
  289. def call_test(self, obj, test_description):
  290. if isinstance(test_description, (tuple, list)):
  291. method, args = test_description[0], test_description[1:]
  292. else:
  293. method = test_description
  294. args = ()
  295. if isinstance(method, str) and hasattr(obj, method):
  296. getattr(obj, method)(*args)
  297. elif isinstance(method, types.FunctionType):
  298. method(*args)
  299. else:
  300. self.test_results.add_error(obj.test_name,
  301. 'method {!r} ({}) not found for {}'
  302. .format(method, method.__class__.__name__, obj.__class__.__name__))
  303. def call_tests(self, obj, test_descriptions):
  304. for test_description in test_descriptions:
  305. self.call_test(obj, test_description)
  306. def search(self, search_query: SearchQuery) -> ResultContainer:
  307. result_container = ResultContainer()
  308. engineref_category = search_query.engineref_list[0].category
  309. params = self.processor.get_params(search_query, engineref_category)
  310. if params is not None:
  311. with threading.RLock():
  312. self.processor.engine.stats['sent_search_count'] += 1
  313. self.processor.search(search_query.query, params, result_container, time(), 5)
  314. return result_container
  315. def get_result_container_tests(self, test_name: str, search_query: SearchQuery) -> ResultContainerTests:
  316. result_container = self.search(search_query)
  317. result_container_check = ResultContainerTests(self.test_results, test_name, search_query, result_container)
  318. result_container_check.check_basic()
  319. return result_container_check
  320. def run_test(self, test_name):
  321. test_parameters = self.tests[test_name]
  322. search_query_list = list(Checker.search_query_matrix_iterator(self.engineref_list, test_parameters['matrix']))
  323. rct_list = [self.get_result_container_tests(test_name, search_query) for search_query in search_query_list]
  324. stop_test = False
  325. if 'result_container' in test_parameters:
  326. for rct in rct_list:
  327. stop_test = stop_test or rct.stop_test
  328. if not rct.stop_test:
  329. self.call_tests(rct, test_parameters['result_container'])
  330. if not stop_test:
  331. if 'test' in test_parameters:
  332. checker_tests = CheckerTests(self.test_results, test_name, rct_list)
  333. self.call_tests(checker_tests, test_parameters['test'])
  334. def run(self):
  335. for test_name in self.tests:
  336. self.run_test(test_name)