nominatim/clicmd/export.py

   1 # SPDX-License-Identifier: GPL-3.0-or-later
   2 #
   3 # This file is part of Nominatim. (https://nominatim.org)
   4 #
   5 # Copyright (C) 2023 by the Nominatim developer community.
   6 # For a full list of authors see the git log.
   7 """
   8 Implementation of the 'export' subcommand.
   9 """
  10 from typing import Optional, List, cast
  11 import logging
  12 import argparse
  13 import asyncio
  14 import csv
  15 import sys
  16
  17 import sqlalchemy as sa
  18
  19 from nominatim.clicmd.args import NominatimArgs
  20 import nominatim.api as napi
  21 from nominatim.api.results import create_from_placex_row, ReverseResult, add_result_details
  22 from nominatim.api.types import LookupDetails
  23 from nominatim.errors import UsageError
  24
  25 # Do not repeat documentation of subcommand classes.
  26 # pylint: disable=C0111
  27 # Using non-top-level imports to avoid eventually unused imports.
  28 # pylint: disable=E0012,C0415
  29 # Needed for SQLAlchemy
  30 # pylint: disable=singleton-comparison
  31
  32 LOG = logging.getLogger()
  33
  34 RANK_RANGE_MAP = {
  35   'country': (4, 4),
  36   'state': (5, 9),
  37   'county': (10, 12),
  38   'city': (13, 16),
  39   'suburb': (17, 21),
  40   'street': (26, 26),
  41   'path': (27, 27)
  42 }
  43
  44 RANK_TO_OUTPUT_MAP = {
  45     4: 'country',
  46     5: 'state', 6: 'state', 7: 'state', 8: 'state', 9: 'state',
  47     10: 'county', 11: 'county', 12: 'county',
  48     13: 'city', 14: 'city', 15: 'city', 16: 'city',
  49     17: 'suburb', 18: 'suburb', 19: 'suburb', 20: 'suburb', 21: 'suburb',
  50     26: 'street', 27: 'path'}
  51
  52 class QueryExport:
  53     """\
  54     Export places as CSV file from the database.
  55     """
  56
  57     def add_args(self, parser: argparse.ArgumentParser) -> None:
  58         group = parser.add_argument_group('Output arguments')
  59         group.add_argument('--output-type', default='street',
  60                            choices=('country', 'state', 'county',
  61                                     'city', 'suburb', 'street', 'path'),
  62                            help='Type of places to output (default: street)')
  63         group.add_argument('--output-format',
  64                            default='street;suburb;city;county;state;country',
  65                            help=("Semicolon-separated list of address types "
  66                                  "(see --output-type)."))
  67         group.add_argument('--language',
  68                            help=("Preferred language for output "
  69                                  "(use local name, if omitted)"))
  70         group = parser.add_argument_group('Filter arguments')
  71         group.add_argument('--restrict-to-country', metavar='COUNTRY_CODE',
  72                            help='Export only objects within country')
  73         group.add_argument('--restrict-to-osm-node', metavar='ID', type=int,
  74                            dest='node',
  75                            help='Export only children of this OSM node')
  76         group.add_argument('--restrict-to-osm-way', metavar='ID', type=int,
  77                            dest='way',
  78                            help='Export only children of this OSM way')
  79         group.add_argument('--restrict-to-osm-relation', metavar='ID', type=int,
  80                            dest='relation',
  81                            help='Export only children of this OSM relation')
  82
  83
  84     def run(self, args: NominatimArgs) -> int:
  85         return asyncio.run(export(args))
  86
  87
  88 async def export(args: NominatimArgs) -> int:
  89     """ The actual export as a asynchronous function.
  90     """
  91
  92     api = napi.NominatimAPIAsync(args.project_dir)
  93
  94     output_range = RANK_RANGE_MAP[args.output_type]
  95
  96     writer = init_csv_writer(args.output_format)
  97
  98     async with api.begin() as conn, api.begin() as detail_conn:
  99         t = conn.t.placex
 100
 101         sql = sa.select(t.c.place_id, t.c.osm_type, t.c.osm_id, t.c.name,
 102                     t.c.class_, t.c.type, t.c.admin_level,
 103                     t.c.address, t.c.extratags,
 104                     t.c.housenumber, t.c.postcode, t.c.country_code,
 105                     t.c.importance, t.c.wikipedia, t.c.indexed_date,
 106                     t.c.rank_address, t.c.rank_search,
 107                     t.c.centroid)\
 108                  .where(t.c.linked_place_id == None)\
 109                  .where(t.c.rank_address.between(*output_range))
 110
 111         parent_place_id = await get_parent_id(conn, args.node, args.way, args.relation)
 112         if parent_place_id:
 113             taddr = conn.t.addressline
 114
 115             sql = sql.join(taddr, taddr.c.place_id == t.c.place_id)\
 116                      .where(taddr.c.address_place_id == parent_place_id)\
 117                      .where(taddr.c.isaddress)
 118
 119         if args.restrict_to_country:
 120             sql = sql.where(t.c.country_code == args.restrict_to_country.lower())
 121
 122         results = []
 123         for row in await conn.execute(sql):
 124             result = create_from_placex_row(row, ReverseResult)
 125             if result is not None:
 126                 results.append(result)
 127
 128             if len(results) == 1000:
 129                 await dump_results(detail_conn, results, writer, args.language)
 130                 results = []
 131
 132         if results:
 133             await dump_results(detail_conn, results, writer, args.language)
 134
 135     return 0
 136
 137
 138 def init_csv_writer(output_format: str) -> 'csv.DictWriter[str]':
 139     fields = output_format.split(';')
 140     writer = csv.DictWriter(sys.stdout, fieldnames=fields, extrasaction='ignore')
 141     writer.writeheader()
 142
 143     return writer
 144
 145
 146 async def dump_results(conn: napi.SearchConnection,
 147                        results: List[ReverseResult],
 148                        writer: 'csv.DictWriter[str]',
 149                        lang: Optional[str]) -> None:
 150     await add_result_details(conn, results,
 151                              LookupDetails(address_details=True))
 152
 153
 154     locale = napi.Locales([lang] if lang else None)
 155
 156     for result in results:
 157         data = {'placeid': result.place_id,
 158                 'postcode': result.postcode}
 159
 160         result.localize(locale)
 161         for line in (result.address_rows or []):
 162             if line.isaddress and line.local_name\
 163                and line.rank_address in RANK_TO_OUTPUT_MAP:
 164                 data[RANK_TO_OUTPUT_MAP[line.rank_address]] = line.local_name
 165
 166         writer.writerow(data)
 167
 168
 169 async def get_parent_id(conn: napi.SearchConnection, node_id: Optional[int],
 170                         way_id: Optional[int],
 171                         relation_id: Optional[int]) -> Optional[int]:
 172     """ Get the place ID for the given OSM object.
 173     """
 174     if node_id is not None:
 175         osm_type, osm_id = 'N', node_id
 176     elif way_id is not None:
 177         osm_type, osm_id = 'W', way_id
 178     elif relation_id is not None:
 179         osm_type, osm_id = 'R', relation_id
 180     else:
 181         return None
 182
 183     t = conn.t.placex
 184     sql = sa.select(t.c.place_id).limit(1)\
 185             .where(t.c.osm_type == osm_type)\
 186             .where(t.c.osm_id == osm_id)\
 187             .where(t.c.rank_address > 0)\
 188             .order_by(t.c.rank_address)
 189
 190     for result in await conn.execute(sql):
 191         return cast(int, result[0])
 192
 193     raise UsageError(f'Cannot find a place {osm_type}{osm_id}.')