populate_nominatim_cache

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""
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 <ypid@riseup.net>'
__copyright__ = 'Copyright (C) 2017-2020 Robin Schneider <ypid@riseup.net>'

__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<cc>\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()