src/nominatim_api/v1/format.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) 2024 by the Nominatim developer community.
   6 # For a full list of authors see the git log.
   7 """
   8 Output formatters for API version v1.
   9 """
  10 from typing import List, Dict, Mapping, Any
  11 import collections
  12 import datetime as dt
  13
  14 from ..utils.json_writer import JsonWriter
  15 from ..status import StatusResult
  16 from ..results import DetailedResult, ReverseResults, SearchResults, \
  17                       AddressLines, AddressLine
  18 from ..localization import Locales
  19 from ..result_formatting import FormatDispatcher
  20 from .classtypes import ICONS
  21 from . import format_json, format_xml
  22 from .. import logging as loglib
  23 from ..server import content_types as ct
  24
  25
  26 class RawDataList(List[Dict[str, Any]]):
  27     """ Data type for formatting raw data lists 'as is' in json.
  28     """
  29
  30
  31 dispatch = FormatDispatcher({'text': ct.CONTENT_TEXT,
  32                              'xml': ct.CONTENT_XML,
  33                              'debug': ct.CONTENT_HTML})
  34
  35
  36 @dispatch.error_format_func
  37 def _format_error(content_type: str, msg: str, status: int) -> str:
  38     if content_type == ct.CONTENT_XML:
  39         return f"""<?xml version="1.0" encoding="UTF-8" ?>
  40                    <error>
  41                      <code>{status}</code>
  42                      <message>{msg}</message>
  43                    </error>
  44                 """
  45
  46     if content_type == ct.CONTENT_JSON:
  47         return f"""{{"error":{{"code":{status},"message":"{msg}"}}}}"""
  48
  49     if content_type == ct.CONTENT_HTML:
  50         loglib.log().section('Execution error')
  51         loglib.log().var_dump('Status', status)
  52         loglib.log().var_dump('Message', msg)
  53         return loglib.get_and_disable()
  54
  55     return f"ERROR {status}: {msg}"
  56
  57
  58 @dispatch.format_func(StatusResult, 'text')
  59 def _format_status_text(result: StatusResult, _: Mapping[str, Any]) -> str:
  60     if result.status:
  61         return f"ERROR: {result.message}"
  62
  63     return 'OK'
  64
  65
  66 @dispatch.format_func(StatusResult, 'json')
  67 def _format_status_json(result: StatusResult, _: Mapping[str, Any]) -> str:
  68     out = JsonWriter()
  69
  70     out.start_object()\
  71         .keyval('status', result.status)\
  72         .keyval('message', result.message)\
  73         .keyval_not_none('data_updated', result.data_updated,
  74                          lambda v: v.isoformat())\
  75         .keyval('software_version', str(result.software_version))\
  76         .keyval_not_none('database_version', result.database_version, str)\
  77         .end_object()
  78
  79     return out()
  80
  81
  82 def _add_address_row(writer: JsonWriter, row: AddressLine,
  83                      locales: Locales) -> None:
  84     writer.start_object()\
  85             .keyval('localname', locales.display_name(row.names))\
  86             .keyval_not_none('place_id', row.place_id)
  87
  88     if row.osm_object is not None:
  89         writer.keyval('osm_id', row.osm_object[1])\
  90               .keyval('osm_type', row.osm_object[0])
  91
  92     if row.extratags:
  93         writer.keyval_not_none('place_type', row.extratags.get('place_type'))
  94
  95     writer.keyval('class', row.category[0])\
  96           .keyval('type', row.category[1])\
  97           .keyval_not_none('admin_level', row.admin_level)\
  98           .keyval('rank_address', row.rank_address)\
  99           .keyval('distance', row.distance)\
 100           .keyval('isaddress', row.isaddress)\
 101         .end_object()
 102
 103
 104 def _add_address_rows(writer: JsonWriter, section: str, rows: AddressLines,
 105                       locales: Locales) -> None:
 106     writer.key(section).start_array()
 107     for row in rows:
 108         _add_address_row(writer, row, locales)
 109         writer.next()
 110     writer.end_array().next()
 111
 112
 113 def _add_parent_rows_grouped(writer: JsonWriter, rows: AddressLines,
 114                              locales: Locales) -> None:
 115     # group by category type
 116     data = collections.defaultdict(list)
 117     for row in rows:
 118         sub = JsonWriter()
 119         _add_address_row(sub, row, locales)
 120         data[row.category[1]].append(sub())
 121
 122     writer.key('hierarchy').start_object()
 123     for group, grouped in data.items():
 124         writer.key(group).start_array()
 125         grouped.sort()  # sorts alphabetically by local name
 126         for line in grouped:
 127             writer.raw(line).next()
 128         writer.end_array().next()
 129
 130     writer.end_object().next()
 131
 132
 133 @dispatch.format_func(DetailedResult, 'json')
 134 def _format_details_json(result: DetailedResult, options: Mapping[str, Any]) -> str:
 135     locales = options.get('locales', Locales())
 136     geom = result.geometry.get('geojson')
 137     centroid = result.centroid.to_geojson()
 138
 139     out = JsonWriter()
 140     out.start_object()\
 141         .keyval_not_none('place_id', result.place_id)\
 142         .keyval_not_none('parent_place_id', result.parent_place_id)
 143
 144     if result.osm_object is not None:
 145         out.keyval('osm_type', result.osm_object[0])\
 146            .keyval('osm_id', result.osm_object[1])
 147
 148     out.keyval('category', result.category[0])\
 149        .keyval('type', result.category[1])\
 150        .keyval('admin_level', result.admin_level)\
 151        .keyval('localname', result.locale_name or '')\
 152        .keyval('names', result.names or {})\
 153        .keyval('addresstags', result.address or {})\
 154        .keyval_not_none('housenumber', result.housenumber)\
 155        .keyval_not_none('calculated_postcode', result.postcode)\
 156        .keyval_not_none('country_code', result.country_code)\
 157        .keyval_not_none('indexed_date', result.indexed_date, lambda v: v.isoformat())\
 158        .keyval_not_none('importance', result.importance)\
 159        .keyval('calculated_importance', result.calculated_importance())\
 160        .keyval('extratags', result.extratags or {})\
 161        .keyval_not_none('calculated_wikipedia', result.wikipedia)\
 162        .keyval('rank_address', result.rank_address)\
 163        .keyval('rank_search', result.rank_search)\
 164        .keyval('isarea', 'Polygon' in (geom or result.geometry.get('type') or ''))\
 165        .key('centroid').raw(centroid).next()\
 166        .key('geometry').raw(geom or centroid).next()
 167
 168     if options.get('icon_base_url', None):
 169         icon = ICONS.get(result.category)
 170         if icon:
 171             out.keyval('icon', f"{options['icon_base_url']}/{icon}.p.20.png")
 172
 173     if result.address_rows is not None:
 174         _add_address_rows(out, 'address', result.address_rows, locales)
 175
 176     if result.linked_rows:
 177         _add_address_rows(out, 'linked_places', result.linked_rows, locales)
 178
 179     if result.name_keywords is not None or result.address_keywords is not None:
 180         out.key('keywords').start_object()
 181
 182         for sec, klist in (('name', result.name_keywords), ('address', result.address_keywords)):
 183             out.key(sec).start_array()
 184             for word in (klist or []):
 185                 out.start_object()\
 186                      .keyval('id', word.word_id)\
 187                      .keyval('token', word.word_token)\
 188                    .end_object().next()
 189             out.end_array().next()
 190
 191         out.end_object().next()
 192
 193     if result.parented_rows is not None:
 194         if options.get('group_hierarchy', False):
 195             _add_parent_rows_grouped(out, result.parented_rows, locales)
 196         else:
 197             _add_address_rows(out, 'hierarchy', result.parented_rows, locales)
 198
 199     if options.get('entrances', False):
 200         format_json.write_entrances(out, result.entrances)
 201
 202     out.end_object()
 203
 204     return out()
 205
 206
 207 @dispatch.format_func(ReverseResults, 'xml')
 208 def _format_reverse_xml(results: ReverseResults, options: Mapping[str, Any]) -> str:
 209     return format_xml.format_base_xml(results,
 210                                       options, True, 'reversegeocode',
 211                                       {'querystring': options.get('query', '')})
 212
 213
 214 @dispatch.format_func(ReverseResults, 'geojson')
 215 def _format_reverse_geojson(results: ReverseResults,
 216                             options: Mapping[str, Any]) -> str:
 217     return format_json.format_base_geojson(results, options, True)
 218
 219
 220 @dispatch.format_func(ReverseResults, 'geocodejson')
 221 def _format_reverse_geocodejson(results: ReverseResults,
 222                                 options: Mapping[str, Any]) -> str:
 223     return format_json.format_base_geocodejson(results, options, True)
 224
 225
 226 @dispatch.format_func(ReverseResults, 'json')
 227 def _format_reverse_json(results: ReverseResults,
 228                          options: Mapping[str, Any]) -> str:
 229     return format_json.format_base_json(results, options, True,
 230                                         class_label='class')
 231
 232
 233 @dispatch.format_func(ReverseResults, 'jsonv2')
 234 def _format_reverse_jsonv2(results: ReverseResults,
 235                            options: Mapping[str, Any]) -> str:
 236     return format_json.format_base_json(results, options, True,
 237                                         class_label='category')
 238
 239
 240 @dispatch.format_func(SearchResults, 'xml')
 241 def _format_search_xml(results: SearchResults, options: Mapping[str, Any]) -> str:
 242     extra = {'querystring': options.get('query', '')}
 243     for attr in ('more_url', 'exclude_place_ids', 'viewbox'):
 244         if options.get(attr):
 245             extra[attr] = options[attr]
 246     return format_xml.format_base_xml(results, options, False, 'searchresults',
 247                                       extra)
 248
 249
 250 @dispatch.format_func(SearchResults, 'geojson')
 251 def _format_search_geojson(results: SearchResults,
 252                            options: Mapping[str, Any]) -> str:
 253     return format_json.format_base_geojson(results, options, False)
 254
 255
 256 @dispatch.format_func(SearchResults, 'geocodejson')
 257 def _format_search_geocodejson(results: SearchResults,
 258                                options: Mapping[str, Any]) -> str:
 259     return format_json.format_base_geocodejson(results, options, False)
 260
 261
 262 @dispatch.format_func(SearchResults, 'json')
 263 def _format_search_json(results: SearchResults,
 264                         options: Mapping[str, Any]) -> str:
 265     return format_json.format_base_json(results, options, False,
 266                                         class_label='class')
 267
 268
 269 @dispatch.format_func(SearchResults, 'jsonv2')
 270 def _format_search_jsonv2(results: SearchResults,
 271                           options: Mapping[str, Any]) -> str:
 272     return format_json.format_base_json(results, options, False,
 273                                         class_label='category')
 274
 275
 276 @dispatch.format_func(RawDataList, 'json')
 277 def _format_raw_data_json(results: RawDataList,  _: Mapping[str, Any]) -> str:
 278     out = JsonWriter()
 279     out.start_array()
 280     for res in results:
 281         out.start_object()
 282         for k, v in res.items():
 283             if isinstance(v, dt.datetime):
 284                 out.keyval(k, v.isoformat(sep=' ', timespec='seconds'))
 285             else:
 286                 out.keyval(k, v)
 287         out.end_object().next()
 288
 289     out.end_array()
 290
 291     return out()