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