From 08bef9006dbdca0ca8ccfa03bdf331526ad9c8ac Mon Sep 17 00:00:00 2001 From: Igor Chubin Date: Mon, 3 Feb 2025 14:05:24 +0100 Subject: [PATCH] Fix formatting of bin/proxy.py --- bin/proxy.py | 184 +++++++++++++++++++++++++++++++-------------------- 1 file changed, 111 insertions(+), 73 deletions(-) diff --git a/bin/proxy.py b/bin/proxy.py index 02af224..552a88c 100755 --- a/bin/proxy.py +++ b/bin/proxy.py @@ -1,4 +1,4 @@ -#vim: fileencoding=utf-8 +# vim: fileencoding=utf-8 """ @@ -16,6 +16,7 @@ from __future__ import print_function from gevent.pywsgi import WSGIServer from gevent.monkey import patch_all + patch_all() # pylint: disable=wrong-import-position,wrong-import-order @@ -29,27 +30,37 @@ import requests import cyrtranslit from flask import Flask, request + APP = Flask(__name__) -MYDIR = os.path.abspath( - os.path.dirname(os.path.dirname('__file__'))) +MYDIR = os.path.abspath(os.path.dirname(os.path.dirname("__file__"))) sys.path.append("%s/lib/" % MYDIR) import proxy_log import globals -from globals import PROXY_CACHEDIR, PROXY_HOST, PROXY_PORT, USE_METNO, USER_AGENT, MISSING_TRANSLATION_LOG +from globals import ( + PROXY_CACHEDIR, + PROXY_HOST, + PROXY_PORT, + USE_METNO, + USER_AGENT, + MISSING_TRANSLATION_LOG, +) from metno import create_standard_json_from_metno, metno_request from translations import PROXY_LANGS + # pylint: enable=wrong-import-position proxy_logger = proxy_log.LoggerWWO(globals.PROXY_LOG_ACCESS, globals.PROXY_LOG_ERRORS) + def is_testmode(): """Server is running in the wttr.in test mode""" return "WTTRIN_TEST" in os.environ + def load_translations(): """ load all translations @@ -57,32 +68,37 @@ def load_translations(): translations = {} for f_name in PROXY_LANGS: - f_name = 'share/translations/%s.txt' % f_name + f_name = "share/translations/%s.txt" % f_name translation = {} - lang = f_name.split('/')[-1].split('.', 1)[0] + lang = f_name.split("/")[-1].split(".", 1)[0] with open(f_name, "r") as f_file: for line in f_file: - if ':' not in line: + if ":" not in line: continue - if line.count(':') == 3: - _, trans, orig, _ = line.strip().split(':', 4) + if line.count(":") == 3: + _, trans, orig, _ = line.strip().split(":", 4) else: - _, trans, orig = line.strip().split(':', 3) + _, trans, orig = line.strip().split(":", 3) trans = trans.strip() orig = orig.strip() translation[orig.lower()] = trans translations[lang] = translation return translations + + TRANSLATIONS = load_translations() + def _is_metno(): return USE_METNO -def _find_srv_for_query(path, query): # pylint: disable=unused-argument + +def _find_srv_for_query(path, query): # pylint: disable=unused-argument if _is_metno(): - return 'https://api.met.no' - return 'http://api.worldweatheronline.com' + return "https://api.met.no" + return "http://api.worldweatheronline.com" + def _cache_file(path, query): """Return cache file name for specified `path` and `query` @@ -94,9 +110,9 @@ def _cache_file(path, query): digest = hashlib.sha1(("%s %s" % (path, query)).encode("utf-8")).hexdigest() digest_number = ord(digest[0].upper()) - expiry_interval = 60*(digest_number+180) + expiry_interval = 60 * (digest_number + 180) - timestamp = "%010d" % (int(time.time())//expiry_interval*expiry_interval) + timestamp = "%010d" % (int(time.time()) // expiry_interval * expiry_interval) filename = os.path.join(PROXY_CACHEDIR, timestamp, path, query) return filename @@ -108,25 +124,30 @@ def _load_content_and_headers(path, query): else: cache_file = _cache_file(path, query) try: - return (open(cache_file, 'r').read(), - json.loads(open(cache_file+".headers", 'r').read())) + return ( + open(cache_file, "r").read(), + json.loads(open(cache_file + ".headers", "r").read()), + ) except IOError: return None, None + def _touch_empty_file(path, query): cache_file = _cache_file(path, query) cache_dir = os.path.dirname(cache_file) if not os.path.exists(cache_dir): os.makedirs(cache_dir) - open(cache_file, 'w').write("") + open(cache_file, "w").write("") + def _save_content_and_headers(path, query, content, headers): cache_file = _cache_file(path, query) cache_dir = os.path.dirname(cache_file) if not os.path.exists(cache_dir): os.makedirs(cache_dir) - open(cache_file + ".headers", 'w').write(json.dumps(headers)) - open(cache_file, 'wb').write(content) + open(cache_file + ".headers", "w").write(json.dumps(headers)) + open(cache_file, "wb").write(content) + def translate(text, lang): """ @@ -137,7 +158,7 @@ def translate(text, lang): def _log_unknown_translation(lang, text): with open(MISSING_TRANSLATION_LOG % lang, "a") as f_missing_translation: - f_missing_translation.write(text+"\n") + f_missing_translation.write(text + "\n") if "," in text: terms = text.split(",") @@ -155,14 +176,17 @@ def translate(text, lang): translated = TRANSLATIONS.get(lang, {}).get(text.lower(), text) return translated + def cyr(to_translate): """ Transliterate `to_translate` from latin into cyrillic """ return cyrtranslit.to_cyrillic(to_translate) + def _patch_greek(original): - return original.replace(u"Ηλιόλουστη/ο", u"Ηλιόλουστη") + return original.replace("Ηλιόλουστη/ο", "Ηλιόλουστη") + def add_translations(content, lang): """ @@ -175,7 +199,7 @@ def add_translations(content, lang): languages_to_translate = TRANSLATIONS.keys() try: - d = json.loads(content) # pylint: disable=invalid-name + d = json.loads(content) # pylint: disable=invalid-name except (ValueError, TypeError) as exception: print("---") print(exception) @@ -183,55 +207,65 @@ def add_translations(content, lang): return {} try: - weather_condition = d['data']['current_condition' - ][0]['weatherDesc'][0]['value'].capitalize() - d['data']['current_condition'][0]['weatherDesc'][0]['value'] = \ - weather_condition + weather_condition = d["data"]["current_condition"][0]["weatherDesc"][0][ + "value" + ].capitalize() + d["data"]["current_condition"][0]["weatherDesc"][0]["value"] = weather_condition if lang in languages_to_translate: - d['data']['current_condition'][0]['lang_%s' % lang] = \ - [{'value': translate(weather_condition, lang)}] - elif lang == 'sr': - d['data']['current_condition'][0]['lang_%s' % lang] = \ - [{'value': cyr( - d['data']['current_condition'][0]['lang_%s' % lang][0]['value']\ - )}] - elif lang == 'el': - d['data']['current_condition'][0]['lang_%s' % lang] = \ - [{'value': _patch_greek( - d['data']['current_condition'][0]['lang_%s' % lang][0]['value']\ - )}] - elif lang == 'sr-lat': - d['data']['current_condition'][0]['lang_%s' % lang] = \ - [{'value':d['data']['current_condition'][0]['lang_sr'][0]['value']\ - }] + d["data"]["current_condition"][0]["lang_%s" % lang] = [ + {"value": translate(weather_condition, lang)} + ] + elif lang == "sr": + d["data"]["current_condition"][0]["lang_%s" % lang] = [ + { + "value": cyr( + d["data"]["current_condition"][0]["lang_%s" % lang][0]["value"] + ) + } + ] + elif lang == "el": + d["data"]["current_condition"][0]["lang_%s" % lang] = [ + { + "value": _patch_greek( + d["data"]["current_condition"][0]["lang_%s" % lang][0]["value"] + ) + } + ] + elif lang == "sr-lat": + d["data"]["current_condition"][0]["lang_%s" % lang] = [ + {"value": d["data"]["current_condition"][0]["lang_sr"][0]["value"]} + ] fixed_weather = [] - for w in d['data']['weather']: # pylint: disable=invalid-name + for w in d["data"]["weather"]: # pylint: disable=invalid-name fixed_hourly = [] - for h in w['hourly']: # pylint: disable=invalid-name - weather_condition = h['weatherDesc'][0]['value'] + for h in w["hourly"]: # pylint: disable=invalid-name + weather_condition = h["weatherDesc"][0]["value"] if lang in languages_to_translate: - h['lang_%s' % lang] = \ - [{'value': translate(weather_condition, lang)}] - elif lang == 'sr': - h['lang_%s' % lang] = \ - [{'value': cyr(h['lang_%s' % lang][0]['value'])}] - elif lang == 'el': - h['lang_%s' % lang] = \ - [{'value': _patch_greek(h['lang_%s' % lang][0]['value'])}] - elif lang == 'sr-lat': - h['lang_%s' % lang] = \ - [{'value': h['lang_sr'][0]['value']}] + h["lang_%s" % lang] = [ + {"value": translate(weather_condition, lang)} + ] + elif lang == "sr": + h["lang_%s" % lang] = [ + {"value": cyr(h["lang_%s" % lang][0]["value"])} + ] + elif lang == "el": + h["lang_%s" % lang] = [ + {"value": _patch_greek(h["lang_%s" % lang][0]["value"])} + ] + elif lang == "sr-lat": + h["lang_%s" % lang] = [{"value": h["lang_sr"][0]["value"]}] fixed_hourly.append(h) - w['hourly'] = fixed_hourly + w["hourly"] = fixed_hourly fixed_weather.append(w) - d['data']['weather'] = fixed_weather + d["data"]["weather"] = fixed_weather content = json.dumps(d) except (IndexError, ValueError) as exception: print(exception) return content + def _fetch_content_and_headers(path, query_string, **kwargs): content, headers = _load_content_and_headers(path, query_string) @@ -265,7 +299,7 @@ def _fetch_content_and_headers(path, query_string, **kwargs): _touch_empty_file(path, query_string) if response: headers = {} - headers['Content-Type'] = response.headers['content-type'] + headers["Content-Type"] = response.headers["content-type"] _save_content_and_headers(path, query_string, response.content, headers) content = response.content else: @@ -279,11 +313,13 @@ def _make_query(path, query_string): if _is_metno(): path, query, days = metno_request(path, query_string) - if USER_AGENT == '': - raise ValueError('User agent must be set to adhere to metno ToS: https://api.met.no/doc/TermsOfService') - content, headers = _fetch_content_and_headers(path, query, headers={ - 'User-Agent': USER_AGENT - }) + if USER_AGENT == "": + raise ValueError( + "User agent must be set to adhere to metno ToS: https://api.met.no/doc/TermsOfService" + ) + content, headers = _fetch_content_and_headers( + path, query, headers={"User-Agent": USER_AGENT} + ) content = create_standard_json_from_metno(content, days) else: # WWO tweaks @@ -293,16 +329,17 @@ def _make_query(path, query_string): return content, headers + @APP.route("/") def proxy(path): """ Main proxy function. Handles incoming HTTP queries. """ - lang = request.args.get('lang', 'en') + lang = request.args.get("lang", "en") query_string = request.query_string.decode("utf-8") - query_string = query_string.replace('sr-lat', 'sr') - query_string = query_string.replace('lang=None', 'lang=en') + query_string = query_string.replace("sr-lat", "sr") + query_string = query_string.replace("lang=None", "lang=en") content = "" headers = "" @@ -314,19 +351,20 @@ def proxy(path): return content, 200, headers + if __name__ == "__main__": - #app.run(host='0.0.0.0', port=5001, debug=False) - #app.debug = True + # app.run(host='0.0.0.0', port=5001, debug=False) + # app.debug = True if len(sys.argv) == 1: bind_addr = "0.0.0.0" SERVER = WSGIServer((bind_addr, PROXY_PORT), APP) SERVER.serve_forever() else: - print('running single request from command line arg') + print("running single request from command line arg") APP.testing = True with APP.test_client() as c: resp = c.get(sys.argv[1]) - print('Status: ' + resp.status) + print("Status: " + resp.status) # print('Headers: ' + dumps(resp.headers)) - print(resp.data.decode('utf-8')) + print(resp.data.decode("utf-8"))