4from __future__
import annotations
14from datetime
import datetime, timedelta
15from typing
import Iterable, List, Tuple, Dict, TYPE_CHECKING
17from io
import StringIO
18from codecs
import getincrementalencoder
20from flask_babel
import gettext, format_date
22from searx
import logger, settings
31VALID_LANGUAGE_CODE = re.compile(
r'^[a-z]{2,3}(-[a-zA-Z]{2})?$')
33logger = logger.getChild(
'webutils')
35timeout_text = gettext(
'timeout')
36parsing_error_text = gettext(
'parsing error')
37http_protocol_error_text = gettext(
'HTTP protocol error')
38network_error_text = gettext(
'network error')
39ssl_cert_error_text = gettext(
"SSL error: certificate validation has failed")
40exception_classname_to_text = {
41 None: gettext(
'unexpected crash'),
42 'timeout': timeout_text,
43 'asyncio.TimeoutError': timeout_text,
44 'httpx.TimeoutException': timeout_text,
45 'httpx.ConnectTimeout': timeout_text,
46 'httpx.ReadTimeout': timeout_text,
47 'httpx.WriteTimeout': timeout_text,
48 'httpx.HTTPStatusError': gettext(
'HTTP error'),
49 'httpx.ConnectError': gettext(
"HTTP connection error"),
50 'httpx.RemoteProtocolError': http_protocol_error_text,
51 'httpx.LocalProtocolError': http_protocol_error_text,
52 'httpx.ProtocolError': http_protocol_error_text,
53 'httpx.ReadError': network_error_text,
54 'httpx.WriteError': network_error_text,
55 'httpx.ProxyError': gettext(
"proxy error"),
56 'searx.exceptions.SearxEngineCaptchaException': gettext(
"CAPTCHA"),
57 'searx.exceptions.SearxEngineTooManyRequestsException': gettext(
"too many requests"),
58 'searx.exceptions.SearxEngineAccessDeniedException': gettext(
"access denied"),
59 'searx.exceptions.SearxEngineAPIException': gettext(
"server API error"),
60 'searx.exceptions.SearxEngineXPathException': parsing_error_text,
61 'KeyError': parsing_error_text,
62 'json.decoder.JSONDecodeError': parsing_error_text,
63 'lxml.etree.ParserError': parsing_error_text,
64 'ssl.SSLCertVerificationError': ssl_cert_error_text,
65 'ssl.CertificateError': ssl_cert_error_text,
70 translated_errors = []
72 for unresponsive_engine
in unresponsive_engines:
73 error_user_text = exception_classname_to_text.get(unresponsive_engine.error_type)
74 if not error_user_text:
75 error_user_text = exception_classname_to_text[
None]
76 error_msg = gettext(error_user_text)
77 if unresponsive_engine.suspended:
78 error_msg = gettext(
'Suspended') +
': ' + error_msg
79 translated_errors.append((unresponsive_engine.engine, error_msg))
81 return sorted(translated_errors, key=
lambda e: e[0])
85 """A CSV writer which will write rows to CSV file "f", which is encoded in
86 the given encoding."""
88 def __init__(self, f, dialect=csv.excel, encoding="utf-8", **kwds):
93 self.
encoder = getincrementalencoder(encoding)()
98 data = self.
queue.getvalue()
99 data = data.strip(
'\x00')
101 data = self.
encoder.encode(data)
103 self.
stream.write(data.decode())
105 self.
queue.truncate(0)
113 """Write rows of the results to a query (``application/csv``) into a CSV
114 table (:py:obj:`CSVWriter`). First line in the table contain the column
115 names. The column "type" specifies the type, the following types are
116 included in the table:
125 keys = (
'title',
'url',
'content',
'host',
'engine',
'score',
'type')
128 for res
in rc.get_ordered_results():
130 row[
'host'] = row[
'parsed_url'].netloc
131 row[
'type'] =
'result'
132 csv.writerow([row.get(key,
'')
for key
in keys])
136 row[
'host'] = row[
'parsed_url'].netloc
137 csv.writerow([row.get(key,
'')
for key
in keys])
139 for a
in rc.suggestions:
140 row = {
'title': a,
'type':
'suggestion'}
141 csv.writerow([row.get(key,
'')
for key
in keys])
143 for a
in rc.corrections:
144 row = {
'title': a,
'type':
'correction'}
145 csv.writerow([row.get(key,
'')
for key
in keys])
150 if isinstance(o, datetime):
152 if isinstance(o, timedelta):
153 return o.total_seconds()
154 if isinstance(o, set):
160 """Returns the JSON string of the results to a query (``application/json``)"""
163 'number_of_results': rc.number_of_results,
164 'results': [_.as_dict()
for _
in rc.get_ordered_results()],
165 'answers': [_.as_dict()
for _
in rc.answers],
166 'corrections': list(rc.corrections),
167 'infoboxes': rc.infoboxes,
168 'suggestions': list(rc.suggestions),
171 response = json.dumps(data, cls=JSONEncoder)
176 """Returns available themes list."""
177 return os.listdir(templates_path)
182 with file.open(
'rb')
as f:
188 static_files: Dict[str, str] = {}
189 static_path_path = pathlib.Path(static_path)
191 def walk(path: pathlib.Path):
192 for file
in path.iterdir():
193 if file.name.startswith(
'.'):
198 if file.is_dir()
and file.name
not in (
'node_modules',
'src'):
202 walk(static_path_path)
207 result_templates = set()
208 templates_path_length = len(templates_path) + 1
209 for directory, _, files
in os.walk(templates_path):
210 if directory.endswith(
'result_templates'):
211 for filename
in files:
212 f = os.path.join(directory[templates_path_length:], filename)
213 result_templates.add(f)
214 return result_templates
218 return hmac.new(secret_key.encode(), url, hashlib.sha256).hexdigest()
222 hmac_of_value =
new_hmac(secret_key, value)
223 return len(hmac_of_value) == len(hmac_to_check)
and hmac.compare_digest(hmac_of_value, hmac_to_check)
227 if len(url) > max_length:
228 chunk_len = int(max_length / 2 + 1)
229 return '{0}[...]{1}'.format(url[:chunk_len], url[-chunk_len:])
234 """This function check whether or not a string contains Chinese, Japanese,
235 or Korean characters. It employs regex and uses the u escape sequence to
236 match any character in a set of Unicode ranges.
239 s (str): string to be checked.
242 bool: True if the input s contains the characters and False otherwise.
252 return bool(re.search(fr
'[{unicode_ranges}]', s))
256 """Generate the regex pattern to match for a given word according
257 to whether or not the word contains CJK characters or not.
258 If the word is and/or contains CJK character, the regex pattern
259 will match standalone word by taking into account the presence
260 of whitespace before and after it; if not, it will match any presence
261 of the word throughout the text, ignoring the whitespace.
264 word (str): the word to be matched with regex pattern.
267 str: the regex pattern for the word.
269 rword = re.escape(word)
272 return fr
'\b({rword})(?!\w)'
281 if content.find(
'<') != -1:
284 querysplit = query.split()
286 for qs
in querysplit:
287 qs = qs.replace(
"'",
"").replace(
'"',
'').replace(
" ",
"")
291 regex = re.compile(
"|".join(map(regex_highlight_cjk, queries)))
292 return regex.sub(
lambda match: f
'<span class="highlight">{match.group(0)}</span>'.replace(
'\\',
r'\\'), content)
297 """Returns a human-readable and translated string indicating how long ago
298 a date was in the past / the time span of the date to the present.
300 On January 1st, midnight, the returned string only indicates how many years
306 if d.month == 1
and d.day == 1
and t.hour == 0
and t.minute == 0
and t.second == 0:
308 if dt.replace(tzinfo=
None) >= datetime.now() - timedelta(days=1):
309 timedifference = datetime.now() - dt.replace(tzinfo=
None)
310 minutes = int((timedifference.seconds / 60) % 60)
311 hours = int(timedifference.seconds / 60 / 60)
313 return gettext(
'{minutes} minute(s) ago').format(minutes=minutes)
314 return gettext(
'{hours} hour(s), {minutes} minute(s) ago').format(hours=hours, minutes=minutes)
315 return format_date(dt)
318NO_SUBGROUPING =
'without further subgrouping'
322 """Groups an Iterable of engines by their first non tab category (first subgroup)"""
324 def get_subgroup(eng):
325 non_tab_categories = [c
for c
in eng.categories
if c
not in tabs + [DEFAULT_CATEGORY]]
326 return non_tab_categories[0]
if len(non_tab_categories) > 0
else NO_SUBGROUPING
328 def group_sort_key(group):
329 return (group[0] == NO_SUBGROUPING, group[0].lower())
331 def engine_sort_key(engine):
332 return (engine.about.get(
'language',
''), engine.name)
334 tabs = list(settings[
'categories_as_tabs'].keys())
335 subgroups = itertools.groupby(sorted(engines, key=get_subgroup), get_subgroup)
336 sorted_groups = sorted(((name, list(engines))
for name, engines
in subgroups), key=group_sort_key)
339 for groupname, _engines
in sorted_groups:
340 group_bang =
'!' + groupname.replace(
' ',
'_')
if groupname != NO_SUBGROUPING
else ''
341 ret_val.append((groupname, group_bang, sorted(_engines, key=engine_sort_key)))
__init__(self, f, dialect=csv.excel, encoding="utf-8", **kwds)
get_themes(templates_path)
Dict[str, str] get_static_files(str static_path)
List[Tuple[str, Iterable[Engine]]] group_engines_in_tab(Iterable[Engine] engines)
prettify_url(url, max_length=74)
str regex_highlight_cjk(str word)
get_result_templates(templates_path)
new_hmac(secret_key, url)
is_hmac_of(secret_key, value, hmac_to_check)
get_translated_errors(Iterable[UnresponsiveEngine] unresponsive_engines)
str get_hash_for_file(pathlib.Path file)
str searxng_l10n_timespan(datetime dt)
highlight_content(content, query)
str get_json_response(SearchQuery sq, ResultContainer rc)
None write_csv_response(CSVWriter csv, ResultContainer rc)
bool contains_cjko(str s)