+# SPDX-License-Identifier: GPL-2.0-only
+#
+# This file is part of Nominatim. (https://nominatim.org)
+#
+# Copyright (C) 2022 by the Nominatim developer community.
+# For a full list of authors see the git log.
"""
Module containing the class handling the import
of the special phrases.
valids anymore are removed.
"""
import logging
-import os
-from os.path import isfile
-from pathlib import Path
import re
-import subprocess
-import json
-from psycopg2.sql import Identifier, Literal, SQL
-from nominatim.errors import UsageError
+from psycopg2.sql import Identifier, SQL
from nominatim.tools.special_phrases.importer_statistics import SpecialPhrasesImporterStatistics
LOG = logging.getLogger()
Take a sp loader which load the phrases from an external source.
"""
- def __init__(self, config, phplib_dir, db_connection, sp_loader) -> None:
+ def __init__(self, config, db_connection, sp_loader) -> None:
self.config = config
- self.phplib_dir = phplib_dir
self.db_connection = db_connection
self.sp_loader = sp_loader
self.statistics_handler = SpecialPhrasesImporterStatistics()
# Store pairs of class/type for further processing
class_type_pairs = set()
- for loaded_phrases in self.sp_loader:
- for phrase in loaded_phrases:
- result = self._process_phrase(phrase)
- if result:
- class_type_pairs.add(result)
+ for phrase in self.sp_loader.generate_phrases():
+ result = self._process_phrase(phrase)
+ if result:
+ class_type_pairs.add(result)
self._create_place_classtype_table_and_indexes(class_type_pairs)
if should_replace:
"""
Load white and black lists from phrases-settings.json.
"""
- settings_path = (self.config.config_dir / 'phrase-settings.json').resolve()
+ settings = self.config.load_sub_configuration('phrase-settings.json')
- if self.config.PHRASE_CONFIG:
- settings_path = self._convert_php_settings_if_needed(self.config.PHRASE_CONFIG)
-
- with settings_path.open("r") as json_settings:
- settings = json.load(json_settings)
return settings['blackList'], settings['whiteList']
def _check_sanity(self, phrase):
def _create_place_classtype_table(self, sql_tablespace, phrase_class, phrase_type):
"""
- Create table place_classtype of the given phrase_class/phrase_type if doesn't exit.
+ Create table place_classtype of the given phrase_class/phrase_type
+ if doesn't exit.
"""
table_name = _classtype_table(phrase_class, phrase_type)
- with self.db_connection.cursor() as db_cursor:
- db_cursor.execute(SQL("""
- CREATE TABLE IF NOT EXISTS {{}} {}
- AS SELECT place_id AS place_id,st_centroid(geometry) AS centroid FROM placex
- WHERE class = {{}} AND type = {{}}""".format(sql_tablespace))
- .format(Identifier(table_name), Literal(phrase_class),
- Literal(phrase_type)))
+ with self.db_connection.cursor() as cur:
+ cur.execute(SQL("""CREATE TABLE IF NOT EXISTS {} {} AS
+ SELECT place_id AS place_id,
+ st_centroid(geometry) AS centroid
+ FROM placex
+ WHERE class = %s AND type = %s
+ """).format(Identifier(table_name), SQL(sql_tablespace)),
+ (phrase_class, phrase_type))
def _create_place_classtype_indexes(self, sql_tablespace, phrase_class, phrase_type):
"""
Create indexes on centroid and place_id for the place_classtype table.
"""
- index_prefix = 'idx_place_classtype_{}_{}_'.format(phrase_class, phrase_type)
+ index_prefix = f'idx_place_classtype_{phrase_class}_{phrase_type}_'
base_table = _classtype_table(phrase_class, phrase_type)
# Index on centroid
if not self.db_connection.index_exists(index_prefix + 'centroid'):
with self.db_connection.cursor() as db_cursor:
- db_cursor.execute(SQL("""
- CREATE INDEX {{}} ON {{}} USING GIST (centroid) {}""".format(sql_tablespace))
+ db_cursor.execute(SQL("CREATE INDEX {} ON {} USING GIST (centroid) {}")
.format(Identifier(index_prefix + 'centroid'),
- Identifier(base_table)), sql_tablespace)
+ Identifier(base_table),
+ SQL(sql_tablespace)))
# Index on place_id
if not self.db_connection.index_exists(index_prefix + 'place_id'):
with self.db_connection.cursor() as db_cursor:
- db_cursor.execute(SQL(
- """CREATE INDEX {{}} ON {{}} USING btree(place_id) {}""".format(sql_tablespace))
+ db_cursor.execute(SQL("CREATE INDEX {} ON {} USING btree(place_id) {}")
.format(Identifier(index_prefix + 'place_id'),
- Identifier(base_table)))
+ Identifier(base_table),
+ SQL(sql_tablespace)))
def _grant_access_to_webuser(self, phrase_class, phrase_type):
for table in self.table_phrases_to_delete:
self.statistics_handler.notify_one_table_deleted()
db_cursor.drop_table(table)
-
-
- def _convert_php_settings_if_needed(self, file_path):
- """
- Convert php settings file of special phrases to json file if it is still in php format.
- """
- if not isfile(file_path):
- raise UsageError(str(file_path) + ' is not a valid file.')
-
- file, extension = os.path.splitext(file_path)
- json_file_path = Path(file + '.json').resolve()
-
- if extension not in ('.php', '.json'):
- raise UsageError('The custom NOMINATIM_PHRASE_CONFIG file has not a valid extension.')
-
- if extension == '.php' and not isfile(json_file_path):
- try:
- subprocess.run(['/usr/bin/env', 'php', '-Cq',
- (self.phplib_dir / 'migration/PhraseSettingsToJson.php').resolve(),
- file_path], check=True)
- LOG.warning('special_phrase configuration file has been converted to json.')
- except subprocess.CalledProcessError:
- LOG.error('Error while converting %s to json.', file_path)
- raise
-
- return json_file_path
projects / nominatim.git / blobdiff