nominatim/tools/special_phrases/sp_importer.py

   1 # SPDX-License-Identifier: GPL-2.0-only
   2 #
   3 # This file is part of Nominatim. (https://nominatim.org)
   4 #
   5 # Copyright (C) 2022 by the Nominatim developer community.
   6 # For a full list of authors see the git log.
   7 """
   8     Module containing the class handling the import
   9     of the special phrases.
  10
  11     Phrases are analyzed and imported into the database.
  12
  13     The phrases already present in the database which are not
  14     valids anymore are removed.
  15 """
  16 import logging
  17 import re
  18
  19 from psycopg2.sql import Identifier, SQL
  20 from nominatim.tools.special_phrases.importer_statistics import SpecialPhrasesImporterStatistics
  21
  22 LOG = logging.getLogger()
  23
  24 def _classtype_table(phrase_class, phrase_type):
  25     """ Return the name of the table for the given class and type.
  26     """
  27     return f'place_classtype_{phrase_class}_{phrase_type}'
  28
  29 class SPImporter():
  30     # pylint: disable-msg=too-many-instance-attributes
  31     """
  32         Class handling the process of special phrases importation into the database.
  33
  34         Take a sp loader which load the phrases from an external source.
  35     """
  36     def __init__(self, config, db_connection, sp_loader) -> None:
  37         self.config = config
  38         self.db_connection = db_connection
  39         self.sp_loader = sp_loader
  40         self.statistics_handler = SpecialPhrasesImporterStatistics()
  41         self.black_list, self.white_list = self._load_white_and_black_lists()
  42         self.sanity_check_pattern = re.compile(r'^\w+$')
  43         # This set will contain all existing phrases to be added.
  44         # It contains tuples with the following format: (lable, class, type, operator)
  45         self.word_phrases = set()
  46         # This set will contain all existing place_classtype tables which doesn't match any
  47         # special phrases class/type on the wiki.
  48         self.table_phrases_to_delete = set()
  49
  50     def import_phrases(self, tokenizer, should_replace):
  51         """
  52             Iterate through all SpecialPhrases extracted from the
  53             loader and import them into the database.
  54
  55             If should_replace is set to True only the loaded phrases
  56             will be kept into the database. All other phrases already
  57             in the database will be removed.
  58         """
  59         LOG.warning('Special phrases importation starting')
  60         self._fetch_existing_place_classtype_tables()
  61
  62         # Store pairs of class/type for further processing
  63         class_type_pairs = set()
  64
  65         for loaded_phrases in self.sp_loader:
  66             for phrase in loaded_phrases:
  67                 result = self._process_phrase(phrase)
  68                 if result:
  69                     class_type_pairs.add(result)
  70
  71         self._create_place_classtype_table_and_indexes(class_type_pairs)
  72         if should_replace:
  73             self._remove_non_existent_tables_from_db()
  74         self.db_connection.commit()
  75
  76         with tokenizer.name_analyzer() as analyzer:
  77             analyzer.update_special_phrases(self.word_phrases, should_replace)
  78
  79         LOG.warning('Import done.')
  80         self.statistics_handler.notify_import_done()
  81
  82
  83     def _fetch_existing_place_classtype_tables(self):
  84         """
  85             Fetch existing place_classtype tables.
  86             Fill the table_phrases_to_delete set of the class.
  87         """
  88         query = """
  89             SELECT table_name
  90             FROM information_schema.tables
  91             WHERE table_schema='public'
  92             AND table_name like 'place_classtype_%';
  93         """
  94         with self.db_connection.cursor() as db_cursor:
  95             db_cursor.execute(SQL(query))
  96             for row in db_cursor:
  97                 self.table_phrases_to_delete.add(row[0])
  98
  99     def _load_white_and_black_lists(self):
 100         """
 101             Load white and black lists from phrases-settings.json.
 102         """
 103         settings = self.config.load_sub_configuration('phrase-settings.json')
 104
 105         return settings['blackList'], settings['whiteList']
 106
 107     def _check_sanity(self, phrase):
 108         """
 109             Check sanity of given inputs in case somebody added garbage in the wiki.
 110             If a bad class/type is detected the system will exit with an error.
 111         """
 112         class_matchs = self.sanity_check_pattern.findall(phrase.p_class)
 113         type_matchs = self.sanity_check_pattern.findall(phrase.p_type)
 114
 115         if not class_matchs or not type_matchs:
 116             LOG.warning("Bad class/type: %s=%s. It will not be imported",
 117                         phrase.p_class, phrase.p_type)
 118             return False
 119         return True
 120
 121     def _process_phrase(self, phrase):
 122         """
 123             Processes the given phrase by checking black and white list
 124             and sanity.
 125             Return the class/type pair corresponding to the phrase.
 126         """
 127
 128         # blacklisting: disallow certain class/type combinations
 129         if phrase.p_class in self.black_list.keys() \
 130            and phrase.p_type in self.black_list[phrase.p_class]:
 131             return None
 132
 133         # whitelisting: if class is in whitelist, allow only tags in the list
 134         if phrase.p_class in self.white_list.keys() \
 135            and phrase.p_type not in self.white_list[phrase.p_class]:
 136             return None
 137
 138         # sanity check, in case somebody added garbage in the wiki
 139         if not self._check_sanity(phrase):
 140             self.statistics_handler.notify_one_phrase_invalid()
 141             return None
 142
 143         self.word_phrases.add((phrase.p_label, phrase.p_class,
 144                                phrase.p_type, phrase.p_operator))
 145
 146         return (phrase.p_class, phrase.p_type)
 147
 148
 149     def _create_place_classtype_table_and_indexes(self, class_type_pairs):
 150         """
 151             Create table place_classtype for each given pair.
 152             Also create indexes on place_id and centroid.
 153         """
 154         LOG.warning('Create tables and indexes...')
 155
 156         sql_tablespace = self.config.TABLESPACE_AUX_DATA
 157         if sql_tablespace:
 158             sql_tablespace = ' TABLESPACE ' + sql_tablespace
 159
 160         with self.db_connection.cursor() as db_cursor:
 161             db_cursor.execute("CREATE INDEX idx_placex_classtype ON placex (class, type)")
 162
 163         for pair in class_type_pairs:
 164             phrase_class = pair[0]
 165             phrase_type = pair[1]
 166
 167             table_name = _classtype_table(phrase_class, phrase_type)
 168
 169             if table_name in self.table_phrases_to_delete:
 170                 self.statistics_handler.notify_one_table_ignored()
 171                 # Remove this table from the ones to delete as it match a
 172                 # class/type still existing on the special phrases of the wiki.
 173                 self.table_phrases_to_delete.remove(table_name)
 174                 # So don't need to create the table and indexes.
 175                 continue
 176
 177             # Table creation
 178             self._create_place_classtype_table(sql_tablespace, phrase_class, phrase_type)
 179
 180             # Indexes creation
 181             self._create_place_classtype_indexes(sql_tablespace, phrase_class, phrase_type)
 182
 183             # Grant access on read to the web user.
 184             self._grant_access_to_webuser(phrase_class, phrase_type)
 185
 186             self.statistics_handler.notify_one_table_created()
 187
 188         with self.db_connection.cursor() as db_cursor:
 189             db_cursor.execute("DROP INDEX idx_placex_classtype")
 190
 191
 192     def _create_place_classtype_table(self, sql_tablespace, phrase_class, phrase_type):
 193         """
 194             Create table place_classtype of the given phrase_class/phrase_type if doesn't exit.
 195         """
 196         table_name = _classtype_table(phrase_class, phrase_type)
 197         with self.db_connection.cursor() as db_cursor:
 198             db_cursor.execute(SQL("""CREATE TABLE IF NOT EXISTS {} {} AS
 199                                        SELECT place_id AS place_id,
 200                                               st_centroid(geometry) AS centroid
 201                                        FROM placex
 202                                        WHERE class = %s AND type = %s""")
 203                                  .format(Identifier(table_name), SQL(sql_tablespace)),
 204                               (phrase_class, phrase_type))
 205
 206
 207     def _create_place_classtype_indexes(self, sql_tablespace, phrase_class, phrase_type):
 208         """
 209             Create indexes on centroid and place_id for the place_classtype table.
 210         """
 211         index_prefix = f'idx_place_classtype_{phrase_class}_{phrase_type}_'
 212         base_table = _classtype_table(phrase_class, phrase_type)
 213         # Index on centroid
 214         if not self.db_connection.index_exists(index_prefix + 'centroid'):
 215             with self.db_connection.cursor() as db_cursor:
 216                 db_cursor.execute(SQL("CREATE INDEX {} ON {} USING GIST (centroid) {}")
 217                                      .format(Identifier(index_prefix + 'centroid'),
 218                                              Identifier(base_table),
 219                                              SQL(sql_tablespace)))
 220
 221         # Index on place_id
 222         if not self.db_connection.index_exists(index_prefix + 'place_id'):
 223             with self.db_connection.cursor() as db_cursor:
 224                 db_cursor.execute(SQL("CREATE INDEX {} ON {} USING btree(place_id) {}")
 225                                   .format(Identifier(index_prefix + 'place_id'),
 226                                           Identifier(base_table),
 227                                           SQL(sql_tablespace)))
 228
 229
 230     def _grant_access_to_webuser(self, phrase_class, phrase_type):
 231         """
 232             Grant access on read to the table place_classtype for the webuser.
 233         """
 234         table_name = _classtype_table(phrase_class, phrase_type)
 235         with self.db_connection.cursor() as db_cursor:
 236             db_cursor.execute(SQL("""GRANT SELECT ON {} TO {}""")
 237                               .format(Identifier(table_name),
 238                                       Identifier(self.config.DATABASE_WEBUSER)))
 239
 240     def _remove_non_existent_tables_from_db(self):
 241         """
 242             Remove special phrases which doesn't exist on the wiki anymore.
 243             Delete the place_classtype tables.
 244         """
 245         LOG.warning('Cleaning database...')
 246
 247         # Delete place_classtype tables corresponding to class/type which
 248         # are not on the wiki anymore.
 249         with self.db_connection.cursor() as db_cursor:
 250             for table in self.table_phrases_to_delete:
 251                 self.statistics_handler.notify_one_table_deleted()
 252                 db_cursor.drop_table(table)