# -*- 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