#!/usr/bin/env python3 # -*- coding: utf-8 -*- # SPDX-FileCopyrightText: © 2017 Robin Schneider # SPDX-License-Identifier: AGPL-3.0-only """ Read _nominatim_url, get the query response and store it as YAML files below ./nominatim_cache. """ from __future__ import absolute_import, division, print_function __version__ = '0.1.0' __license__ = 'AGPL-3.0' __author__ = 'Robin Schneider ' __copyright__ = 'Copyright (C) 2017 Robin Schneider ' __all__ = ['main'] import sys import os import re import json import logging import argparse from glob import iglob from collections import OrderedDict from copy import deepcopy if sys.version_info[0] == 2: from io import open from ruamel import yaml # import requests import urllib.request import urllib.parse from hc.yaml import dump_holidays_as_yaml LOG = logging.getLogger(__name__) NOMINATIM_SORTING = { 'place_id': '10', 'licence': '20', 'osm_type': '30', 'osm_id': '40', 'lat': '50', 'lon': '60', 'display_name': '70', 'address': '80', } KEEP_STATE_LEVEL_ADDRESS_ATTRS = [ 'country', 'country_code', ] def handle_country_holiday_def_file(def_file): c_def = yaml.safe_load(open(def_file, 'r', encoding='utf-8')) if type(c_def.get('_nominatim_url')) == str: handle_nominatim_url(c_def['_nominatim_url'], def_file) for r_name, r_def in c_def.items(): if type(r_def) == dict and type(r_def.get('_nominatim_url')) == str: if any(c.isupper() for c in str(r_def.get('_state_code', ''))): raise Exception("Found upper case _state_code {} in file {}!".format( r_def.get('_state_code', ''), def_file, )) handle_nominatim_url( r_def['_nominatim_url'], def_file, state=r_def.get('_state_code', r_name), ) def handle_nominatim_url(nominatim_url, def_file, state=None): LOG.debug('Calling handle_nominatim_url for {}, state: {}'.format( nominatim_url, state)) level = 'state' if type(state) in [str, int] else 'country' country_code_re = re.match(r'.*?(?P\w+)\.', def_file) country_code = country_code_re.group('cc') cache_file = '{}{}.yaml'.format( country_code, '_' + str(state) if level == 'state' else '', ) nominatim_file = os.path.join('nominatim_cache', cache_file) LOG.debug('nominatim_file: {}'.format(nominatim_file)) if not os.path.exists(nominatim_file): LOG.info('Loading {}'.format(nominatim_url)) # Caused http.client.BadStatusLine exception with wired encoding in them. # nominatim_data = json.loads(requests.get(nominatim_url).text) req = urllib.request.Request(nominatim_url, headers={ 'User-Agent': 'curl/7.38.0'}) nominatim_data = json.loads( urllib.request.urlopen(req).read().decode('utf-8')) if isinstance(nominatim_data, list): # We expect that the URL was for geocoding and not reverse geocoding. parsed_url = urllib.parse.urlparse(nominatim_url) url_qs = urllib.parse.parse_qs(parsed_url.query) reverse_geocoding_qs = { 'format': 'json', 'lat': nominatim_data[0]['lat'], 'lon': nominatim_data[0]['lon'], 'zoom': 18, 'addressdetails': 1, 'accept-language': url_qs['accept-language'], } parsed_url = list(parsed_url) parsed_url[2] = '/reverse' parsed_url[4] = urllib.parse.urlencode( reverse_geocoding_qs, doseq=True, safe=",") correct_nominatim_url = urllib.parse.urlunparse(parsed_url) return handle_nominatim_url(correct_nominatim_url, def_file, state) nominatim_data = OrderedDict(sorted( nominatim_data.items(), key=lambda k: NOMINATIM_SORTING.get(k[0], k[0]))) if level == 'country': nominatim_data = yaml.load( dump_holidays_as_yaml(nominatim_data, add_vspacing=False), Loader=yaml.RoundTripLoader, ) nominatim_old_address = deepcopy(nominatim_data['address']) nominatim_data['address'] = OrderedDict() address_attrs_commented_out = [] for address_attr in sorted(nominatim_old_address.keys()): if address_attr in KEEP_STATE_LEVEL_ADDRESS_ATTRS: nominatim_data['address'][address_attr] = nominatim_old_address[address_attr] else: address_attrs_commented_out.append('{}: {}'.format( address_attr, nominatim_old_address[address_attr], )) nominatim_data.yaml_set_comment_before_after_key( 'boundingbox', before='\n'.join(address_attrs_commented_out), indent=2, ) nominatim_data['address'] = OrderedDict( sorted(nominatim_data['address'].items())) nominatim_yaml = dump_holidays_as_yaml( nominatim_data, add_vspacing=False) LOG.debug(nominatim_yaml) with open(nominatim_file, 'w') as nominatim_fh: LOG.info('Writing file {}'.format(nominatim_file)) nominatim_fh.write(nominatim_yaml) def get_args_parser(): args_parser = argparse.ArgumentParser( description=__doc__, # epilog=__doc__, ) args_parser.add_argument( '-V', '--version', action='version', version=__version__, ) args_parser.add_argument( '-d', '--debug', help="Write debugging and higher to STDOUT|STDERR.", action="store_const", dest="loglevel", const=logging.DEBUG, ) args_parser.add_argument( '-v', '--verbose', help="Write information and higher to STDOUT|STDERR.", action="store_const", dest="loglevel", const=logging.INFO, ) args_parser.add_argument( '-q', '--quiet', '--silent', help="Only write errors and higher to STDOUT|STDERR.", action="store_const", dest="loglevel", const=logging.ERROR, ) args_parser.add_argument( '-n', '--no-cache', help="Do not cache intermediary files.", action="store_false", dest="cache", ) args_parser.add_argument( '-c', '--cache-dir', help="Cache directory, defaults to the default cache directory of your operating system.", ) args_parser.add_argument( '-i', '--input-file', help="File path to the input file to process." " '-' will read from STDIN.", ) return args_parser def main(): # pylint: disable=too-many-branches args_parser = get_args_parser() args = args_parser.parse_args() if args.loglevel is None: args.loglevel = logging.WARNING logging.basicConfig( format='%(levelname)s{}: %(message)s'.format( ' (%(filename)s:%(lineno)s)' if args.loglevel <= logging.DEBUG else '', ), level=args.loglevel, ) if args.cache: import requests_cache requests_cache.install_cache('requests_cache') if args.input_file: handle_country_holiday_def_file(args.input_file) else: for def_file in iglob('*.yaml'): handle_country_holiday_def_file(def_file) if __name__ == '__main__': main()