mirror of
https://github.com/chubin/wttr.in
synced 2024-11-15 00:27:09 +00:00
280 lines
8.2 KiB
Python
280 lines
8.2 KiB
Python
"""
|
|
All location related functions and converters.
|
|
|
|
The main entry point is `location_processing`
|
|
which gets `location` and `source_ip_address`
|
|
and basing on this information generates
|
|
precise location description.
|
|
|
|
"""
|
|
from __future__ import print_function
|
|
|
|
import os
|
|
import json
|
|
import re
|
|
import socket
|
|
import requests
|
|
import geoip2.database
|
|
|
|
from globals import GEOLITE, GEOLOCATOR_SERVICE, IP2LCACHE, IP2LOCATION_KEY, NOT_FOUND_LOCATION, \
|
|
ALIASES, BLACKLIST, IATA_CODES_FILE
|
|
|
|
GEOIP_READER = geoip2.database.Reader(GEOLITE)
|
|
|
|
def ascii_only(string):
|
|
"Check if `string` contains only ASCII symbols"
|
|
|
|
try:
|
|
for _ in range(5):
|
|
string = string.encode('utf-8')
|
|
return True
|
|
except UnicodeDecodeError:
|
|
return False
|
|
|
|
def is_ip(ip_addr):
|
|
"""
|
|
Check if `ip_addr` looks like an IP Address
|
|
"""
|
|
|
|
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_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(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)
|
|
return location
|
|
|
|
|
|
|
|
def geolocator(location):
|
|
"""
|
|
Return a GPS pair for specified `location` or None
|
|
if nothing can't be found
|
|
"""
|
|
|
|
try:
|
|
geo = requests.get('%s/%s' % (GEOLOCATOR_SERVICE, location)).text
|
|
except requests.exceptions.ConnectionError as exception:
|
|
print("ERROR: %s" % exception)
|
|
return None
|
|
|
|
if geo == "":
|
|
return None
|
|
|
|
try:
|
|
answer = json.loads(geo.encode('utf-8'))
|
|
return answer
|
|
except ValueError as exception:
|
|
print("ERROR: %s" % exception)
|
|
return None
|
|
|
|
return None
|
|
|
|
def ip2location(ip_addr):
|
|
"Convert IP address `ip_addr` to a location name"
|
|
|
|
cached = os.path.join(IP2LCACHE, ip_addr)
|
|
if not os.path.exists(IP2LCACHE):
|
|
os.makedirs(IP2LCACHE)
|
|
|
|
location = None
|
|
|
|
if os.path.exists(cached):
|
|
location = open(cached, 'r').read()
|
|
else:
|
|
# if IP2LOCATION_KEY is not set, do not the query,
|
|
# because the query wont be processed anyway
|
|
if IP2LOCATION_KEY:
|
|
try:
|
|
ip2location_response = requests\
|
|
.get('http://api.ip2location.com/?ip=%s&key=%s&package=WS3' \
|
|
% (ip_addr, IP2LOCATION_KEY)).text
|
|
if ';' in ip2location_response:
|
|
open(cached, 'w').write(ip2location_response)
|
|
location = ip2location_response
|
|
except requests.exceptions.ConnectionError:
|
|
pass
|
|
|
|
if location and ';' in location:
|
|
location = location.split(';')[3], location.split(';')[1]
|
|
else:
|
|
location = location, None
|
|
|
|
return location
|
|
|
|
def get_location(ip_addr):
|
|
"""
|
|
Return location pair (CITY, COUNTRY) for `ip_addr`
|
|
"""
|
|
|
|
try:
|
|
response = GEOIP_READER.city(ip_addr)
|
|
country = response.country.name
|
|
city = response.city.name
|
|
except geoip2.errors.AddressNotFoundError:
|
|
country = None
|
|
city = None
|
|
|
|
#
|
|
# temporary disabled it because of geoip services capcacity
|
|
#
|
|
#if city is None and response.location:
|
|
# coord = "%s, %s" % (response.location.latitude, response.location.longitude)
|
|
# try:
|
|
# location = geolocator.reverse(coord, language='en')
|
|
# city = location.raw.get('address', {}).get('city')
|
|
# except:
|
|
# pass
|
|
if city is None:
|
|
city, country = ip2location(ip_addr)
|
|
|
|
# workaround for the strange bug with the country name
|
|
# maybe some other countries has this problem too
|
|
if country == 'Russian Federation':
|
|
country = 'Russia'
|
|
|
|
if city:
|
|
return city, country
|
|
else:
|
|
return NOT_FOUND_LOCATION, None
|
|
|
|
|
|
def location_canonical_name(location):
|
|
"Find canonical name for `location`"
|
|
|
|
location = location_normalize(location)
|
|
if location.lower() in LOCATION_ALIAS:
|
|
return LOCATION_ALIAS[location.lower()]
|
|
return location
|
|
|
|
def load_aliases(aliases_filename):
|
|
"""
|
|
Load aliases from the aliases file
|
|
"""
|
|
aliases_db = {}
|
|
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):
|
|
"""
|
|
Load IATA codes from the IATA codes file
|
|
"""
|
|
with open(iata_codes_filename, 'r') as f_iata_codes:
|
|
result = []
|
|
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)
|
|
|
|
def is_location_blocked(location):
|
|
"""
|
|
Return True if this location is blocked
|
|
or False if it is allowed
|
|
"""
|
|
return location is not None and location.lower() in LOCATION_BLACK_LIST
|
|
|
|
|
|
def location_processing(location, ip_addr):
|
|
"""
|
|
"""
|
|
|
|
# if location is starting with ~
|
|
# or has non ascii symbols
|
|
# it should be handled like a search term (for geolocator)
|
|
override_location_name = None
|
|
full_address = None
|
|
hide_full_address = False
|
|
force_show_full_address = location is not None and location.startswith('~')
|
|
|
|
# location ~ means that it should be detected automatically,
|
|
# and shown in the location line below the report
|
|
if location == '~':
|
|
location = None
|
|
|
|
if location and location.lstrip('~ ').startswith('@'):
|
|
try:
|
|
location, country = get_location(
|
|
socket.gethostbyname(
|
|
location.lstrip('~ ')[1:]))
|
|
location = '~' + location
|
|
if country:
|
|
location += ", %s" % country
|
|
hide_full_address = not force_show_full_address
|
|
except:
|
|
location, country = NOT_FOUND_LOCATION, None
|
|
|
|
query_source_location = get_location(ip_addr)
|
|
|
|
country = None
|
|
if location is None or location == 'MyLocation':
|
|
location, country = query_source_location
|
|
|
|
# Let us try to use geolocation services for locations
|
|
# that were derived from IPs.
|
|
if location:
|
|
location = '~' + location
|
|
if country:
|
|
location += ", %s" % country
|
|
hide_full_address = not force_show_full_address
|
|
elif is_ip(location):
|
|
location, country = get_location(location)
|
|
|
|
# here too
|
|
if location:
|
|
location = '~' + location
|
|
if country:
|
|
location += ", %s" % country
|
|
hide_full_address = not force_show_full_address
|
|
|
|
if location and not location.startswith('~'):
|
|
tmp_location = location_canonical_name(location)
|
|
if tmp_location != location:
|
|
override_location_name = location
|
|
location = tmp_location
|
|
|
|
# up to this point it is possible that the name
|
|
# contains some unicode symbols
|
|
# here we resolve them
|
|
if location is not None and not ascii_only(location):
|
|
location = "~" + location.lstrip('~ ')
|
|
|
|
if location is not None and location.upper() in IATA_CODES:
|
|
location = '~%s' % location
|
|
|
|
if location is not None and location.startswith('~'):
|
|
geolocation = geolocator(location_canonical_name(location[1:]))
|
|
if geolocation is not None:
|
|
override_location_name = location[1:].replace('+', ' ')
|
|
location = "%s,%s" % (geolocation['latitude'], geolocation['longitude'])
|
|
country = None
|
|
if not hide_full_address:
|
|
full_address = geolocation['address']
|
|
else:
|
|
full_address = None
|
|
else:
|
|
location = NOT_FOUND_LOCATION #location[1:]
|
|
|
|
return location, \
|
|
override_location_name, \
|
|
full_address, \
|
|
country, \
|
|
query_source_location
|