Fix formatting of bin/geo-proxy.py

pull/1062/head
Igor Chubin 1 month ago
parent 64eb93b35f
commit 312428bb6e

@ -1,88 +1,109 @@
import gevent import gevent
from gevent.pywsgi import WSGIServer from gevent.pywsgi import WSGIServer
from gevent.queue import Queue from gevent.queue import Queue
from gevent.monkey import patch_all from gevent.monkey import patch_all
from gevent.subprocess import Popen, PIPE, STDOUT from gevent.subprocess import Popen, PIPE, STDOUT
patch_all() patch_all()
import sys import sys
import os import os
import json import json
from flask import Flask, request, render_template, send_from_directory, send_file, make_response, jsonify, Response from flask import (
Flask,
request,
render_template,
send_from_directory,
send_file,
make_response,
jsonify,
Response,
)
app = Flask(__name__) app = Flask(__name__)
MYDIR = os.path.abspath(os.path.dirname('__file__')) MYDIR = os.path.abspath(os.path.dirname("__file__"))
sys.path.append(os.path.join(MYDIR, 'lib')) sys.path.append(os.path.join(MYDIR, "lib"))
CACHEDIR = os.path.join(MYDIR, "cache")
CACHEDIR = os.path.join(MYDIR, 'cache') from geopy.geocoders import Nominatim # , Mapzen
from geopy.geocoders import Nominatim #, Mapzen # geomapzen = Mapzen("mapzen-RBNbmcZ") # Nominatim()
#geomapzen = Mapzen("mapzen-RBNbmcZ") # Nominatim()
geoosm = Nominatim(timeout=7, user_agent="wttrin-geo/0.0.2") geoosm = Nominatim(timeout=7, user_agent="wttrin-geo/0.0.2")
import airports import airports
# from tzwhere import tzwhere # from tzwhere import tzwhere
import timezonefinder import timezonefinder
tf = timezonefinder.TimezoneFinder() tf = timezonefinder.TimezoneFinder()
def load_cache(location_string): def load_cache(location_string):
try: try:
location_string = location_string.replace('/', '_') location_string = location_string.replace("/", "_")
cachefile = os.path.join(CACHEDIR, location_string) cachefile = os.path.join(CACHEDIR, location_string)
return json.loads(open(cachefile, 'r').read()) return json.loads(open(cachefile, "r").read())
except: except:
return None return None
def shorten_full_address(address): def shorten_full_address(address):
parts = address.split(',') parts = address.split(",")
if len(parts) > 6: if len(parts) > 6:
parts = parts[:2] + [x for x in parts[-4:] if len(x) < 20] parts = parts[:2] + [x for x in parts[-4:] if len(x) < 20]
return ','.join(parts) return ",".join(parts)
return address return address
def save_cache(location_string, answer): def save_cache(location_string, answer):
location_string = location_string.replace('/', '_') location_string = location_string.replace("/", "_")
cachefile = os.path.join(CACHEDIR, location_string) cachefile = os.path.join(CACHEDIR, location_string)
open(cachefile, 'w').write(json.dumps(answer)) open(cachefile, "w").write(json.dumps(answer))
def query_osm(location_string): def query_osm(location_string):
try: try:
location = geoosm.geocode(location_string) location = geoosm.geocode(location_string)
return { return {
'address': location.address, "address": location.address,
'latitude': location.latitude, "latitude": location.latitude,
'longitude':location.longitude, "longitude": location.longitude,
} }
except Exception as e: except Exception as e:
print(e) print(e)
return None return None
def add_timezone_information(geo_data): def add_timezone_information(geo_data):
# tzwhere_ = tzwhere.tzwhere() # tzwhere_ = tzwhere.tzwhere()
# timezone_str = tzwhere_.tzNameAt(geo_data["latitude"], geo_data["longitude"]) # timezone_str = tzwhere_.tzNameAt(geo_data["latitude"], geo_data["longitude"])
timezone_str = tf.certain_timezone_at(lat=geo_data["latitude"], lng=geo_data["longitude"]) timezone_str = tf.certain_timezone_at(
lat=geo_data["latitude"], lng=geo_data["longitude"]
)
answer = geo_data.copy() answer = geo_data.copy()
answer["timezone"] = timezone_str answer["timezone"] = timezone_str
return answer return answer
@app.route("/<string:location>") @app.route("/<string:location>")
def find_location(location): def find_location(location):
airport_gps_location = airports.get_airport_gps_location(location.upper().lstrip('~')) airport_gps_location = airports.get_airport_gps_location(
location.upper().lstrip("~")
)
is_airport = False is_airport = False
if airport_gps_location is not None: if airport_gps_location is not None:
location = airport_gps_location location = airport_gps_location
is_airport = True is_airport = True
location = location.replace('+', ' ') location = location.replace("+", " ")
answer = load_cache(location) answer = load_cache(location)
loaded_answer = None loaded_answer = None
@ -92,12 +113,12 @@ def find_location(location):
if answer is None: if answer is None:
answer = query_osm(location) answer = query_osm(location)
if is_airport: if is_airport:
answer['address'] = shorten_full_address(answer['address']) answer["address"] = shorten_full_address(answer["address"])
if "timezone" not in answer: if "timezone" not in answer:
answer = add_timezone_information(answer) answer = add_timezone_information(answer)
if answer is not None and loaded_answer != answer: if answer is not None and loaded_answer != answer:
save_cache(location, answer) save_cache(location, answer)
@ -105,10 +126,11 @@ def find_location(location):
if answer is None: if answer is None:
return "" return ""
else: else:
r = Response(json.dumps(answer)) r = Response(json.dumps(answer))
r.headers["Content-Type"] = "text/json; charset=utf-8" r.headers["Content-Type"] = "text/json; charset=utf-8"
return r return r
app.config['JSON_AS_ASCII'] = False
app.config["JSON_AS_ASCII"] = False
server = WSGIServer(("127.0.0.1", 8004), app) server = WSGIServer(("127.0.0.1", 8004), app)
server.serve_forever() server.serve_forever()

Loading…
Cancel
Save