#!/usr/bin/env python
# SPDX-License-Identifier: AGPL-3.0-or-later
""" WebbApp
"""
# pylint: disable=use-dict-literal
from __future__ import annotations
import hashlib
import hmac
import json
import os
import sys
import base64
from timeit import default_timer
from html import escape
from io import StringIO
import typing
import urllib
import urllib . parse
from urllib . parse import urlencode , urlparse , unquote
import warnings
import httpx
from pygments import highlight
from pygments . lexers import get_lexer_by_name
from pygments . formatters import HtmlFormatter # pylint: disable=no-name-in-module
import flask
from flask import (
Flask ,
render_template ,
url_for ,
make_response ,
redirect ,
send_from_directory ,
)
from flask . wrappers import Response
from flask . json import jsonify
from flask_babel import (
Babel ,
gettext ,
format_decimal ,
)
from searx . extended_types import sxng_request
from searx import (
logger ,
get_setting ,
settings ,
searx_debug ,
)
from searx import infopage
from searx import limiter
from searx . botdetection import link_token
from searx . data import ENGINE_DESCRIPTIONS
from searx . result_types import Answer
from searx . settings_defaults import OUTPUT_FORMATS
from searx . settings_loader import DEFAULT_SETTINGS_FILE
from searx . exceptions import SearxParameterException
from searx . engines import (
DEFAULT_CATEGORY ,
categories ,
engines ,
engine_shortcuts ,
)
from searx import webutils
from searx . webutils import (
highlight_content ,
get_static_files ,
get_result_templates ,
get_themes ,
exception_classname_to_text ,
new_hmac ,
is_hmac_of ,
is_flask_run_cmdline ,
group_engines_in_tab ,
)
from searx . webadapter import (
get_search_query_from_webapp ,
get_selected_categories ,
parse_lang ,
)
from searx . utils import gen_useragent , dict_subset
from searx . version import VERSION_STRING , GIT_URL , GIT_BRANCH
from searx . query import RawTextQuery
from searx . plugins . oa_doi_rewrite import get_doi_resolver
from searx . preferences import (
Preferences ,
ClientPref ,
ValidationException ,
)
import searx . answerers
import searx . plugins
from searx . metrics import get_engines_stats , get_engine_errors , get_reliabilities , histogram , counter , openmetrics
from searx . flaskfix import patch_application
from searx . locales import (
LOCALE_NAMES ,
RTL_LOCALES ,
localeselector ,
locales_initialize ,
match_locale ,
)
# renaming names from searx imports ...
from searx . autocomplete import search_autocomplete , backends as autocomplete_backends
from searx import favicons
from searx . redisdb import initialize as redis_initialize
from searx . sxng_locales import sxng_locales
import searx . search
from searx . network import stream as http_stream , set_context_network_name
from searx . search . checker import get_result as checker_get_result
logger = logger . getChild ( ' webapp ' )
warnings . simplefilter ( " always " )
# check secret_key
if not searx_debug and settings [ ' server ' ] [ ' secret_key ' ] == ' ultrasecretkey ' :
logger . error ( ' server.secret_key is not changed. Please use something else instead of ultrasecretkey. ' )
sys . exit ( 1 )
# about static
logger . debug ( ' static directory is %s ' , settings [ ' ui ' ] [ ' static_path ' ] )
static_files = get_static_files ( settings [ ' ui ' ] [ ' static_path ' ] )
# about templates
logger . debug ( ' templates directory is %s ' , settings [ ' ui ' ] [ ' templates_path ' ] )
default_theme = settings [ ' ui ' ] [ ' default_theme ' ]
templates_path = settings [ ' ui ' ] [ ' templates_path ' ]
themes = get_themes ( templates_path )
result_templates = get_result_templates ( templates_path )
STATS_SORT_PARAMETERS = {
' name ' : ( False , ' name ' , ' ' ) ,
' score ' : ( True , ' score_per_result ' , 0 ) ,
' result_count ' : ( True , ' result_count ' , 0 ) ,
' time ' : ( False , ' total ' , 0 ) ,
' reliability ' : ( False , ' reliability ' , 100 ) ,
}
# Flask app
app = Flask ( __name__ , static_folder = settings [ ' ui ' ] [ ' static_path ' ] , template_folder = templates_path )
app . jinja_env . trim_blocks = True
app . jinja_env . lstrip_blocks = True
app . jinja_env . add_extension ( ' jinja2.ext.loopcontrols ' ) # pylint: disable=no-member
app . jinja_env . filters [ ' group_engines_in_tab ' ] = group_engines_in_tab # pylint: disable=no-member
app . secret_key = settings [ ' server ' ] [ ' secret_key ' ]
def get_locale ( ) :
locale = localeselector ( )
logger . debug ( " %s uses locale ` %s ` " , urllib . parse . quote ( sxng_request . url ) , locale )
return locale
babel = Babel ( app , locale_selector = get_locale )
def _get_browser_language ( req , lang_list ) :
client = ClientPref . from_http_request ( req )
locale = match_locale ( client . locale_tag , lang_list , fallback = ' en ' )
return locale
def _get_locale_rfc5646 ( locale ) :
""" Get locale name for <html lang= " ... " >
Chrom* browsers don ' t detect the language when there is a subtag (ie a territory).
For example " zh-TW " is detected but not " zh-Hant-TW " .
This function returns a locale without the subtag.
"""
parts = locale . split ( ' - ' )
return parts [ 0 ] . lower ( ) + ' - ' + parts [ - 1 ] . upper ( )
# code-highlighter
@app.template_filter ( ' code_highlighter ' )
def code_highlighter ( codelines , language = None ) :
if not language :
language = ' text '
try :
# find lexer by programming language
lexer = get_lexer_by_name ( language , stripall = True )
except Exception as e : # pylint: disable=broad-except
logger . warning ( " pygments lexer: %s " % e )
# if lexer is not found, using default one
lexer = get_lexer_by_name ( ' text ' , stripall = True )
html_code = ' '
tmp_code = ' '
last_line = None
line_code_start = None
# parse lines
for line , code in codelines :
if not last_line :
line_code_start = line
# new codeblock is detected
if last_line is not None and last_line + 1 != line :
# highlight last codepart
formatter = HtmlFormatter ( linenos = ' inline ' , linenostart = line_code_start , cssclass = " code-highlight " )
html_code = html_code + highlight ( tmp_code , lexer , formatter )
# reset conditions for next codepart
tmp_code = ' '
line_code_start = line
# add codepart
tmp_code + = code + ' \n '
# update line
last_line = line
# highlight last codepart
formatter = HtmlFormatter ( linenos = ' inline ' , linenostart = line_code_start , cssclass = " code-highlight " )
html_code = html_code + highlight ( tmp_code , lexer , formatter )
return html_code
def get_result_template ( theme_name : str , template_name : str ) :
themed_path = theme_name + ' /result_templates/ ' + template_name
if themed_path in result_templates :
return themed_path
return ' result_templates/ ' + template_name
def custom_url_for ( endpoint : str , * * values ) :
suffix = " "
if endpoint == ' static ' and values . get ( ' filename ' ) :
file_hash = static_files . get ( values [ ' filename ' ] )
if not file_hash :
# try file in the current theme
theme_name = sxng_request . preferences . get_value ( ' theme ' )
filename_with_theme = " themes/ {} / {} " . format ( theme_name , values [ ' filename ' ] )
file_hash = static_files . get ( filename_with_theme )
if file_hash :
values [ ' filename ' ] = filename_with_theme
if get_setting ( ' ui.static_use_hash ' ) and file_hash :
suffix = " ? " + file_hash
if endpoint == ' info ' and ' locale ' not in values :
locale = sxng_request . preferences . get_value ( ' locale ' )
if infopage . INFO_PAGES . get_page ( values [ ' pagename ' ] , locale ) is None :
locale = infopage . INFO_PAGES . locale_default
values [ ' locale ' ] = locale
return url_for ( endpoint , * * values ) + suffix
def morty_proxify ( url : str ) :
if url . startswith ( ' // ' ) :
url = ' https: ' + url
if not settings [ ' result_proxy ' ] [ ' url ' ] :
return url
url_params = dict ( mortyurl = url )
if settings [ ' result_proxy ' ] [ ' key ' ] :
url_params [ ' mortyhash ' ] = hmac . new ( settings [ ' result_proxy ' ] [ ' key ' ] , url . encode ( ) , hashlib . sha256 ) . hexdigest ( )
return ' {0} ? {1} ' . format ( settings [ ' result_proxy ' ] [ ' url ' ] , urlencode ( url_params ) )
def image_proxify ( url : str ) :
if url . startswith ( ' // ' ) :
url = ' https: ' + url
if not sxng_request . preferences . get_value ( ' image_proxy ' ) :
return url
if url . startswith ( ' data:image/ ' ) :
# 50 is an arbitrary number to get only the beginning of the image.
partial_base64 = url [ len ( ' data:image/ ' ) : 50 ] . split ( ' ; ' )
if (
len ( partial_base64 ) == 2
and partial_base64 [ 0 ] in [ ' gif ' , ' png ' , ' jpeg ' , ' pjpeg ' , ' webp ' , ' tiff ' , ' bmp ' ]
and partial_base64 [ 1 ] . startswith ( ' base64, ' )
) :
return url
return None
if settings [ ' result_proxy ' ] [ ' url ' ] :
return morty_proxify ( url )
h = new_hmac ( settings [ ' server ' ] [ ' secret_key ' ] , url . encode ( ) )
return ' {0} ? {1} ' . format ( url_for ( ' image_proxy ' ) , urlencode ( dict ( url = url . encode ( ) , h = h ) ) )
def get_translations ( ) :
return {
# when there is autocompletion
' no_item_found ' : gettext ( ' No item found ' ) ,
# /preferences: the source of the engine description (wikipedata, wikidata, website)
' Source ' : gettext ( ' Source ' ) ,
# infinite scroll
' error_loading_next_page ' : gettext ( ' Error loading the next page ' ) ,
}
def get_enabled_categories ( category_names : typing . Iterable [ str ] ) :
""" The categories in ``category_names```for which there is no active engine
are filtered out and a reduced list is returned. """
enabled_engines = [ item [ 0 ] for item in sxng_request . preferences . engines . get_enabled ( ) ]
enabled_categories = set ( )
for engine_name in enabled_engines :
enabled_categories . update ( engines [ engine_name ] . categories )
return [ x for x in category_names if x in enabled_categories ]
def get_pretty_url ( parsed_url : urllib . parse . ParseResult ) :
url_formatting_pref = sxng_request . preferences . get_value ( ' url_formatting ' )
if url_formatting_pref == ' full ' :
return [ parsed_url . geturl ( ) ]
if url_formatting_pref == ' host ' :
return [ parsed_url . netloc ]
path = parsed_url . path
path = path [ : - 1 ] if len ( path ) > 0 and path [ - 1 ] == ' / ' else path
path = unquote ( path . replace ( " / " , " › " ) )
return [ parsed_url . scheme + " :// " + parsed_url . netloc , path ]
def get_client_settings ( ) :
req_pref = sxng_request . preferences
return {
' autocomplete ' : req_pref . get_value ( ' autocomplete ' ) ,
' autocomplete_min ' : get_setting ( ' search.autocomplete_min ' ) ,
' method ' : req_pref . get_value ( ' method ' ) ,
' infinite_scroll ' : req_pref . get_value ( ' infinite_scroll ' ) ,
' translations ' : get_translations ( ) ,
' search_on_category_select ' : req_pref . get_value ( ' search_on_category_select ' ) ,
' hotkeys ' : req_pref . get_value ( ' hotkeys ' ) ,
' url_formatting ' : req_pref . get_value ( ' url_formatting ' ) ,
' theme_static_path ' : custom_url_for ( ' static ' , filename = ' themes/simple ' ) ,
' results_on_new_tab ' : req_pref . get_value ( ' results_on_new_tab ' ) ,
' favicon_resolver ' : req_pref . get_value ( ' favicon_resolver ' ) ,
' advanced_search ' : req_pref . get_value ( ' advanced_search ' ) ,
' query_in_title ' : req_pref . get_value ( ' query_in_title ' ) ,
' safesearch ' : str ( req_pref . get_value ( ' safesearch ' ) ) ,
' theme ' : req_pref . get_value ( ' theme ' ) ,
' doi_resolver ' : get_doi_resolver ( req_pref ) ,
}
def render ( template_name : str , * * kwargs ) :
# values from the preferences
# pylint: disable=too-many-statements
client_settings = get_client_settings ( )
kwargs [ ' client_settings ' ] = str (
base64 . b64encode (
bytes (
json . dumps ( client_settings ) ,
encoding = ' utf-8 ' ,
)
) ,
encoding = ' utf-8 ' ,
)
kwargs [ ' preferences ' ] = sxng_request . preferences
kwargs . update ( client_settings )
# values from the HTTP requests
kwargs [ ' endpoint ' ] = ' results ' if ' q ' in kwargs else sxng_request . endpoint
kwargs [ ' cookies ' ] = sxng_request . cookies
kwargs [ ' errors ' ] = sxng_request . errors
kwargs [ ' link_token ' ] = link_token . get_token ( )
kwargs [ ' categories_as_tabs ' ] = list ( settings [ ' categories_as_tabs ' ] . keys ( ) )
kwargs [ ' categories ' ] = get_enabled_categories ( settings [ ' categories_as_tabs ' ] . keys ( ) )
kwargs [ ' DEFAULT_CATEGORY ' ] = DEFAULT_CATEGORY
# i18n
kwargs [ ' sxng_locales ' ] = [ l for l in sxng_locales if l [ 0 ] in settings [ ' search ' ] [ ' languages ' ] ]
locale = sxng_request . preferences . get_value ( ' locale ' )
kwargs [ ' locale_rfc5646 ' ] = _get_locale_rfc5646 ( locale )
if locale in RTL_LOCALES and ' rtl ' not in kwargs :
kwargs [ ' rtl ' ] = True
if ' current_language ' not in kwargs :
kwargs [ ' current_language ' ] = parse_lang ( sxng_request . preferences , { } , RawTextQuery ( ' ' , [ ] ) )
# values from settings
kwargs [ ' search_formats ' ] = [ x for x in settings [ ' search ' ] [ ' formats ' ] if x != ' html ' ]
kwargs [ ' instance_name ' ] = get_setting ( ' general.instance_name ' )
kwargs [ ' searx_version ' ] = VERSION_STRING
kwargs [ ' searx_git_url ' ] = GIT_URL
kwargs [ ' enable_metrics ' ] = get_setting ( ' general.enable_metrics ' )
kwargs [ ' get_setting ' ] = get_setting
kwargs [ ' get_pretty_url ' ] = get_pretty_url
# values from settings: donation_url
donation_url = get_setting ( ' general.donation_url ' )
if donation_url is True :
donation_url = custom_url_for ( ' info ' , pagename = ' donate ' )
kwargs [ ' donation_url ' ] = donation_url
# helpers to create links to other pages
kwargs [ ' url_for ' ] = custom_url_for # override url_for function in templates
kwargs [ ' image_proxify ' ] = image_proxify
kwargs [ ' favicon_url ' ] = favicons . favicon_url
kwargs [ ' proxify ' ] = morty_proxify if settings [ ' result_proxy ' ] [ ' url ' ] is not None else None
kwargs [ ' proxify_results ' ] = settings [ ' result_proxy ' ] [ ' proxify_results ' ]
kwargs [ ' cache_url ' ] = settings [ ' ui ' ] [ ' cache_url ' ]
kwargs [ ' get_result_template ' ] = get_result_template
kwargs [ ' opensearch_url ' ] = (
url_for ( ' opensearch ' )
+ ' ? '
+ urlencode (
{
' method ' : sxng_request . preferences . get_value ( ' method ' ) ,
' autocomplete ' : sxng_request . preferences . get_value ( ' autocomplete ' ) ,
}
)
)
kwargs [ ' urlparse ' ] = urlparse
start_time = default_timer ( )
result = render_template ( ' {} / {} ' . format ( kwargs [ ' theme ' ] , template_name ) , * * kwargs )
sxng_request . render_time + = default_timer ( ) - start_time # pylint: disable=assigning-non-slot
return result
@app.before_request
def pre_request ( ) :
sxng_request . start_time = default_timer ( ) # pylint: disable=assigning-non-slot
sxng_request . render_time = 0 # pylint: disable=assigning-non-slot
sxng_request . timings = [ ] # pylint: disable=assigning-non-slot
sxng_request . errors = [ ] # pylint: disable=assigning-non-slot
client_pref = ClientPref . from_http_request ( sxng_request )
# pylint: disable=redefined-outer-name
preferences = Preferences ( themes , list ( categories . keys ( ) ) , engines , searx . plugins . STORAGE , client_pref )
user_agent = sxng_request . headers . get ( ' User-Agent ' , ' ' ) . lower ( )
if ' webkit ' in user_agent and ' android ' in user_agent :
preferences . key_value_settings [ ' method ' ] . value = ' GET '
sxng_request . preferences = preferences # pylint: disable=assigning-non-slot
try :
preferences . parse_dict ( sxng_request . cookies )
except Exception as e : # pylint: disable=broad-except
logger . exception ( e , exc_info = True )
sxng_request . errors . append ( gettext ( ' Invalid settings, please edit your preferences ' ) )
# merge GET, POST vars
# HINT request.form is of type werkzeug.datastructures.ImmutableMultiDict
sxng_request . form = dict ( sxng_request . form . items ( ) ) # type: ignore
for k , v in sxng_request . args . items ( ) :
if k not in sxng_request . form :
sxng_request . form [ k ] = v
if sxng_request . form . get ( ' preferences ' ) :
preferences . parse_encoded_data ( sxng_request . form [ ' preferences ' ] )
else :
try :
preferences . parse_dict ( sxng_request . form )
except Exception as e : # pylint: disable=broad-except
logger . exception ( e , exc_info = True )
sxng_request . errors . append ( gettext ( ' Invalid settings ' ) )
# language is defined neither in settings nor in preferences
# use browser headers
if not preferences . get_value ( " language " ) :
language = _get_browser_language ( sxng_request , settings [ ' search ' ] [ ' languages ' ] )
preferences . parse_dict ( { " language " : language } )
logger . debug ( ' set language %s (from browser) ' , preferences . get_value ( " language " ) )
# UI locale is defined neither in settings nor in preferences
# use browser headers
if not preferences . get_value ( " locale " ) :
locale = _get_browser_language ( sxng_request , LOCALE_NAMES . keys ( ) )
preferences . parse_dict ( { " locale " : locale } )
logger . debug ( ' set locale %s (from browser) ' , preferences . get_value ( " locale " ) )
# request.user_plugins
sxng_request . user_plugins = [ ] # pylint: disable=assigning-non-slot
allowed_plugins = preferences . plugins . get_enabled ( )
disabled_plugins = preferences . plugins . get_disabled ( )
for plugin in searx . plugins . STORAGE :
if ( plugin . id not in disabled_plugins ) or plugin . id in allowed_plugins :
sxng_request . user_plugins . append ( plugin . id )
@app.after_request
def add_default_headers ( response : flask . Response ) :
# set default http headers
for header , value in settings [ ' server ' ] [ ' default_http_headers ' ] . items ( ) :
if header in response . headers :
continue
response . headers [ header ] = value
return response
@app.after_request
def post_request ( response : flask . Response ) :
total_time = default_timer ( ) - sxng_request . start_time
timings_all = [
' total;dur= ' + str ( round ( total_time * 1000 , 3 ) ) ,
' render;dur= ' + str ( round ( sxng_request . render_time * 1000 , 3 ) ) ,
]
if len ( sxng_request . timings ) > 0 :
timings = sorted ( sxng_request . timings , key = lambda t : t . total )
timings_total = [
' total_ ' + str ( i ) + ' _ ' + t . engine + ' ;dur= ' + str ( round ( t . total * 1000 , 3 ) ) for i , t in enumerate ( timings )
]
timings_load = [
' load_ ' + str ( i ) + ' _ ' + t . engine + ' ;dur= ' + str ( round ( t . load * 1000 , 3 ) )
for i , t in enumerate ( timings )
if t . load
]
timings_all = timings_all + timings_total + timings_load
response . headers . add ( ' Server-Timing ' , ' , ' . join ( timings_all ) )
return response
def index_error ( output_format : str , error_message : str ) :
if output_format == ' json ' :
return Response ( json . dumps ( { ' error ' : error_message } ) , mimetype = ' application/json ' )
if output_format == ' csv ' :
response = Response ( ' ' , mimetype = ' application/csv ' )
cont_disp = ' attachment;Filename=searx.csv '
response . headers . add ( ' Content-Disposition ' , cont_disp )
return response
if output_format == ' rss ' :
response_rss = render (
' opensearch_response_rss.xml ' ,
results = [ ] ,
q = sxng_request . form [ ' q ' ] if ' q ' in sxng_request . form else ' ' ,
number_of_results = 0 ,
error_message = error_message ,
)
return Response ( response_rss , mimetype = ' text/xml ' )
# html
sxng_request . errors . append ( gettext ( ' search error ' ) )
return render (
# fmt: off
' index.html ' ,
selected_categories = get_selected_categories ( sxng_request . preferences , sxng_request . form ) ,
# fmt: on
)
@app.route ( ' / ' , methods = [ ' GET ' , ' POST ' ] )
def index ( ) :
""" Render index page. """
# redirect to search if there's a query in the request
if sxng_request . form . get ( ' q ' ) :
query = ( ' ? ' + sxng_request . query_string . decode ( ) ) if sxng_request . query_string else ' '
return redirect ( url_for ( ' search ' ) + query , 308 )
return render (
# fmt: off
' index.html ' ,
selected_categories = get_selected_categories ( sxng_request . preferences , sxng_request . form ) ,
current_locale = sxng_request . preferences . get_value ( " locale " ) ,
# fmt: on
)
@app.route ( ' /healthz ' , methods = [ ' GET ' ] )
def health ( ) :
return Response ( ' OK ' , mimetype = ' text/plain ' )
@app.route ( ' /client<token>.css ' , methods = [ ' GET ' , ' POST ' ] )
def client_token ( token = None ) :
link_token . ping ( sxng_request , token )
return Response ( ' ' , mimetype = ' text/css ' )
@app.route ( ' /rss.xsl ' , methods = [ ' GET ' , ' POST ' ] )
def rss_xsl ( ) :
return render_template (
f " { sxng_request . preferences . get_value ( ' theme ' ) } /rss.xsl " ,
url_for = custom_url_for ,
)
@app.route ( ' /search ' , methods = [ ' GET ' , ' POST ' ] )
def search ( ) :
""" Search query in q and return results.
Supported outputs: html, json, csv, rss.
"""
# pylint: disable=too-many-locals, too-many-return-statements, too-many-branches
# pylint: disable=too-many-statements
# output_format
output_format = sxng_request . form . get ( ' format ' , ' html ' )
if output_format not in OUTPUT_FORMATS :
output_format = ' html '
if output_format not in settings [ ' search ' ] [ ' formats ' ] :
flask . abort ( 403 )
# check if there is query (not None and not an empty string)
if not sxng_request . form . get ( ' q ' ) :
if output_format == ' html ' :
return render (
# fmt: off
' index.html ' ,
selected_categories = get_selected_categories ( sxng_request . preferences , sxng_request . form ) ,
# fmt: on
)
return index_error ( output_format , ' No query ' ) , 400
# search
search_query = None
raw_text_query = None
result_container = None
try :
search_query , raw_text_query , _ , _ , selected_locale = get_search_query_from_webapp (
sxng_request . preferences , sxng_request . form
)
search_obj = searx . search . SearchWithPlugins ( search_query , sxng_request , sxng_request . user_plugins )
result_container = search_obj . search ( )
except SearxParameterException as e :
logger . exception ( ' search error: SearxParameterException ' )
return index_error ( output_format , e . message ) , 400
except Exception as e : # pylint: disable=broad-except
logger . exception ( e , exc_info = True )
return index_error ( output_format , gettext ( ' search error ' ) ) , 500
# 1. check if the result is a redirect for an external bang
if result_container . redirect_url :
return redirect ( result_container . redirect_url )
# 2. add Server-Timing header for measuring performance characteristics of
# web applications
sxng_request . timings = result_container . get_timings ( ) # pylint: disable=assigning-non-slot
# 3. formats without a template
if output_format == ' json ' :
response = webutils . get_json_response ( search_query , result_container )
return Response ( response , mimetype = ' application/json ' )
if output_format == ' csv ' :
csv = webutils . CSVWriter ( StringIO ( ) )
webutils . write_csv_response ( csv , result_container )
csv . stream . seek ( 0 )
response = Response ( csv . stream . read ( ) , mimetype = ' application/csv ' )
cont_disp = ' attachment;Filename=searx_-_ {0} .csv ' . format ( search_query . query )
response . headers . add ( ' Content-Disposition ' , cont_disp )
return response
# 4. formats rendered by a template / RSS & HTML
current_template = None
previous_result = None
results = result_container . get_ordered_results ( )
if search_query . redirect_to_first_result and results :
return redirect ( results [ 0 ] [ ' url ' ] , 302 )
for result in results :
if output_format == ' html ' :
if ' content ' in result and result [ ' content ' ] :
result [ ' content ' ] = highlight_content ( escape ( result [ ' content ' ] [ : 1024 ] ) , search_query . query )
if ' title ' in result and result [ ' title ' ] :
result [ ' title ' ] = highlight_content ( escape ( result [ ' title ' ] or ' ' ) , search_query . query )
if getattr ( result , ' publishedDate ' , None ) : # do not try to get a date from an empty string or a None type
try : # test if publishedDate >= 1900 (datetime module bug)
result [ ' pubdate ' ] = result [ ' publishedDate ' ] . strftime ( ' % Y- % m- %d % H: % M: % S % z ' )
except ValueError :
result [ ' publishedDate ' ] = None
else :
result [ ' publishedDate ' ] = webutils . searxng_l10n_timespan ( result [ ' publishedDate ' ] )
# set result['open_group'] = True when the template changes from the previous result
# set result['close_group'] = True when the template changes on the next result
if current_template != result . template :
result . open_group = True
if previous_result :
previous_result . close_group = True # pylint: disable=unsupported-assignment-operation
current_template = result . template
previous_result = result
if previous_result :
previous_result . close_group = True
# 4.a RSS
if output_format == ' rss ' :
response_rss = render (
' opensearch_response_rss.xml ' ,
results = results ,
q = sxng_request . form [ ' q ' ] ,
number_of_results = result_container . number_of_results ,
)
return Response ( response_rss , mimetype = ' text/xml ' )
# 4.b HTML
# suggestions: use RawTextQuery to get the suggestion URLs with the same bang
suggestion_urls = list (
map (
lambda suggestion : { ' url ' : raw_text_query . changeQuery ( suggestion ) . getFullQuery ( ) , ' title ' : suggestion } ,
result_container . suggestions ,
)
)
correction_urls = list (
map (
lambda correction : { ' url ' : raw_text_query . changeQuery ( correction ) . getFullQuery ( ) , ' title ' : correction } ,
result_container . corrections ,
)
)
# engine_timings: get engine response times sorted from slowest to fastest
engine_timings = sorted ( result_container . get_timings ( ) , reverse = True , key = lambda e : e . total )
max_response_time = engine_timings [ 0 ] . total if engine_timings else None
engine_timings_pairs = [ ( timing . engine , timing . total ) for timing in engine_timings ]
# search_query.lang contains the user choice (all, auto, en, ...)
# when the user choice is "auto", search.search_query.lang contains the detected language
# otherwise it is equals to search_query.lang
return render (
# fmt: off
' results.html ' ,
results = results ,
q = sxng_request . form [ ' q ' ] ,
selected_categories = search_query . categories ,
pageno = search_query . pageno ,
time_range = search_query . time_range or ' ' ,
number_of_results = format_decimal ( result_container . number_of_results ) ,
suggestions = suggestion_urls ,
answers = result_container . answers ,
corrections = correction_urls ,
infoboxes = result_container . infoboxes ,
engine_data = result_container . engine_data ,
paging = result_container . paging ,
unresponsive_engines = webutils . get_translated_errors (
result_container . unresponsive_engines
) ,
current_locale = sxng_request . preferences . get_value ( " locale " ) ,
current_language = selected_locale ,
search_language = match_locale (
search_obj . search_query . lang ,
settings [ ' search ' ] [ ' languages ' ] ,
fallback = sxng_request . preferences . get_value ( " language " )
) ,
timeout_limit = sxng_request . form . get ( ' timeout_limit ' , None ) ,
timings = engine_timings_pairs ,
max_response_time = max_response_time
# fmt: on
)
@app.route ( ' /about ' , methods = [ ' GET ' ] )
def about ( ) :
""" Redirect to about page """
# custom_url_for is going to add the locale
return redirect ( custom_url_for ( ' info ' , pagename = ' about ' ) )
@app.route ( ' /info/<locale>/<pagename> ' , methods = [ ' GET ' ] )
def info ( pagename , locale ) :
""" Render page of online user documentation """
page = infopage . INFO_PAGES . get_page ( pagename , locale )
if page is None :
flask . abort ( 404 )
user_locale = sxng_request . preferences . get_value ( ' locale ' )
return render (
' info.html ' ,
all_pages = infopage . INFO_PAGES . iter_pages ( user_locale , fallback_to_default = True ) ,
active_page = page ,
active_pagename = pagename ,
)
@app.route ( ' /autocompleter ' , methods = [ ' GET ' , ' POST ' ] )
def autocompleter ( ) :
""" Return autocompleter results """
# run autocompleter
results = [ ]
# set blocked engines
disabled_engines = sxng_request . preferences . engines . get_disabled ( )
# parse query
raw_text_query = RawTextQuery ( sxng_request . form . get ( ' q ' , ' ' ) , disabled_engines )
sug_prefix = raw_text_query . getQuery ( )
for obj in searx . answerers . STORAGE . ask ( sug_prefix ) :
if isinstance ( obj , Answer ) :
results . append ( obj . answer )
# normal autocompletion results only appear if no inner results returned
# and there is a query part
if len ( raw_text_query . autocomplete_list ) == 0 and len ( sug_prefix ) > 0 :
# get SearXNG's locale and autocomplete backend from cookie
sxng_locale = sxng_request . preferences . get_value ( ' language ' )
backend_name = sxng_request . preferences . get_value ( ' autocomplete ' )
for result in search_autocomplete ( backend_name , sug_prefix , sxng_locale ) :
# attention: this loop will change raw_text_query object and this is
# the reason why the sug_prefix was stored before (see above)
if result != sug_prefix :
results . append ( raw_text_query . changeQuery ( result ) . getFullQuery ( ) )
if len ( raw_text_query . autocomplete_list ) > 0 :
for autocomplete_text in raw_text_query . autocomplete_list :
results . append ( raw_text_query . get_autocomplete_full_query ( autocomplete_text ) )
if sxng_request . headers . get ( ' X-Requested-With ' ) == ' XMLHttpRequest ' :
# the suggestion request comes from the searx search form
suggestions = json . dumps ( results )
mimetype = ' application/json '
else :
# the suggestion request comes from browser's URL bar
suggestions = json . dumps ( [ sug_prefix , results ] )
mimetype = ' application/x-suggestions+json '
suggestions = escape ( suggestions , False )
return Response ( suggestions , mimetype = mimetype )
@app.route ( ' /preferences ' , methods = [ ' GET ' , ' POST ' ] )
def preferences ( ) :
""" Render preferences page && save user preferences """
# pylint: disable=too-many-locals, too-many-return-statements, too-many-branches
# pylint: disable=too-many-statements
# save preferences using the link the /preferences?preferences=...
if sxng_request . args . get ( ' preferences ' ) or sxng_request . form . get ( ' preferences ' ) :
resp = make_response ( redirect ( url_for ( ' index ' , _external = True ) ) )
return sxng_request . preferences . save ( resp )
# save preferences
if sxng_request . method == ' POST ' :
resp = make_response ( redirect ( url_for ( ' index ' , _external = True ) ) )
try :
sxng_request . preferences . parse_form ( sxng_request . form )
except ValidationException :
sxng_request . errors . append ( gettext ( ' Invalid settings, please edit your preferences ' ) )
return resp
return sxng_request . preferences . save ( resp )
# render preferences
image_proxy = sxng_request . preferences . get_value ( ' image_proxy ' ) # pylint: disable=redefined-outer-name
disabled_engines = sxng_request . preferences . engines . get_disabled ( )
allowed_plugins = sxng_request . preferences . plugins . get_enabled ( )
# stats for preferences page
filtered_engines = dict ( filter ( lambda kv : sxng_request . preferences . validate_token ( kv [ 1 ] ) , engines . items ( ) ) )
engines_by_category = { }
for c in categories : # pylint: disable=consider-using-dict-items
engines_by_category [ c ] = [ e for e in categories [ c ] if e . name in filtered_engines ]
# sort the engines alphabetically since the order in settings.yml is meaningless.
list . sort ( engines_by_category [ c ] , key = lambda e : e . name )
# get first element [0], the engine time,
# and then the second element [1] : the time (the first one is the label)
stats = { } # pylint: disable=redefined-outer-name
max_rate95 = 0
for _ , e in filtered_engines . items ( ) :
h = histogram ( ' engine ' , e . name , ' time ' , ' total ' )
median = round ( h . percentage ( 50 ) , 1 ) if h . count > 0 else None
rate80 = round ( h . percentage ( 80 ) , 1 ) if h . count > 0 else None
rate95 = round ( h . percentage ( 95 ) , 1 ) if h . count > 0 else None
max_rate95 = max ( max_rate95 , rate95 or 0 )
result_count_sum = histogram ( ' engine ' , e . name , ' result ' , ' count ' ) . sum
successful_count = counter ( ' engine ' , e . name , ' search ' , ' count ' , ' successful ' )
result_count = int ( result_count_sum / float ( successful_count ) ) if successful_count else 0
stats [ e . name ] = {
' time ' : median ,
' rate80 ' : rate80 ,
' rate95 ' : rate95 ,
' warn_timeout ' : e . timeout > settings [ ' outgoing ' ] [ ' request_timeout ' ] ,
' supports_selected_language ' : e . traits . is_locale_supported (
str ( sxng_request . preferences . get_value ( ' language ' ) or ' all ' )
) ,
' result_count ' : result_count ,
}
# end of stats
# reliabilities
reliabilities = { }
engine_errors = get_engine_errors ( filtered_engines )
checker_results = checker_get_result ( )
checker_results = (
checker_results [ ' engines ' ] if checker_results [ ' status ' ] == ' ok ' and ' engines ' in checker_results else { }
)
for _ , e in filtered_engines . items ( ) :
checker_result = checker_results . get ( e . name , { } )
checker_success = checker_result . get ( ' success ' , True )
errors = engine_errors . get ( e . name ) or [ ]
if counter ( ' engine ' , e . name , ' search ' , ' count ' , ' sent ' ) == 0 :
# no request
reliability = None
elif checker_success and not errors :
reliability = 100
elif ' simple ' in checker_result . get ( ' errors ' , { } ) :
# the basic (simple) test doesn't work: the engine is broken according to the checker
# even if there is no exception
reliability = 0
else :
# pylint: disable=consider-using-generator
reliability = 100 - sum ( [ error [ ' percentage ' ] for error in errors if not error . get ( ' secondary ' ) ] )
reliabilities [ e . name ] = {
' reliability ' : reliability ,
' errors ' : [ ] ,
' checker ' : checker_results . get ( e . name , { } ) . get ( ' errors ' , { } ) . keys ( ) ,
}
# keep the order of the list checker_results[e.name]['errors'] and deduplicate.
# the first element has the highest percentage rate.
reliabilities_errors = [ ]
for error in errors :
error_user_text = None
if error . get ( ' secondary ' ) or ' exception_classname ' not in error :
continue
error_user_text = exception_classname_to_text . get ( error . get ( ' exception_classname ' ) )
if not error :
error_user_text = exception_classname_to_text [ None ]
if error_user_text not in reliabilities_errors :
reliabilities_errors . append ( error_user_text )
reliabilities [ e . name ] [ ' errors ' ] = reliabilities_errors
# supports
supports = { }
for _ , e in filtered_engines . items ( ) :
supports_selected_language = e . traits . is_locale_supported (
str ( sxng_request . preferences . get_value ( ' language ' ) or ' all ' )
)
safesearch = e . safesearch
time_range_support = e . time_range_support
for checker_test_name in checker_results . get ( e . name , { } ) . get ( ' errors ' , { } ) :
if supports_selected_language and checker_test_name . startswith ( ' lang_ ' ) :
supports_selected_language = ' ? '
elif safesearch and checker_test_name == ' safesearch ' :
safesearch = ' ? '
elif time_range_support and checker_test_name == ' time_range ' :
time_range_support = ' ? '
supports [ e . name ] = {
' supports_selected_language ' : supports_selected_language ,
' safesearch ' : safesearch ,
' time_range_support ' : time_range_support ,
}
return render (
# fmt: off
' preferences.html ' ,
preferences = True ,
selected_categories = get_selected_categories ( sxng_request . preferences , sxng_request . form ) ,
locales = LOCALE_NAMES ,
current_locale = sxng_request . preferences . get_value ( " locale " ) ,
image_proxy = image_proxy ,
engines_by_category = engines_by_category ,
stats = stats ,
max_rate95 = max_rate95 ,
reliabilities = reliabilities ,
supports = supports ,
answer_storage = searx . answerers . STORAGE . info ,
disabled_engines = disabled_engines ,
autocomplete_backends = autocomplete_backends ,
favicon_resolver_names = favicons . proxy . CFG . resolver_map . keys ( ) ,
shortcuts = { y : x for x , y in engine_shortcuts . items ( ) } ,
themes = themes ,
plugins_storage = searx . plugins . STORAGE . info ,
current_doi_resolver = get_doi_resolver ( sxng_request . preferences ) ,
allowed_plugins = allowed_plugins ,
preferences_url_params = sxng_request . preferences . get_as_url_params ( ) ,
locked_preferences = get_setting ( " preferences.lock " , [ ] ) ,
doi_resolvers = get_setting ( " doi_resolvers " , { } ) ,
# fmt: on
)
app . add_url_rule ( ' /favicon_proxy ' , methods = [ ' GET ' ] , endpoint = " favicon_proxy " , view_func = favicons . favicon_proxy )
@app.route ( ' /image_proxy ' , methods = [ ' GET ' ] )
def image_proxy ( ) :
# pylint: disable=too-many-return-statements, too-many-branches
url = sxng_request . args . get ( ' url ' )
if not url :
return ' ' , 400
if not is_hmac_of ( settings [ ' server ' ] [ ' secret_key ' ] , url . encode ( ) , sxng_request . args . get ( ' h ' , ' ' ) ) :
return ' ' , 400
maximum_size = 5 * 1024 * 1024
forward_resp = False
resp = None
try :
request_headers = {
' User-Agent ' : gen_useragent ( ) ,
' Accept ' : ' image/webp,*/* ' ,
' Accept-Encoding ' : ' gzip, deflate ' ,
' Sec-GPC ' : ' 1 ' ,
' DNT ' : ' 1 ' ,
}
set_context_network_name ( ' image_proxy ' )
resp , stream = http_stream ( method = ' GET ' , url = url , headers = request_headers , allow_redirects = True )
content_length = resp . headers . get ( ' Content-Length ' )
if content_length and content_length . isdigit ( ) and int ( content_length ) > maximum_size :
return ' Max size ' , 400
if resp . status_code != 200 :
logger . debug ( ' image-proxy: wrong response code: %i ' , resp . status_code )
if resp . status_code > = 400 :
return ' ' , resp . status_code
return ' ' , 400
if not resp . headers . get ( ' Content-Type ' , ' ' ) . startswith ( ' image/ ' ) and not resp . headers . get (
' Content-Type ' , ' '
) . startswith ( ' binary/octet-stream ' ) :
logger . debug ( ' image-proxy: wrong content-type: %s ' , resp . headers . get ( ' Content-Type ' , ' ' ) )
return ' ' , 400
forward_resp = True
except httpx . HTTPError :
logger . exception ( ' HTTP error ' )
return ' ' , 400
finally :
if resp and not forward_resp :
# the code is about to return an HTTP 400 error to the browser
# we make sure to close the response between searxng and the HTTP server
try :
resp . close ( )
except httpx . HTTPError :
logger . exception ( ' HTTP error on closing ' )
def close_stream ( ) :
nonlocal resp , stream
try :
if resp :
resp . close ( )
del resp
del stream
except httpx . HTTPError as e :
logger . debug ( ' Exception while closing response ' , e )
try :
headers = dict_subset ( resp . headers , { ' Content-Type ' , ' Content-Encoding ' , ' Content-Length ' , ' Length ' } )
response = Response ( stream , mimetype = resp . headers [ ' Content-Type ' ] , headers = headers , direct_passthrough = True )
response . call_on_close ( close_stream )
return response
except httpx . HTTPError :
close_stream ( )
return ' ' , 400
@app.route ( ' /engine_descriptions.json ' , methods = [ ' GET ' ] )
def engine_descriptions ( ) :
locale = get_locale ( ) . split ( ' _ ' ) [ 0 ]
result = ENGINE_DESCRIPTIONS [ ' en ' ] . copy ( )
if locale != ' en ' :
for engine , description in ENGINE_DESCRIPTIONS . get ( locale , { } ) . items ( ) :
result [ engine ] = description
for engine , description in result . items ( ) :
if len ( description ) == 2 and description [ 1 ] == ' ref ' :
ref_engine , ref_lang = description [ 0 ] . split ( ' : ' )
description = ENGINE_DESCRIPTIONS [ ref_lang ] [ ref_engine ]
if isinstance ( description , str ) :
description = [ description , ' wikipedia ' ]
result [ engine ] = description
# overwrite by about:description (from settings)
for engine_name , engine_mod in engines . items ( ) :
descr = getattr ( engine_mod , ' about ' , { } ) . get ( ' description ' , None )
if descr is not None :
result [ engine_name ] = [ descr , " SearXNG config " ]
return jsonify ( result )
@app.route ( ' /stats ' , methods = [ ' GET ' ] )
def stats ( ) :
""" Render engine statistics page. """
sort_order = sxng_request . args . get ( ' sort ' , default = ' name ' , type = str )
selected_engine_name = sxng_request . args . get ( ' engine ' , default = None , type = str )
filtered_engines = dict ( filter ( lambda kv : sxng_request . preferences . validate_token ( kv [ 1 ] ) , engines . items ( ) ) )
if selected_engine_name :
if selected_engine_name not in filtered_engines :
selected_engine_name = None
else :
filtered_engines = [ selected_engine_name ]
checker_results = checker_get_result ( )
checker_results = (
checker_results [ ' engines ' ] if checker_results [ ' status ' ] == ' ok ' and ' engines ' in checker_results else { }
)
engine_stats = get_engines_stats ( filtered_engines )
engine_reliabilities = get_reliabilities ( filtered_engines , checker_results )
if sort_order not in STATS_SORT_PARAMETERS :
sort_order = ' name '
reverse , key_name , default_value = STATS_SORT_PARAMETERS [ sort_order ]
def get_key ( engine_stat ) :
reliability = engine_reliabilities . get ( engine_stat [ ' name ' ] , { } ) . get ( ' reliability ' , 0 )
reliability_order = 0 if reliability else 1
if key_name == ' reliability ' :
key = reliability
reliability_order = 0
else :
key = engine_stat . get ( key_name ) or default_value
if reverse :
reliability_order = 1 - reliability_order
return ( reliability_order , key , engine_stat [ ' name ' ] )
technical_report = [ ]
for error in engine_reliabilities . get ( selected_engine_name , { } ) . get ( ' errors ' , [ ] ) :
technical_report . append (
f " \
Error: { error [ ' exception_classname ' ] or error [ ' log_message ' ] } \
Parameters: { error [ ' log_parameters ' ] } \
File name: { error [ ' filename ' ] } : { error [ ' line_no ' ] } \
Error Function: { error [ ' function ' ] } \
Code: { error [ ' code ' ] } \
" . replace (
' ' * 12 , ' '
) . strip ( )
)
technical_report = ' ' . join ( technical_report )
engine_stats [ ' time ' ] = sorted ( engine_stats [ ' time ' ] , reverse = reverse , key = get_key )
return render (
# fmt: off
' stats.html ' ,
sort_order = sort_order ,
engine_stats = engine_stats ,
engine_reliabilities = engine_reliabilities ,
selected_engine_name = selected_engine_name ,
searx_git_branch = GIT_BRANCH ,
technical_report = technical_report ,
# fmt: on
)
@app.route ( ' /stats/errors ' , methods = [ ' GET ' ] )
def stats_errors ( ) :
filtered_engines = dict ( filter ( lambda kv : sxng_request . preferences . validate_token ( kv [ 1 ] ) , engines . items ( ) ) )
result = get_engine_errors ( filtered_engines )
return jsonify ( result )
@app.route ( ' /stats/checker ' , methods = [ ' GET ' ] )
def stats_checker ( ) :
result = checker_get_result ( )
return jsonify ( result )
@app.route ( ' /metrics ' )
def stats_open_metrics ( ) :
password = settings [ ' general ' ] . get ( " open_metrics " )
if not ( settings [ ' general ' ] . get ( " enable_metrics " ) and password ) :
return Response ( ' open metrics is disabled ' , status = 404 , mimetype = ' text/plain ' )
if not sxng_request . authorization or sxng_request . authorization . password != password :
return Response ( ' access forbidden ' , status = 401 , mimetype = ' text/plain ' )
filtered_engines = dict ( filter ( lambda kv : sxng_request . preferences . validate_token ( kv [ 1 ] ) , engines . items ( ) ) )
checker_results = checker_get_result ( )
checker_results = (
checker_results [ ' engines ' ] if checker_results [ ' status ' ] == ' ok ' and ' engines ' in checker_results else { }
)
engine_stats = get_engines_stats ( filtered_engines )
engine_reliabilities = get_reliabilities ( filtered_engines , checker_results )
metrics_text = openmetrics ( engine_stats , engine_reliabilities )
return Response ( metrics_text , mimetype = ' text/plain ' )
@app.route ( ' /robots.txt ' , methods = [ ' GET ' ] )
def robots ( ) :
return Response (
""" User-agent: *
Allow: /info/en/about
Disallow: /stats
Disallow: /image_proxy
Disallow: /preferences
Disallow: /*?*q=*
""" ,
mimetype = ' text/plain ' ,
)
@app.route ( ' /opensearch.xml ' , methods = [ ' GET ' ] )
def opensearch ( ) :
method = sxng_request . preferences . get_value ( ' method ' )
autocomplete = sxng_request . preferences . get_value ( ' autocomplete ' )
# chrome/chromium only supports HTTP GET....
if sxng_request . headers . get ( ' User-Agent ' , ' ' ) . lower ( ) . find ( ' webkit ' ) > = 0 :
method = ' GET '
if method not in ( ' POST ' , ' GET ' ) :
method = ' POST '
ret = render ( ' opensearch.xml ' , opensearch_method = method , autocomplete = autocomplete )
resp = Response ( response = ret , status = 200 , mimetype = " application/opensearchdescription+xml " )
return resp
@app.route ( ' /favicon.ico ' )
def favicon ( ) :
theme = sxng_request . preferences . get_value ( " theme " )
return send_from_directory (
os . path . join ( app . root_path , settings [ ' ui ' ] [ ' static_path ' ] , ' themes ' , theme , ' img ' ) , # type: ignore
' favicon.png ' ,
mimetype = ' image/vnd.microsoft.icon ' ,
)
@app.route ( ' /clear_cookies ' )
def clear_cookies ( ) :
resp = make_response ( redirect ( url_for ( ' index ' , _external = True ) ) )
for cookie_name in sxng_request . cookies :
resp . delete_cookie ( cookie_name )
return resp
@app.route ( ' /config ' )
def config ( ) :
""" Return configuration in JSON format. """
_engines = [ ]
for name , engine in engines . items ( ) :
if not sxng_request . preferences . validate_token ( engine ) :
continue
_languages = engine . traits . languages . keys ( )
_engines . append (
{
' name ' : name ,
' categories ' : engine . categories ,
' shortcut ' : engine . shortcut ,
' enabled ' : not engine . disabled ,
' paging ' : engine . paging ,
' language_support ' : engine . language_support ,
' languages ' : list ( _languages ) ,
' regions ' : list ( engine . traits . regions . keys ( ) ) ,
' safesearch ' : engine . safesearch ,
' time_range_support ' : engine . time_range_support ,
' timeout ' : engine . timeout ,
}
)
_plugins = [ ]
for _ in searx . plugins . STORAGE :
_plugins . append ( { ' name ' : _ . id , ' enabled ' : _ . default_on } )
_limiter_cfg = limiter . get_cfg ( )
return jsonify (
{
' categories ' : list ( categories . keys ( ) ) ,
' engines ' : _engines ,
' plugins ' : _plugins ,
' instance_name ' : settings [ ' general ' ] [ ' instance_name ' ] ,
' locales ' : LOCALE_NAMES ,
' default_locale ' : settings [ ' ui ' ] [ ' default_locale ' ] ,
' autocomplete ' : settings [ ' search ' ] [ ' autocomplete ' ] ,
' safe_search ' : settings [ ' search ' ] [ ' safe_search ' ] ,
' default_theme ' : settings [ ' ui ' ] [ ' default_theme ' ] ,
' version ' : VERSION_STRING ,
' brand ' : {
' PRIVACYPOLICY_URL ' : get_setting ( ' general.privacypolicy_url ' ) ,
' CONTACT_URL ' : get_setting ( ' general.contact_url ' ) ,
' GIT_URL ' : GIT_URL ,
' GIT_BRANCH ' : GIT_BRANCH ,
' DOCS_URL ' : get_setting ( ' brand.docs_url ' ) ,
} ,
' limiter ' : {
' enabled ' : limiter . is_installed ( ) ,
' botdetection.ip_limit.link_token ' : _limiter_cfg . get ( ' botdetection.ip_limit.link_token ' ) ,
' botdetection.ip_lists.pass_searxng_org ' : _limiter_cfg . get ( ' botdetection.ip_lists.pass_searxng_org ' ) ,
} ,
' doi_resolvers ' : list ( settings [ ' doi_resolvers ' ] . keys ( ) ) ,
' default_doi_resolver ' : settings [ ' default_doi_resolver ' ] ,
' public_instance ' : settings [ ' server ' ] [ ' public_instance ' ] ,
}
)
@app.errorhandler ( 404 )
def page_not_found ( _e ) :
return render ( ' 404.html ' ) , 404
# see https://flask.palletsprojects.com/en/1.1.x/cli/
# True if "FLASK_APP=searx/webapp.py FLASK_ENV=development flask run"
flask_run_development = (
os . environ . get ( " FLASK_APP " ) is not None and os . environ . get ( " FLASK_ENV " ) == ' development ' and is_flask_run_cmdline ( )
)
# True if reload feature is activated of werkzeug, False otherwise (including uwsgi, etc..)
# __name__ != "__main__" if searx.webapp is imported (make test, make docs, uwsgi...)
# see run() at the end of this file : searx_debug activates the reload feature.
werkzeug_reloader = flask_run_development or ( searx_debug and __name__ == " __main__ " )
# initialize the engines except on the first run of the werkzeug server.
if not werkzeug_reloader or ( werkzeug_reloader and os . environ . get ( " WERKZEUG_RUN_MAIN " ) == " true " ) :
locales_initialize ( )
redis_initialize ( )
searx . plugins . initialize ( app )
searx . search . initialize (
enable_checker = True ,
check_network = True ,
enable_metrics = get_setting ( " general.enable_metrics " ) ,
)
limiter . initialize ( app , settings )
favicons . init ( )
def run ( ) :
logger . debug ( ' starting webserver on %s : %s ' , settings [ ' server ' ] [ ' bind_address ' ] , settings [ ' server ' ] [ ' port ' ] )
app . run (
debug = searx_debug ,
use_debugger = searx_debug ,
port = settings [ ' server ' ] [ ' port ' ] ,
host = settings [ ' server ' ] [ ' bind_address ' ] ,
threaded = True ,
extra_files = [ DEFAULT_SETTINGS_FILE ] ,
)
application = app
patch_application ( app )
if __name__ == " __main__ " :
run ( )