13from datetime
import datetime, timedelta
14from typing
import Iterable, List, Tuple, TYPE_CHECKING
16from io
import StringIO
17from codecs
import getincrementalencoder
19from flask_babel
import gettext, format_date
21from searx
import logger, get_setting
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 static_path = pathlib.Path(str(
get_setting(
"ui.static_path")))
184 def _walk(path: pathlib.Path):
185 for f
in path.iterdir():
186 if f.name.startswith(
'.'):
190 file_list.append(str(f.relative_to(static_path)))
199 result_templates = set()
200 templates_path_length = len(templates_path) + 1
201 for directory, _, files
in os.walk(templates_path):
202 if directory.endswith(
'result_templates'):
203 for filename
in files:
204 f = os.path.join(directory[templates_path_length:], filename)
205 result_templates.add(f)
206 return result_templates
210 return hmac.new(secret_key.encode(), url, hashlib.sha256).hexdigest()
214 hmac_of_value =
new_hmac(secret_key, value)
215 return len(hmac_of_value) == len(hmac_to_check)
and hmac.compare_digest(hmac_of_value, hmac_to_check)
219 if len(url) > max_length:
220 chunk_len = int(max_length / 2 + 1)
221 return '{0}[...]{1}'.format(url[:chunk_len], url[-chunk_len:])
226 """This function check whether or not a string contains Chinese, Japanese,
227 or Korean characters. It employs regex and uses the u escape sequence to
228 match any character in a set of Unicode ranges.
231 s (str): string to be checked.
234 bool: True if the input s contains the characters and False otherwise.
244 return bool(re.search(fr
'[{unicode_ranges}]', s))
248 """Generate the regex pattern to match for a given word according
249 to whether or not the word contains CJK characters or not.
250 If the word is and/or contains CJK character, the regex pattern
251 will match standalone word by taking into account the presence
252 of whitespace before and after it; if not, it will match any presence
253 of the word throughout the text, ignoring the whitespace.
256 word (str): the word to be matched with regex pattern.
259 str: the regex pattern for the word.
261 rword = re.escape(word)
264 return fr
'\b({rword})(?!\w)'
273 if content.find(
'<') != -1:
276 querysplit = query.split()
278 for qs
in querysplit:
279 qs = qs.replace(
"'",
"").replace(
'"',
'').replace(
" ",
"")
283 regex = re.compile(
"|".join(map(regex_highlight_cjk, queries)))
284 return regex.sub(
lambda match: f
'<span class="highlight">{match.group(0)}</span>'.replace(
'\\',
r'\\'), content)
289 """Returns a human-readable and translated string indicating how long ago
290 a date was in the past / the time span of the date to the present.
292 On January 1st, midnight, the returned string only indicates how many years
298 if d.month == 1
and d.day == 1
and t.hour == 0
and t.minute == 0
and t.second == 0:
300 if dt.replace(tzinfo=
None) >= datetime.now() - timedelta(days=1):
301 timedifference = datetime.now() - dt.replace(tzinfo=
None)
302 minutes = int((timedifference.seconds / 60) % 60)
303 hours = int(timedifference.seconds / 60 / 60)
305 return gettext(
'{minutes} minute(s) ago').format(minutes=minutes)
306 return gettext(
'{hours} hour(s), {minutes} minute(s) ago').format(hours=hours, minutes=minutes)
307 return format_date(dt)
310NO_SUBGROUPING =
'without further subgrouping'
314 """Groups an Iterable of engines by their first non tab category (first subgroup)"""
316 def get_subgroup(eng):
317 non_tab_categories = [c
for c
in eng.categories
if c
not in tabs + [DEFAULT_CATEGORY]]
318 return non_tab_categories[0]
if len(non_tab_categories) > 0
else NO_SUBGROUPING
320 def group_sort_key(group):
321 return (group[0] == NO_SUBGROUPING, group[0].lower())
323 def engine_sort_key(engine):
324 return (engine.about.get(
'language',
''), engine.name)
326 tabs = list(
get_setting(
'categories_as_tabs').keys())
327 subgroups = itertools.groupby(sorted(engines, key=get_subgroup), get_subgroup)
328 sorted_groups = sorted(((name, list(engines))
for name, engines
in subgroups), key=group_sort_key)
331 for groupname, _engines
in sorted_groups:
332 group_bang =
'!' + groupname.replace(
' ',
'_')
if groupname != NO_SUBGROUPING
else ''
333 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)
prettify_url(url, max_length=74)
List[Tuple[str, "Iterable[Engine]"]] group_engines_in_tab("Iterable[Engine]" engines)
str regex_highlight_cjk(str word)
get_result_templates(templates_path)
new_hmac(secret_key, url)
str get_json_response("SearchQuery" sq, "ResultContainer" rc)
is_hmac_of(secret_key, value, hmac_to_check)
str searxng_l10n_timespan(datetime dt)
highlight_content(content, query)
None write_csv_response(CSVWriter csv, "ResultContainer" rc)
get_translated_errors("Iterable[UnresponsiveEngine]" unresponsive_engines)
bool contains_cjko(str s)
list[str] get_static_file_list()
t.Any get_setting(str name, t.Any default=_unset)