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

syntax clean up

Browse Source
pull/223/head
Igor Chubin 6 years ago
parent fcc7b79272
commit fb89c4d985
1 changed files with 109 additions and 84 deletions
Show Stats Download Patch File Download Diff File

193
lib/srv.py
Unescape View File

@ -1,13 +1,11 @@
#!/usr/bin/env python
# vim: set encoding=utf-8
import gevent
from gevent.wsgi import WSGIServer
from gevent.queue import Queue
from gevent.pywsgi import WSGIServer
from gevent.monkey import patch_all
from gevent.subprocess import Popen, PIPE, STDOUT
patch_all()
# pylint: disable=wrong-import-position,wrong-import-order
import sys
import logging
import os
@ -20,37 +18,44 @@ import json
import geoip2.database
import jinja2
from flask import Flask, request, render_template, send_from_directory, send_file, make_response
app = Flask(__name__)
from flask import Flask, request, render_template, \
send_from_directory, send_file, make_response
APP = Flask(__name__)
MYDIR = os.path.abspath(
os.path.dirname(os.path.dirname('__file__')))
sys.path.append("%s/lib/" % MYDIR)
import wttrin_png, parse_query
import wttrin_png
import parse_query
from translations import get_message, FULL_TRANSLATION, PARTIAL_TRANSLATION, SUPPORTED_LANGS
from buttons import TWITTER_BUTTON, GITHUB_BUTTON, GITHUB_BUTTON_2, GITHUB_BUTTON_3, GITHUB_BUTTON_FOOTER
from buttons import TWITTER_BUTTON, \
GITHUB_BUTTON, GITHUB_BUTTON_2, GITHUB_BUTTON_3, \
GITHUB_BUTTON_FOOTER
from globals import GEOLITE, \
IP2LCACHE, ALIASES, BLACKLIST, \
get_help_file, BASH_FUNCTION_FILE, TRANSLATION_FILE, LOG_FILE, TEST_FILE, \
get_help_file, BASH_FUNCTION_FILE, TRANSLATION_FILE, LOG_FILE, \
TEMPLATES, STATIC, \
NOT_FOUND_LOCATION, \
MALFORMED_RESPONSE_HTML_PAGE, \
IATA_CODES_FILE, \
log, error, \
LISTEN_PORT, LISTEN_HOST, PLAIN_TEXT_AGENTS
log, \
LISTEN_PORT, LISTEN_HOST, PLAIN_TEXT_AGENTS, PLAIN_TEXT_PAGES, \
IP2LOCATION_KEY
from wttr import get_wetter, get_moon
# pylint: enable=wrong-import-position,wrong-import-order
if not os.path.exists(os.path.dirname(LOG_FILE)):
os.makedirs(os.path.dirname(LOG_FILE))
logging.basicConfig(filename=LOG_FILE, level=logging.DEBUG, format='%(asctime)s %(message)s')
my_loader = jinja2.ChoiceLoader([
app.jinja_loader,
MY_LOADER = jinja2.ChoiceLoader([
APP.jinja_loader,
jinja2.FileSystemLoader(TEMPLATES),
])
app.jinja_loader = my_loader
APP.jinja_loader = MY_LOADER
class Limits:
def __init__(self):
@ -106,56 +111,69 @@ class Limits:
self.counter[interval] = {}
self.last_update[interval] = t_int / self.divisor[interval]
limits = Limits()
LIMITS = Limits()
def is_ip(ip_addr):
"""
Check if `ip_addr` looks like an IP Address
"""
def is_ip(ip):
if re.match('\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}', ip) is None:
if re.match(r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}', ip_addr) is None:
return False
try:
socket.inet_aton(ip)
socket.inet_aton(ip_addr)
return True
except socket.error:
return False
def location_normalize(location):
"""
Normalize location name `location`
"""
#translation_table = dict.fromkeys(map(ord, '!@#$*;'), None)
def remove_chars( c, s ):
return ''.join(x for x in s if x not in c )
def _remove_chars(chars, string):
return ''.join(x for x in string if x not in chars)
location = location.lower().replace('_', ' ').replace('+', ' ').strip()
if not location.startswith('moon@'):
location = remove_chars(r'!@#$*;:\\', location)
location = _remove_chars(r'!@#$*;:\\', location)
return location
def load_aliases(aliases_filename):
"""
Load aliases from the aliases file
"""
aliases_db = {}
with open(aliases_filename, 'r') as f:
for line in f.readlines():
with open(aliases_filename, 'r') as f_aliases:
for line in f_aliases.readlines():
from_, to_ = line.decode('utf-8').split(':', 1)
aliases_db[location_normalize(from_)] = location_normalize(to_)
return aliases_db
def load_iata_codes(iata_codes_filename):
with open(iata_codes_filename, 'r') as f:
"""
Load IATA codes from the IATA codes file
"""
with open(iata_codes_filename, 'r') as f_iata_codes:
result = []
for line in f.readlines():
for line in f_iata_codes.readlines():
result.append(line.strip())
return set(result)
location_alias = load_aliases(ALIASES)
location_black_list = [x.strip() for x in open(BLACKLIST, 'r').readlines()]
iata_codes = load_iata_codes(IATA_CODES_FILE)
print "IATA CODES LOADED: %s" % len(iata_codes)
LOCATION_ALIAS = load_aliases(ALIASES)
LOCATION_BLACK_LIST = [x.strip() for x in open(BLACKLIST, 'r').readlines()]
IATA_CODES = load_iata_codes(IATA_CODES_FILE)
GEOIP_READER = geoip2.database.Reader(GEOLITE)
def location_canonical_name(location):
location = location_normalize(location)
if location in location_alias:
return location_alias[location.lower()]
if location in LOCATION_ALIAS:
return LOCATION_ALIAS[location.lower()]
return location
def ascii_only(string):
try:
for i in range(5):
for _ in range(5):
string = string.encode('utf-8')
return True
except:
@ -188,19 +206,23 @@ def ip2location(ip):
return location
try:
t = requests.get( 'http://api.ip2location.com/?ip=%s&key=%s&package=WS10' % (IP2LOCATION_KEY, ip)).text
if ';' in t:
location = t.split(';')[3]
ip2location_response = requests\
.get('http://api.ip2location.com/?ip=%s&key=%s&package=WS10' \
% (IP2LOCATION_KEY, ip)).text
if ';' in ip2location_response:
location = ip2location_response.split(';')[3]
open(cached, 'w').write(location)
print "ip2location says: %s" % location
return location
except:
pass
reader = geoip2.database.Reader(GEOLITE)
def get_location(ip_addr):
response = reader.city(ip_addr)
"""
Return location pair (CITY, COUNTRY) for `ip_addr`
"""
response = GEOIP_READER.city(ip_addr)
country = response.country.iso_code
city = response.city.name
@ -219,28 +241,28 @@ def get_location(ip_addr):
return (city or NOT_FOUND_LOCATION), country
def parse_accept_language(acceptLanguage):
languages = acceptLanguage.split(",")
locale_q_pairs = []
languages = acceptLanguage.split(",")
locale_q_pairs = []
for language in languages:
try:
if language.split(";")[0] == language:
# no q => q = 1
locale_q_pairs.append((language.strip(), "1"))
else:
locale = language.split(";")[0].strip()
q = language.split(";")[1].split("=")[1]
locale_q_pairs.append((locale, q))
except:
pass
for language in languages:
try:
if language.split(";")[0] == language:
# no q => q = 1
locale_q_pairs.append((language.strip(), "1"))
else:
locale = language.split(";")[0].strip()
weight = language.split(";")[1].split("=")[1]
locale_q_pairs.append((locale, weight))
except:
pass
return locale_q_pairs
return locale_q_pairs
def find_supported_language(accepted_languages):
for p in accepted_languages:
lang = p[0]
for lang_tuple in accepted_languages:
lang = lang_tuple[0]
if '-' in lang:
lang = lang.split('-',1)[0]
lang = lang.split('-', 1)[0]
if lang in SUPPORTED_LANGS:
return lang
return None
@ -255,24 +277,28 @@ def show_help(location, lang):
text = open(BASH_FUNCTION_FILE, 'r').read()
elif location == ":translation":
text = open(TRANSLATION_FILE, 'r').read()
text = text.replace('NUMBER_OF_LANGUAGES', str(len(SUPPORTED_LANGS))).replace('SUPPORTED_LANGUAGES', ' '.join(SUPPORTED_LANGS))
text = text\
.replace('NUMBER_OF_LANGUAGES', str(len(SUPPORTED_LANGS)))\
.replace('SUPPORTED_LANGUAGES', ' '.join(SUPPORTED_LANGS))
return text.decode('utf-8')
show_help.pages = [':help', ':bash.function', ':translation' ]
@app.route('/files/<path:path>')
@APP.route('/files/<path:path>')
def send_static(path):
"Send any static file located in /files/"
return send_from_directory(STATIC, path)
@app.route('/favicon.ico')
@APP.route('/favicon.ico')
def send_favicon():
"Send static file favicon.ico"
return send_from_directory(STATIC, 'favicon.ico')
@app.route('/malformed-response.html')
@APP.route('/malformed-response.html')
def send_malformed():
"Send static file malformed-response.html"
return send_from_directory(STATIC, 'malformed-response.html')
@app.route("/")
@app.route("/<string:location>")
@APP.route("/")
@APP.route("/<string:location>")
def wttr(location = None):
"""
Main rendering function, it processes incoming weather queries.
@ -294,21 +320,21 @@ def wttr(location = None):
lang = hostname[:-8]
if request.headers.getlist("X-Forwarded-For"):
ip = request.headers.getlist("X-Forwarded-For")[0]
if ip.startswith('::ffff:'):
ip = ip[7:]
ip_addr = request.headers.getlist("X-Forwarded-For")[0]
if ip_addr.startswith('::ffff:'):
ip_addr = ip_addr[7:]
else:
ip = request.remote_addr
ip_addr = request.remote_addr
try:
limits.check_ip(ip)
LIMITS.check_ip(ip_addr)
except RuntimeError, e:
return str(e)
except Exception, e:
logging.error("Exception has occured", exc_info=1)
return "ERROR"
if location is not None and location.lower() in location_black_list:
if location is not None and location.lower() in LOCATION_BLACK_LIST:
return ""
png_filename = None
@ -321,28 +347,26 @@ def wttr(location = None):
if 'lang' in request.args:
lang = request.args.get('lang')
if lang is None and 'Accept-Language' in request.headers:
lang = find_supported_language(parse_accept_language(request.headers.get('Accept-Language', '')))
lang = find_supported_language(
parse_accept_language(
request.headers.get('Accept-Language', '')))
user_agent = request.headers.get('User-Agent', '').lower()
html_output = not any(agent in user_agent for agent in PLAIN_TEXT_AGENTS)
if location in show_help.pages:
if location in PLAIN_TEXT_PAGES:
help_ = show_help(location, lang)
if html_output:
return render_template('index.html', body=help_)
else:
return help_
return help_
orig_location = location
if request.headers.getlist("X-Forwarded-For"):
ip = request.headers.getlist("X-Forwarded-For")[0]
if ip.startswith('::ffff:'):
ip = ip[7:]
ip_addr = request.headers.getlist("X-Forwarded-For")[0]
if ip_addr.startswith('::ffff:'):
ip_addr = ip_addr[7:]
else:
ip = request.remote_addr
ip_addr = request.remote_addr
try:
# if location is starting with ~
@ -354,7 +378,7 @@ def wttr(location = None):
if location is not None and not ascii_only(location):
location = "~" + location
if location is not None and location.upper() in iata_codes:
if location is not None and location.upper() in IATA_CODES:
location = '~%s' % location
if location is not None and location.startswith('~'):
@ -367,7 +391,7 @@ def wttr(location = None):
else:
location = NOT_FOUND_LOCATION #location[1:]
try:
query_source_location = get_location(ip)
query_source_location = get_location(ip_addr)
except:
query_source_location = NOT_FOUND_LOCATION, None
@ -407,7 +431,8 @@ def wttr(location = None):
location = location_canonical_name(location)
log("%s %s %s %s %s %s" \
% (ip, user_agent, orig_location, location, query.get('use_imperial', False), lang))
% (ip_addr, user_agent, orig_location, location,
query.get('use_imperial', False), lang))
# We are ready to return the answer
if png_filename:
@ -434,7 +459,7 @@ def wttr(location = None):
else:
if country and location != NOT_FOUND_LOCATION:
location = "%s, %s" % (location, country)
output = get_wetter(location, ip,
output = get_wetter(location, ip_addr,
html=html_output,
lang=lang,
query=query,
@ -473,5 +498,5 @@ def wttr(location = None):
logging.error("Exception has occured", exc_info=1)
return "ERROR"
server = WSGIServer((LISTEN_HOST, LISTEN_PORT), app)
server.serve_forever()
SERVER = WSGIServer((LISTEN_HOST, LISTEN_PORT), APP)
SERVER.serve_forever()

Write Preview
Loading…
Cancel
Save