292 lines
9.2 KiB
Python
292 lines
9.2 KiB
Python
# -*- coding: utf-8 -*-
|
||
import time
|
||
|
||
import overpass
|
||
import copy
|
||
from geopy.geocoders import Nominatim
|
||
|
||
Nominatim_last_request = None
|
||
|
||
def osm_api_request(data, res_type):
|
||
|
||
time.sleep(1)
|
||
|
||
api = overpass.API(timeout=100)
|
||
|
||
res = api.get(data, responseformat=res_type)
|
||
|
||
pres_names = []
|
||
res_list = []
|
||
if res_type[:3].lower() == 'csv':
|
||
data_type_list = copy.deepcopy(res[0])
|
||
for line in res[1:]:
|
||
if line[0] in pres_names:
|
||
continue
|
||
|
||
i = 0
|
||
line_Dict = {}
|
||
while i < len(data_type_list):
|
||
line_Dict.update({data_type_list[i]: line[i]})
|
||
i += 1
|
||
if 'name:en' in line_Dict and 'name' in line_Dict:
|
||
if not line_Dict["name:en"]:
|
||
if line_Dict["name"]:
|
||
line_Dict["name:en"] = line_Dict["name"]
|
||
|
||
if not line_Dict["name:en"]:
|
||
continue
|
||
|
||
res_list.append(line_Dict)
|
||
pres_names.append(line[0])
|
||
else:
|
||
res_list = copy.deepcopy(res)
|
||
|
||
return res_list
|
||
|
||
|
||
def osm_get_area_id_by_params_dict(paramsDict):
|
||
|
||
area_id = None
|
||
|
||
try:
|
||
|
||
# Geocoding request via Nominatim
|
||
time.sleep(1)
|
||
geolocator = Nominatim(user_agent='TWB')
|
||
geo_results = geolocator.geocode(paramsDict, exactly_one=False, limit=3)
|
||
|
||
if not geo_results:
|
||
return area_id
|
||
|
||
# Searching for relation in result set
|
||
country_obj = None
|
||
for r in geo_results:
|
||
print(r.address, r.raw.get("osm_type"))
|
||
if r.raw.get("osm_type") == "relation":
|
||
country_obj = r
|
||
break
|
||
|
||
if not country_obj:
|
||
return area_id
|
||
|
||
# Calculating area id
|
||
area_id = int(country_obj.raw.get("osm_id")) + 3600000000
|
||
|
||
except Exception as e:
|
||
print(f'osm_get_area_id_by_params_dict Error = {e}')
|
||
|
||
return area_id
|
||
|
||
|
||
def osm_get_countries():
|
||
res_type = 'csv("name", "name:en", "name:ru", "ISO3166-1", "flag", "int_name", "official_name", "ISO3166-1:alpha3", ' \
|
||
'"ISO3166-1:numeric", ::lon, ::lat)'
|
||
data = 'relation["admin_level"="2"]["ISO3166-1:alpha3"~"^...$"]; ' \
|
||
'out center;'
|
||
|
||
res = osm_api_request(data, res_type)
|
||
from operator import itemgetter
|
||
res = sorted(res, key=itemgetter('name'))
|
||
|
||
return res
|
||
|
||
|
||
def osm_get_cities_by_country(country_Dict):
|
||
|
||
res = []
|
||
|
||
try:
|
||
|
||
if 'area_id' in country_Dict and country_Dict['area_id']:
|
||
area_id = country_Dict['area_id']
|
||
else:
|
||
return []
|
||
|
||
res_type = 'csv("name", "name:ru", "name:en", ::lon, ::lat)'
|
||
data = f'area({area_id})->.searchArea;' \
|
||
'(node[place~"city$|town$"](area.searchArea););' \
|
||
'out center;'
|
||
|
||
res = osm_api_request(data, res_type)
|
||
|
||
except Exception as e:
|
||
print(f'osm_get_cities_by_country Error = {e}')
|
||
|
||
return res
|
||
|
||
|
||
|
||
def osm_get_airports(area_id, city_find_str=None):
|
||
|
||
airports_cities_Dict = {}
|
||
|
||
try:
|
||
|
||
if not area_id:
|
||
return airports_cities_Dict
|
||
|
||
res_type = 'csv("name", "name:ru", "name:en", "iata", "icao", "place", "int_name", "addr:country", "city_served", ::lon, ::lat)'
|
||
data = f'area({area_id})->.searchArea;' \
|
||
f'nwr["aeroway"="aerodrome"]["iata"~"^...$"]["icao"~"^....$"]["abandoned"!~".*"]["landuse"!="military"]["was:landuse"!="military"]["aerodrome:type"!~"airfield|military|private"]["amenity"!="flight_school"]["closed"!="yes"](area.searchArea)->.airports;' \
|
||
|
||
if city_find_str:
|
||
data = f'{data}' \
|
||
f'foreach.airports->.elem(nwr(around.elem:20000)[{city_find_str}]->.city; .elem out center; .city out center;);'
|
||
else:
|
||
data = f'{data} .airports out center;'
|
||
|
||
res = osm_api_request(data, res_type)
|
||
|
||
if not city_find_str:
|
||
return res
|
||
|
||
present_IATA_list = []
|
||
|
||
i = 0
|
||
while i < len(res):
|
||
|
||
if not res[i]['iata'] or not res[i]['icao']:
|
||
del res[i]
|
||
continue
|
||
|
||
if res[i]['iata'] and res[i]['iata'] in present_IATA_list:
|
||
del res[i]
|
||
continue
|
||
present_IATA_list.append(res[i]['iata'])
|
||
|
||
# если аэропорт
|
||
if res[i]['iata']:
|
||
if i + 1 < len(res) and res[i+1]['place']:
|
||
|
||
# ищем город на следующих строках
|
||
i2 = i + 1
|
||
linked = None
|
||
while i2 < len(res) and res[i2]['place']:
|
||
if not linked:
|
||
if (res[i2]['name'] and res[i]['name'] and res[i2]['name'] in res[i]['name']) or \
|
||
(res[i2]['name:en'] and res[i]['name:en'] and res[i2]['name:en'] in res[i]['name:en']) or \
|
||
(res[i2]['name:ru'] and res[i]['name:ru'] and res[i2]['name:ru'] in res[i]['name:ru']):
|
||
linked = i2
|
||
|
||
i2 += 1
|
||
|
||
if not linked:
|
||
linked = i + 1
|
||
|
||
res[i]['city'] = copy.deepcopy(res[linked])
|
||
if not res[i]['city']['name'] in airports_cities_Dict:
|
||
airports_cities_Dict.update({res[i]['city']['name:en']: [res[i]]})
|
||
else:
|
||
airports_cities_Dict[res[i]['city']['name:en']].append(res[i])
|
||
|
||
while i < i2:
|
||
del res[i]
|
||
i2 -= 1
|
||
|
||
continue
|
||
|
||
# если не найдена связка с городом
|
||
elif res[i]['city_served']:
|
||
|
||
res[i]['city'] = {'name:en': res[i]['city_served']}
|
||
if not res[i]['city']['name:en'] in airports_cities_Dict:
|
||
airports_cities_Dict.update({res[i]['city']['name:en']: [res[i]]})
|
||
else:
|
||
airports_cities_Dict[res[i]['city']['name:en']].append(res[i])
|
||
del res[i]
|
||
continue
|
||
|
||
# текущий элемент - не аэропорт - удаляем
|
||
else:
|
||
del res[i]
|
||
continue
|
||
|
||
i += 1
|
||
|
||
|
||
if res:
|
||
airports_cities_Dict[None] = res
|
||
|
||
except Exception as e:
|
||
print(f'osm_get_airports Error = {e}')
|
||
if e.args[0] == 25:
|
||
return {'error': 'timeout'}
|
||
|
||
|
||
return airports_cities_Dict
|
||
|
||
|
||
|
||
def osm_get_country_w_cities_n_airports(country_Dict, area_id):
|
||
print(f'{country_Dict["name:en"]}')
|
||
|
||
if area_id:
|
||
country_Dict['area_id'] = area_id
|
||
else:
|
||
country_Dict['area_id'] = osm_get_area_id_by_params_dict({'country': country_Dict['name']})
|
||
|
||
airports_Dict = osm_get_airports(country_Dict['area_id'], 'place~"city$|town$"')
|
||
if airports_Dict and 'error' in airports_Dict and airports_Dict['error'] == 'timeout':
|
||
airports_Dict = osm_get_airports(country_Dict['area_id'], 'place="city"')
|
||
|
||
from ReferenceDataApp.funcs import get_countries_key_data, get_cities_by_country_name_en
|
||
db_cities = get_cities_by_country_name_en(country_Dict["name:en"])
|
||
|
||
cities = osm_get_cities_by_country(country_Dict)
|
||
for city in cities:
|
||
try:
|
||
print(f' > {city["name:en"]}')
|
||
except Exception as e:
|
||
print(str(e))
|
||
|
||
if airports_Dict and 'error' in airports_Dict and airports_Dict['error'] == 'timeout':
|
||
|
||
if city['name:en'] in db_cities.keys():
|
||
city['area_id'] = db_cities[city['name:en']]
|
||
else:
|
||
city['area_id'] = osm_get_area_id_by_params_dict(
|
||
{'country': country_Dict['name:en'], 'city': city['name:en']})
|
||
|
||
airports_list = osm_get_airports(city['area_id'])
|
||
if airports_list:
|
||
city['airports'] = copy.deepcopy(airports_list)
|
||
else:
|
||
if city['name:en'] in airports_Dict.keys():
|
||
city['airports'] = copy.deepcopy(airports_Dict[city['name:en']])
|
||
del airports_Dict[city['name:en']]
|
||
|
||
if not 'airports' in city:
|
||
city['airports'] = []
|
||
|
||
print(f' > > airports count={str(len(city["airports"]))}')
|
||
|
||
city['parsing_status'] = 'finished'
|
||
|
||
country_Dict['cities'] = cities
|
||
country_Dict['parsing_status'] = 'finished'
|
||
|
||
airports_wo_city = []
|
||
# if airports_Dict and None in airports_Dict:
|
||
# airports_wo_city = airports_Dict[None]
|
||
|
||
return country_Dict, airports_wo_city
|
||
|
||
|
||
def osm_get_countries_n_cities_n_airports():
|
||
|
||
airports_wo_city = []
|
||
|
||
from ReferenceDataApp.funcs import get_countries_key_data, get_cities_by_country_name_en
|
||
db_countries = get_countries_key_data()
|
||
|
||
countries = osm_get_countries()
|
||
i = 0
|
||
while i < len(countries):
|
||
area_id = None
|
||
if countries[i]['name:en'] in db_countries.keys():
|
||
area_id = db_countries[countries[i]['name:en']]
|
||
countries[i], airports_wo_city_for_country = osm_get_country_w_cities_n_airports(countries[i], area_id)
|
||
if airports_wo_city_for_country:
|
||
airports_wo_city.extend(airports_wo_city_for_country)
|
||
|
||
return countries |