psycodad
/
wttr.in
mirror of https://github.com/chubin/wttr.in
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix formatting in lib/wttr_srv.py

Browse Source
pull/1062/head
Igor Chubin 1 month ago
parent 6b7ca7d60a
commit 6964912fb0
1 changed files with 145 additions and 94 deletions
Show Stats Download Patch File Download Diff File

239
lib/wttr_srv.py
Unescape View File

@ -15,14 +15,27 @@ from flask import render_template, send_file, make_response
import fmt.png
import parse_query
from translations import get_message, FULL_TRANSLATION, PARTIAL_TRANSLATION, SUPPORTED_LANGS
from translations import (
get_message,
FULL_TRANSLATION,
PARTIAL_TRANSLATION,
SUPPORTED_LANGS,
)
from buttons import add_buttons
from globals import get_help_file, remove_ansi, TRANSLATION_TABLE, \
BASH_FUNCTION_FILE, TRANSLATION_FILE, LOG_FILE, \
NOT_FOUND_LOCATION, \
MALFORMED_RESPONSE_HTML_PAGE, \
PLAIN_TEXT_AGENTS, PLAIN_TEXT_PAGES, \
MY_EXTERNAL_IP, QUERY_LIMITS
from globals import (
get_help_file,
remove_ansi,
TRANSLATION_TABLE,
BASH_FUNCTION_FILE,
TRANSLATION_FILE,
LOG_FILE,
NOT_FOUND_LOCATION,
MALFORMED_RESPONSE_HTML_PAGE,
PLAIN_TEXT_AGENTS,
PLAIN_TEXT_PAGES,
MY_EXTERNAL_IP,
QUERY_LIMITS,
)
from location import is_location_blocked, location_processing
from limits import Limits
from view.wttr import get_wetter
@ -33,49 +46,53 @@ import cache
if not os.path.exists(os.path.dirname(LOG_FILE)):
os.makedirs(os.path.dirname(LOG_FILE))
logging.basicConfig(filename=LOG_FILE, level=logging.INFO, format='%(asctime)s %(message)s')
logging.basicConfig(
filename=LOG_FILE, level=logging.INFO, format="%(asctime)s %(message)s"
)
LIMITS = Limits(whitelist=[MY_EXTERNAL_IP], limits=QUERY_LIMITS)
TASKS = ThreadPool(25)
def show_text_file(name, lang):
"""
show static file `name` for `lang`
"""
text = ""
if name == ":help":
text = open(get_help_file(lang), 'r').read()
text = text.replace('FULL_TRANSLATION', ' '.join(FULL_TRANSLATION))
text = text.replace('PARTIAL_TRANSLATION', ' '.join(PARTIAL_TRANSLATION))
text = open(get_help_file(lang), "r").read()
text = text.replace("FULL_TRANSLATION", " ".join(FULL_TRANSLATION))
text = text.replace("PARTIAL_TRANSLATION", " ".join(PARTIAL_TRANSLATION))
elif name == ":bash.function":
text = open(BASH_FUNCTION_FILE, 'r').read()
text = open(BASH_FUNCTION_FILE, "r").read()
elif name == ":iterm2":
text = open("share/iterm2.txt", 'r').read()
text = open("share/iterm2.txt", "r").read()
elif name == ":translation":
text = open(TRANSLATION_FILE, 'r').read()
text = text\
.replace('NUMBER_OF_LANGUAGES', str(len(SUPPORTED_LANGS)))\
.replace('SUPPORTED_LANGUAGES', ' '.join(SUPPORTED_LANGS))
text = open(TRANSLATION_FILE, "r").read()
text = text.replace("NUMBER_OF_LANGUAGES", str(len(SUPPORTED_LANGS))).replace(
"SUPPORTED_LANGUAGES", " ".join(SUPPORTED_LANGS)
)
return text
def _client_ip_address(request):
"""Return client ip address for flask `request`.
"""
"""Return client ip address for flask `request`."""
if request.headers.getlist("X-PNG-Query-For"):
ip_addr = request.headers.getlist("X-PNG-Query-For")[0]
if ip_addr.startswith('::ffff:'):
if ip_addr.startswith("::ffff:"):
ip_addr = ip_addr[7:]
elif request.headers.getlist("X-Forwarded-For"):
ip_addr = request.headers.getlist("X-Forwarded-For")[0]
if ip_addr.startswith('::ffff:'):
if ip_addr.startswith("::ffff:"):
ip_addr = ip_addr[7:]
else:
ip_addr = request.remote_addr
return ip_addr
def _parse_language_header(header):
"""
>>> _parse_language_header("en-US,en;q=0.9")
@ -103,18 +120,18 @@ def _parse_language_header(header):
return locale_q_pairs
def _find_supported_language(accepted_languages):
def supported_langs():
"""Yields all pairs in the Accept-Language header
supported in SUPPORTED_LANGS or None if 'en' is the preferred"""
for lang_tuple in accepted_languages:
lang = lang_tuple[0]
if '-' in lang:
lang = lang.split('-', 1)[0]
if "-" in lang:
lang = lang.split("-", 1)[0]
if lang in SUPPORTED_LANGS:
yield lang, lang_tuple[1]
elif lang == 'en':
elif lang == "en":
yield None, lang_tuple[1]
try:
return max(supported_langs(), key=lambda lang_tuple: lang_tuple[1])[0]
except ValueError:
@ -122,6 +139,7 @@ def _parse_language_header(header):
return _find_supported_language(_parse_accept_language(header))
def get_answer_language_and_view(request):
"""
Return preferred answer language based on
@ -130,24 +148,25 @@ def get_answer_language_and_view(request):
lang = None
view_name = None
hostname = request.headers['Host']
if hostname != 'wttr.in' and hostname.endswith('.wttr.in'):
hostname = request.headers["Host"]
if hostname != "wttr.in" and hostname.endswith(".wttr.in"):
lang = hostname[:-8]
if lang.startswith("v2") or lang.startswith("v3"):
view_name = lang
lang = None
if 'lang' in request.args:
lang = request.args.get('lang')
if lang.lower() == 'none':
if "lang" in request.args:
lang = request.args.get("lang")
if lang.lower() == "none":
lang = None
header_accept_language = request.headers.get('Accept-Language', '')
header_accept_language = request.headers.get("Accept-Language", "")
if lang is None and header_accept_language:
lang = _parse_language_header(header_accept_language)
return lang, view_name
def get_output_format(query, parsed_query):
"""
Return preferred output format: ansi, text, html or png
@ -155,23 +174,28 @@ def get_output_format(query, parsed_query):
Return new location (can be rewritten)
"""
if ('view' in query
if (
(
"view" in query
and not query["view"].startswith("v2")
and not query["view"].startswith("v3")) \
or parsed_query.get("png_filename") \
or query.get('force-ansi'):
and not query["view"].startswith("v3")
)
or parsed_query.get("png_filename")
or query.get("force-ansi")
):
return False
user_agent = parsed_query.get("user_agent", "").lower()
html_output = not any(agent in user_agent for agent in PLAIN_TEXT_AGENTS)
return html_output
def _cyclic_location_selection(locations, period):
"""Return one of `locations` (: separated list)
basing on the current time and query interval `period`
"""
locations = locations.split(':')
locations = locations.split(":")
max_len = max(len(x) for x in locations)
locations = [x.rjust(max_len) for x in locations]
@ -180,7 +204,7 @@ def _cyclic_location_selection(locations, period):
except ValueError:
period = 1
index = int(time.time()/period) % len(locations)
index = int(time.time() / period) % len(locations)
return locations[index]
@ -195,31 +219,32 @@ def _response(parsed_query, query, fast_mode=False):
parsed_query["user_agent"],
parsed_query["request_url"],
parsed_query["ip_addr"],
parsed_query["lang"])
parsed_query["lang"],
)
answer = cache.get(cache_signature)
if parsed_query['orig_location'] in PLAIN_TEXT_PAGES:
answer = show_text_file(parsed_query['orig_location'], parsed_query['lang'])
if parsed_query['html_output']:
answer = render_template('index.html', body=answer)
if parsed_query["orig_location"] in PLAIN_TEXT_PAGES:
answer = show_text_file(parsed_query["orig_location"], parsed_query["lang"])
if parsed_query["html_output"]:
answer = render_template("index.html", body=answer)
if answer or fast_mode:
return answer
# at this point, we could not handle the query fast,
# so we handle it with all available logic
loc = (parsed_query['orig_location'] or "").lower()
loc = (parsed_query["orig_location"] or "").lower()
if parsed_query.get("view"):
if not parsed_query.get("location"):
parsed_query["location"] = loc
output = wttr_line(query, parsed_query)
elif loc == 'moon' or loc.startswith('moon@'):
elif loc == "moon" or loc.startswith("moon@"):
output = get_moon(parsed_query)
else:
output = get_wetter(parsed_query)
if parsed_query.get('png_filename'):
if parsed_query.get("png_filename"):
if parsed_query.get("view") != "v3":
# originally it was just a usual function call,
# but it was a blocking call, so it was moved
@ -227,24 +252,29 @@ def _response(parsed_query, query, fast_mode=False):
#
# output = fmt.png.render_ansi(
# output, options=parsed_query)
result = TASKS.spawn(fmt.png.render_ansi, cache._update_answer(output), options=parsed_query)
result = TASKS.spawn(
fmt.png.render_ansi, cache._update_answer(output), options=parsed_query
)
output = result.get()
else:
if query.get('days', '3') != '0' \
and not query.get('no-follow-line') \
and ((parsed_query.get("view") or "v2")[:2] in ["v2", "v3"]):
if parsed_query['html_output']:
if (
query.get("days", "3") != "0"
and not query.get("no-follow-line")
and ((parsed_query.get("view") or "v2")[:2] in ["v2", "v3"])
):
if parsed_query["html_output"]:
output = add_buttons(output)
else:
message = get_message('FOLLOW_ME', parsed_query['lang'])
if parsed_query.get('no-terminal', False):
message = get_message("FOLLOW_ME", parsed_query["lang"])
if parsed_query.get("no-terminal", False):
message = remove_ansi(message)
if parsed_query.get('dumb', False):
if parsed_query.get("dumb", False):
message = message.translate(TRANSLATION_TABLE)
output += '\n' + message + '\n'
output += "\n" + message + "\n"
return cache.store(cache_signature, output)
def parse_request(location, request, query, fast_mode=False):
"""Parse request and provided extended information for the query,
including location data, language, output format, view, etc.
@ -279,14 +309,14 @@ def parse_request(location, request, query, fast_mode=False):
if location is not None and location.lower().endswith(".png"):
png_filename = location
location = location[:-4]
if location and ':' in location and location[0] != ":":
location = _cyclic_location_selection(location, query.get('period', 1))
if location and ":" in location and location[0] != ":":
location = _cyclic_location_selection(location, query.get("period", 1))
parsed_query = {
'ip_addr': _client_ip_address(request),
'user_agent': request.headers.get('User-Agent', '').lower(),
'request_url': request.url,
}
"ip_addr": _client_ip_address(request),
"user_agent": request.headers.get("User-Agent", "").lower(),
"request_url": request.url,
}
if png_filename:
parsed_query["png_filename"] = png_filename
@ -302,24 +332,35 @@ def parse_request(location, request, query, fast_mode=False):
parsed_query["html_output"] = get_output_format(query, parsed_query)
parsed_query["json_output"] = (parsed_query.get("view", "") or "").startswith("j")
if not fast_mode: # not png_filename and not fast_mode:
location, override_location_name, full_address, country, query_source_location, hemisphere = \
location_processing(parsed_query["location"], parsed_query["ip_addr"])
us_ip = query_source_location[2] in ["United States", "United States of America"] \
and 'slack' not in parsed_query['user_agent']
if not fast_mode: # not png_filename and not fast_mode:
(
location,
override_location_name,
full_address,
country,
query_source_location,
hemisphere,
) = location_processing(parsed_query["location"], parsed_query["ip_addr"])
us_ip = (
query_source_location[2] in ["United States", "United States of America"]
and "slack" not in parsed_query["user_agent"]
)
query = parse_query.metric_or_imperial(query, lang, us_ip=us_ip)
if country and location != NOT_FOUND_LOCATION:
location = "%s,%s" % (location, country)
parsed_query.update({
'location': location,
'override_location_name': override_location_name,
'full_address': full_address,
'country': country,
'query_source_location': query_source_location,
'hemisphere': hemisphere})
parsed_query.update(
{
"location": location,
"override_location_name": override_location_name,
"full_address": full_address,
"country": country,
"query_source_location": query_source_location,
"hemisphere": hemisphere,
}
)
parsed_query.update(query)
return parsed_query
@ -332,21 +373,23 @@ def wttr(location, request):
"""
def _wrap_response(response_text, html_output, json_output, png_filename=None):
if not isinstance(response_text, str) and \
not isinstance(response_text, bytes):
if not isinstance(response_text, str) and not isinstance(response_text, bytes):
return response_text
if png_filename:
response = make_response(send_file(
io.BytesIO(response_text),
attachment_filename=png_filename,
mimetype='image/png'))
response = make_response(
send_file(
io.BytesIO(response_text),
attachment_filename=png_filename,
mimetype="image/png",
)
)
for key, value in {
'Cache-Control': 'no-cache, no-store, must-revalidate',
'Pragma': 'no-cache',
'Expires': '0',
}.items():
"Cache-Control": "no-cache, no-store, must-revalidate",
"Pragma": "no-cache",
"Expires": "0",
}.items():
response.headers[key] = value
else:
response = make_response(response_text)
@ -359,12 +402,12 @@ def wttr(location, request):
return response
if is_location_blocked(location):
return ("", 403) # Forbidden
return ("", 403) # Forbidden
try:
LIMITS.check_ip(_client_ip_address(request))
except RuntimeError as exception:
return (str(exception), 429) # Too many requests
return (str(exception), 429) # Too many requests
query = parse_query.parse_query(request.args)
@ -408,29 +451,37 @@ def wttr(location, request):
if not response:
parsed_query = parse_request(location, request, query)
response = _response(parsed_query, query)
#if not response or (isinstance(response, str) and not response.strip()):
# if not response or (isinstance(response, str) and not response.strip()):
# return RuntimeError("Empty answer")
if parsed_query["location"] == NOT_FOUND_LOCATION:
http_code = 404
# pylint: disable=broad-except
except Exception as exception:
except Exception:
logging.error("Exception has occured", exc_info=1)
if parsed_query['html_output']:
if parsed_query["html_output"]:
response = MALFORMED_RESPONSE_HTML_PAGE
http_code = 500 # Internal Server Error
http_code = 500 # Internal Server Error
else:
response = get_message('CAPACITY_LIMIT_REACHED', parsed_query['lang'])
http_code = 503 # Service Unavailable
response = get_message("CAPACITY_LIMIT_REACHED", parsed_query["lang"])
http_code = 503 # Service Unavailable
# if exception is occured, we return not a png file but text
if "png_filename" in parsed_query:
del parsed_query["png_filename"]
return (_wrap_response(
response, parsed_query['html_output'], parsed_query['json_output'],
png_filename=parsed_query.get('png_filename')), http_code)
return (
_wrap_response(
response,
parsed_query["html_output"],
parsed_query["json_output"],
png_filename=parsed_query.get("png_filename"),
),
http_code,
)
if __name__ == "__main__":
import doctest
doctest.testmod()

Write Preview
Loading…
Cancel
Save