How to become an author
Search
Write a publication
Pull to refresh
431.4
Rating
Тензор
Разработчик системы СБИС

КЛАДРируем адреса произвольной формы (ч.1 — импорт)

Reading time25 min
Views6.6K
Тензор corporate blogSystem administration*PostgreSQL*SQL*ERP-systems*
Tutorial

Достаточно часто при работе с вводимыми пользователем адресными данными возникает необходимость сначала подсказать ему, как правильно свой адрес указать, а потом - полученную строку адреса привести в некоторый машинно-читаемый вид.

Таким видом у нас в стране можно считать код по справочникам КЛАДР или ФИАС.

Первый из них уже несколько лет считается устаревающим, но отличается более простой структурой и исторически продолжает использоваться во множестве систем, поскольку вполне подходит для большинства задач.

Давайте научимся разбирать строку адреса "туда и обратно", а заодно познакомимся с некоторыми алгоритмическими подходами и их реализацией на SQL.

Получение справочника КЛАДР

База КЛАДР в настоящее время администрируется ФНС и представлена на сайте ГНИВЦ в виде периодически (примерно раз в неделю) обновляемого архива. Для начала мы научимся его скачивать, исправлять некоторые ошибки и преобразовывать в более подходящую для наших задач структуру.

Для этого нам понадобятся wget, p7zip, dbview, recode и psql.

Исходный архив

База выкладывается на сервере ГНИВЦ всегда по одному и тому же адресу в виде 7z-архива Base.7z - версии отличаются друг от друга только заголовком Last-Modified в ответе сервера.

При этом сервер достаточно нервно реагирует на скорость загрузки и не поддерживает HEAD-запросы, увы. Поэтому нам придется скачивать архив ежесуточно полностью, но чтобы снизить нагрузку на сервер установим ограничение скорости, а заголовки ответа сохраним и разберем отдельно:

wget -S https://gnivc.ru/html/gnivcsoft/KLADR/Base.7z --limit-rate=8k 2>.hdr

В сгенерированном .hdr нас будет интересовать эта строка:

...
  Last-Modified: Thu, 05 Aug 2021 09:28:39 GMT
...

Чтобы не пытаться обрабатывать архив и обновлять данные в нашей базе повторно, будем сохранять в ней этот таймстамп для последующих сравнений.

Если же таймстамп файла не совпал с сохраненным, распакуем полученный архив:

p7zip -d Base.7z

Результатом будет 7 DBF-файлов в DOS-кодировке:

Extracting  ALTNAMES.DBF
Extracting  DOMA.DBF
Extracting  FLAT.DBF
Extracting  KLADR.DBF
Extracting  NAMEMAP.DBF
Extracting  SOCRBASE.DBF
Extracting  STREET.DBF

Переберем все эти файлы, формируя единый скрипт выгрузки данных через psql в COPY-формате:

# получаем структуру полей DBF
#   | DOS2WIN
#   | берем только описания полей (skip 2 строки)
#   | оставляем только их имена
dbview -b -t -e -o -r ALTNAMES.DBF \
  | recode CP866..CP1251 \
  | tail -n+2 \
  | xargs -l \
  | egrep -io "^[a-z0-9_]+"

# ... формируем SQL-заголовок временной таблицы

# получаем данные DBF, разделенные '~'
#   | склеиваем "висящие" строки ([\t\r\n] в теле поля данных)
#   | DOS2WIN
#   | убираем все '\t', убираем концевые '~', заменяем '~'->'\t'
dbview -d~ -b -t ALTNAMES.DBF \
  | sed -e :a -e '/[\r\t]$/N; s/[\r\t]\n//g; ta' \
  | recode CP866..CP1251 \
  | sed -e 's/\t//g; s/~\r//g; s/~,/,/g; s/~/\t/g' >>.sql

По итогу мы получаем большой-большой SQL-файл примерно такого вида:

CREATE TEMPORARY TABLE "STREET.DBF"(
  "NAME"
    varchar,
  "SOCR"
    varchar,
  "CODE"
    varchar,
  "INDEX"
    varchar,
  "GNINMB"
    varchar,
  "UNO"
    varchar,
  "OCATD"
    varchar
);
COPY "STREET.DBF"(
  "NAME",
  "SOCR",
  "CODE",
  "INDEX",
  "GNINMB",
  "UNO",
  "OCATD"
) FROM stdin;
Абадзехская     ул      01000001000000100       385013  0105            79401000000
Абрикосовая     ул      01000001000000200       385013  0105            79401000000
Авиационный     пер     01000001000000300       385006  0105            79401000000
Автодорога 7    ул      01000001000000400       385019  0105            79401000000
...

Импорт данных

Поскольку сами данные в таблицах архива представлены в DOS-кодировке, но это как-то совсем уж немодно, развернем нашу базу в WIN1251 - она тоже однобайтовая, поэтому все поиски данных не станут тяжелее:

CREATE DATABASE kladr
  WITH ENCODING='WIN1251'
    OWNER=postgres
    TEMPLATE=template0
    LC_COLLATE='ru_RU.CP1251'
    LC_CTYPE='ru_RU.CP1251'
    CONNECTION LIMIT=-1;
-- лог проверок обновлений
CREATE TABLE kladr_chk(
  id
    serial
      PRIMARY KEY,
  ts
    timestamp
      DEFAULT now(),
  hostname
    varchar
);

-- лог проведенных обновлений
CREATE TABLE kladr_upd(
  id
    serial
      PRIMARY KEY,
  ts
    timestamp
      DEFAULT now(),
  lm
    varchar,
  hostname
    varchar
);

Защита от параллельной загрузки

Поскольку импорт данных из ГНИВЦ у нас может проходить длительное время и инициироваться с разных источников (мало ли, один из инстансов прикажет долго жить), неплохо бы позаботиться о защите данных, чтобы они не пострадали - дописываем все это в начале .sql:

-- защита от автоотключения по таймауту
SET statement_timeout = 0;
-- включаем WIN-кодировку
SET client_encoding = 'WIN1251';
-- включаем application_name для мониторинга активного процесса
SET application_name = 'kladr : import [`hostname`]';
-- включаем "последовательные" транзакции
BEGIN TRANSACTION ISOLATION LEVEL SERIALIZABLE;
  -- блокируем эксклюзивно upd-таблицу для "замораживания" параллельных операций
  LOCK TABLE kladr_upd IN EXCLUSIVE MODE NOWAIT;
  -- сбрасываем информацию в таблицу протокола проверок
  INSERT INTO kladr_chk(hostname) VALUES('`hostname`');
  -- сравниваем Last-Modified загруженного файла и состояния в базе
  SELECT lm FROM kladr_upd ORDER BY id DESC LIMIT 1;
  -- если lm не совпадает
    -- для каждого DBF
    CREATE TEMPORARY TABLE ...;
    COPY ... FROM stdin;
    ...
    \.
    --
    -- обновляем "версию" нашей базы
    INSERT INTO kladr_upd(lm, hostname) VALUES('$lm', '`hostname`');
COMMIT;

В результате такого подхода мы всегда знаем, кто и когда проверял наличие обновлений, и кому это удалось.

Соберем все вместе в единый скрипт:

Много bash-кода
#!/bin/sh

. `dirname "$0"`/../app.conf

echo "`date '+%F %T'` ==== Connecting to DB : $pghost:$pgport:$pgbase:$pguser"
# тестирование подключения к БД
psql -t -c 'SELECT 1' -h $pghost -p $pgport -U $pguser -w $pgbase 1>/dev/null 2>/dev/null
rv="$?"
if [ "$rv" != "0" ]; then
  echo "$pghost:$pgport:$pgbase:$pguser:$pgpass" >>~/.pgpass
  chmod 0600 ~/.pgpass
  psql -t -c 'SELECT 1' -h $pghost -p $pgport -U $pguser -w $pgbase 1>/dev/null 2>/dev/null
  rv="$?"
fi

if [ "$rv" != "0" ]; then
  echo "DB not connected : $pghost:$pgport:$pgbase:$pguser"
  exit 1
fi

# инициализация каталога _dbf
#_dbf=`mktemp -d`
mkdir ./dbf && chmod 777 ./dbf
_dbf=`readlink -f ./dbf`
rm -rf ${_dbf} 2>/dev/null
mkdir ${_dbf} 2>/dev/null
touch ${_dbf}/.sql
dir=`dirname "$0"`
dir=`readlink -f $dir`

## импорт базы КЛАДР'а в _dbf/.sql
# защита от автоотключения по таймауту
echo "SET statement_timeout = 0;" >>${_dbf}/.sql
# включаем WIN-кодировку
echo "SET client_encoding = 'WIN1251';" >>${_dbf}/.sql
# включаем application_name для мониторинга активного процесса
echo "SET application_name = 'kladr : import [`hostname`]';" >>${_dbf}/.sql
# включаем "последовательные" транзакции
echo "BEGIN TRANSACTION ISOLATION LEVEL SERIALIZABLE;" >>${_dbf}/.sql
# блокируем эксклюзивно upd-таблицу для "замораживания" параллельных операций
echo "LOCK TABLE kladr_upd IN EXCLUSIVE MODE NOWAIT;" >>${_dbf}/.sql
# сбрасываем информацию в таблицу протокола проверок
echo "INSERT INTO kladr_chk(hostname) VALUES('`hostname`');" >>${_dbf}/.sql

# инициализация временного каталога импорта
#tmp=`mktemp -d`
mkdir ./tmp && chmod 777 ./tmp
tmp=`readlink -f ./tmp`
cd $tmp

echo "`date '+%F %T'` ==== Downloading : $source"
# загрузка базы КЛАДР'а с ограничением по скорости или без
wget -S $source --limit-rate=8k 2>.hdr
# wget -S $source 2>.hdr

echo "`date '+%F %T'` ==== Comparing 'Last-Modified'"
rc=`cat .hdr | egrep 'HTTP/[0-9]\.[0-9] [0-9]{3}' | sed -e 's/^[ ]*HTTP\/[0-9]\.[0-9][ ]*\([0-9]*\).*$/\1/i' | egrep -v '301' | head -1`
lm=`cat .hdr | egrep 'Last-Modified' | sed -e 's/^[ ]*Last-Modified:[ ]*//i' | head -1`
echo "  -- HTTP code : $rc"
echo "  -- HTTP 'Last-Modified' : $lm"
pglm=`psql -h $pghost -p $pgport -U $pguser -w -t -c 'SELECT lm FROM kladr_upd ORDER BY id DESC LIMIT 1' $pgbase | sed -e 's/^[ ]*//i'`
echo "  -- PGDB 'Last-Modified' : $pglm"

if [ "$rc" = "200" ] && [ "$lm" != "" ] && [ "$lm" != "$pglm" ]; then
  # распаковка базы
  echo "`date '+%F %T'` ==== Unpacking 7z"
  p7zip -d Base.7z 1>/dev/null 2>/dev/null
  cp $tmp/* ${_dbf}
  cd $dir

  echo "`date '+%F %T'` ==== Processing DBF"
  # обработка всех .DBF
  for dbf in `find ${_dbf} -maxdepth 1 -iname '*.DBF'`; do
    dbfn=`basename $dbf | tr '[:lower:]' '[:upper:]'`
    # преобразование заголовков
    echo "  -- DBF : $dbfn"
    echo "    -- header"
    # получаем структуру полей DBF | DOS2WIN | берем только описания полей (skip 2 строки) | оставляем только их имена
    fld=`dbview -b -t -e -o -r $dbf | recode CP866..CP1251 | tail -n+2 | xargs -l | egrep -io "^[a-z0-9_]+"`
    echo "CREATE TEMPORARY TABLE \"$dbfn\"(" >>${_dbf}/.sql
    fl="0"
    for i in ${fld}; do
      [ "$fl" = "1" ] && echo ',' >>${_dbf}/.sql
      echo -n "  \"$i\"\n    varchar" >>${_dbf}/.sql
      fl="1"
    done
    echo ");" >>${_dbf}/.sql
    # преобразование данных
    echo "    -- data"
    echo "COPY \"$dbfn\"(" >>${_dbf}/.sql
    fl="0"
    for i in ${fld}; do
      [ "$fl" = "1" ] && echo ',' >>${_dbf}/.sql
      echo -n "  \"$i\"" >>${_dbf}/.sql
      fl="1"
    done
    echo ") FROM stdin;" >>${_dbf}/.sql
    # получаем данные DBF, разделенные '~' | склеиваем "висящие" строки ([\t\r\n] в теле поля данных) | DOS2WIN | убираем все '\t' | убираем концевые ';' | заменяем ';'->'\t'
    dbview -d~ -b -t $dbf | sed -e :a -e '/[\r\t]$/N; s/[\r\t]\n//g; ta' | recode CP866..CP1251 | sed -e 's/\t//g; s/~\r//g; s/~,/,/g; s/~/\t/g' >>${_dbf}/.sql
    echo "\\." >>${_dbf}/.sql
  done

  # интеграция процедуры обновления базы - последовательное подключение всех sql-файлов импорта
  ls ${dir}/import/*.sql | xargs -l readlink -f | xargs -l -I{} cat {} >>${_dbf}/.sql
  # вставка метки обновления
  echo "INSERT INTO kladr_upd(lm, hostname) VALUES('$lm', '`hostname`');" >>${_dbf}/.sql
fi
echo "COMMIT;" >>${_dbf}/.sql

cd $dir
rm -rf $tmp

echo "`date '+%F %T'` ==== Processing SQL"
psql -h $pghost -p $pgport -U $pguser -w -f ${_dbf}/.sql $pgbase
rv="$?"

if [ "$rv" = "0" ]; then
  rm -rf ${_dbf}/ 2>/dev/null
fi
echo "`date '+%F %T'` ==== Exit : $rv"

exit "$rv"

Тут можно использовать для импорта и временные каталоги, создаваемые через mktemp, но хватит ли у вас места на tmp-разделе?..

Реквизиты доступа к базе и КЛАДР-источник в нашем случае будут храниться в app.conf:

pghost="kladr.tensor.ru"
pgport="5432"
pguser="postgres"
pgpass="postgres"
pgbase="kladr"
source="https://gnivc.ru/html/gnivcsoft/KLADR/Base.7z"

Поисковая база

А зачем нам вообще нужна какая-то другая структура? Чем нас не устраивают таблицы в оригинальном КЛАДР-архиве?

  • хранение адресных объектов (улиц и населенных пунктов) в разных структурах

  • хранение статуса/версии объекта (00, 51, ...) вместе с его кодом

  • невозможность наложить эффективные для поиска индексы

Напомню, что оригинальный код КЛАДР, согласно документации имеет вид СС РРР ГГГ ППП УУУУ АА, где:

  • СС - код субъекта РФ (региона)

  • РРР - код района

  • ГГГ - код города

  • ППП - код населенного пункта

  • УУУУ - код улицы (отсутствует у населенных пунктов)

  • АА - признак актуальности

То есть если вынести признак актуальности в отдельное поле, то у кодов многих объектов (например, городов) в конце окажутся нули, которые стоит безболезненно отсечь. И тогда коды будут иметь строго ограниченный набор длин в соответствии с "уровнем" объекта:

  • 2 - регионы и города федерального подчинения (77 - г. Москва, 76 - Ярославская обл.)

  • 5 - районы (76 001 - Ярославская обл., Ярославский р-н)

  • 8 - города (76 000 001 - Ярославская обл., г. Ярославль)

  • 11 - населенные пункты (76 001 000 008 - Ярославская обл., Ярославский р-н, д.Алешково)

  • 15 - улицы (76 001 000 008 0001 - Ярославская обл., Ярославский р-н, д.Алешково, ул.Малиновая)

При этом, как видим, коды "вышестоящих" объектов становятся префиксами кодов объектов вложенных.

Что же получилось по структуре?

-- основная таблица хранения адресных объектов
CREATE TABLE kladr(
  code
    varchar,
  status
    varchar,
  name
    varchar,
  abbr
    varchar,
  idx
    varchar,
  ifns
    varchar,
  ocato
    varchar,
  lvl
    smallint,
  norm
    varchar,
  PRIMARY KEY(code, status)
);
-- индекс по почтовому индексу
CREATE INDEX "kladr-idx"
  ON kladr(idx);
-- префиксный индекс по названию объекта
CREATE INDEX "kladr-nm"
  ON kladr(length(code), code varchar_pattern_ops, lower(name));

-- дома, привязанные к адресным объектам
CREATE TABLE kladr_house(
  code
    varchar,
  codeExt
    varchar,
  name
    varchar,
  idx
    varchar,
  ifns
    varchar,
  ocato
    varchar,
  PRIMARY KEY(code, codeExt)
);

-- замены и объединения объектов
CREATE TABLE kladr_repl(
  oldCode
    varchar,
  newCode
    varchar,
  PRIMARY KEY(oldCode, newCode)
);
CREATE INDEX "kladr-repl-new"
  ON kladr_repl(newCode);
CREATE INDEX "kladr-repl-old"
  ON kladr_repl(oldCode);

-- аббревиатуры административно-территориальных единиц ("ул", "пер", "г")
CREATE TABLE kladr_abbr(
  code
    varchar,
  lvl
    smallint,
  name
    varchar,
  PRIMARY KEY(code, lvl)
);
-- индекс по возможному "уровню" объекта
CREATE INDEX "kladr-abbr-lvl"
  ON kladr_abbr(lvl);

Это основные таблицы, данные в которые импортируются непосредственно из соответствующих DBF по модели наложения "диффов", описанной в статье "DBA: грамотно организовываем синхронизации и импорты":

kladr-abbr
--// КЛАДР : сокращения

-- создаем временную таблицу с импортируемыми данными КЛАДР'а
CREATE TEMPORARY TABLE _kladr_abbr(
  LIKE kladr_abbr INCLUDING INDEXES
);

-- заполняем преобразованными из DBF данными
INSERT INTO _kladr_abbr(code, lvl, name)
  SELECT
    "SCNAME",
    "KOD_T_ST"::smallint,
    "SOCRNAME"
  FROM
    "SOCRBASE.DBF";

-- удаляем отсутствующие
DELETE FROM
  kladr_abbr T
USING
  kladr_abbr X LEFT JOIN
  _kladr_abbr Y
    USING(code, lvl)
WHERE
  (T.code, T.lvl) = (X.code, X.lvl) AND
  Y IS NULL;

-- обновляем оставшиеся
UPDATE
  kladr_abbr kl
SET
  name = kli.name
FROM
  _kladr_abbr kli
WHERE
  (kl.code, kl.lvl) = (kli.code, kli.lvl) AND
  (
    kl.name
  ) IS DISTINCT FROM
  (
    kli.name
  );

-- очищаем совпадающие
DELETE FROM
  _kladr_abbr kli
USING
  kladr_abbr kl
WHERE
  (kli.code, kli.lvl) = (kl.code, kl.lvl);

-- вставляем оставшиеся
INSERT INTO kladr_abbr
  SELECT
    *
  FROM
    _kladr_abbr;
kladr
--// КЛАДР : от регионов до улиц

-- создаем временную таблицу с импортируемыми данными КЛАДР'а
CREATE TEMPORARY TABLE _kladr(
  LIKE kladr INCLUDING INDEXES
);

-- заполняем преобразованными из DBF данными
INSERT INTO _kladr(code, status, name, abbr, idx, ifns, ocato, lvl)
  SELECT DISTINCT ON(code, status)
    *
  FROM
    (
      SELECT
        regexp_replace(rpad(substr("CODE", 1, length("CODE") - 2), 15, '0'), '^(.{0,}?)((((0{3})?0{3})?0{3})?0{4})?$', E'\\1', 'ig') code,
        substr("CODE", length("CODE") - 1, 2) status,
        "NAME",
        "SOCR",
        nullif("INDEX", ''),
        nullif("GNINMB", ''),
        nullif("OCATD", ''),
        "STATUS"
      FROM
        (
          SELECT
            "CODE",
            "NAME",
            "SOCR",
            "INDEX",
            "GNINMB",
            "OCATD",
            "STATUS"::smallint
          FROM
            "KLADR.DBF"
        UNION ALL
          SELECT
            "CODE",
            "NAME",
            "SOCR",
            "INDEX",
            "GNINMB",
            "OCATD",
            NULL::smallint "STATUS"
          FROM
            "STREET.DBF"
        ) T
    ) T;

-- удаляем отсутствующие
DELETE FROM
  kladr T
USING
  kladr X LEFT JOIN
  _kladr Y
    USING(code, status)
WHERE
  (T.code, T.status) = (X.code, X.status) AND
  Y IS NULL;

-- обновляем оставшиеся
UPDATE
  kladr kl
SET
  (
    name,
    abbr,
    idx,
    ifns,
    ocato,
    lvl
  ) =
  (
    kli.name,
    kli.abbr,
    kli.idx,
    kli.ifns,
    kli.ocato,
    kli.lvl
  )
FROM
  _kladr kli
WHERE
  (kl.code, kl.status) = (kli.code, kli.status) AND
  (
    kl.name,
    kl.abbr,
    kl.idx,
    kl.ifns,
    kl.ocato,
    kl.lvl
  ) IS DISTINCT FROM
  (
    kli.name,
    kli.abbr,
    kli.idx,
    kli.ifns,
    kli.ocato,
    kli.lvl
  );

-- очищаем совпадающие
DELETE FROM
  _kladr kli
USING
  kladr kl
WHERE
  (kli.code, kli.status) = (kl.code, kl.status);

-- вставляем оставшиеся
INSERT INTO kladr
  SELECT
    *
  FROM
    _kladr;

-- обновляем поисковый кэш
DELETE FROM
  kladr_kw
WHERE
  (code, status) IN (
    SELECT
      (ro).code,
      (ro).status
    FROM
      kladr$log
    WHERE
      ro IS DISTINCT FROM NULL
  );

INSERT INTO
  kladr_kw(code, status, keyword)
SELECT DISTINCT
  code,
  status,
  kw
FROM
  (
    SELECT
      (rn).code,
      (rn).status,
      regexp_split_to_table(lower((rn).name), E'[^\\-a-zа-яё0-9]+', 'i') kw
    FROM
      kladr$log
    WHERE
      rn IS DISTINCT FROM NULL
  ) T
WHERE
  kw <> '';

DELETE FROM kladr$log;

Здесь регулярное выражение используется для отсечения "хвостовых" нулей по маске до необходимой нам длины. То есть нельзя просто так взять 76 000 010 000 и убрать все 4 последних ноля, поскольку 010 тут является значимым кодом города.

kladr-house
--// КЛАДР : дома

-- создаем временную таблицу с импортируемыми данными КЛАДР'а
CREATE TEMPORARY TABLE _kladr_house(
  LIKE kladr_house INCLUDING INDEXES
);

-- заполняем преобразованными из DBF данными
INSERT INTO _kladr_house(code, codeExt, name, idx, ifns, ocato)
  SELECT
    regexp_replace(substr("CODE", 1, 15), '^(.{0,}?)((((0{3})?0{3})?0{3})?0{4})?$', E'\\1', 'ig'),
    substr("CODE", 16, 4),
    "NAME",
    nullif("INDEX", ''),
    nullif("GNINMB", ''),
    nullif("OCATD", '')
  FROM
    "DOMA.DBF";

-- удаляем отсутствующие
DELETE FROM
  kladr_house T
USING
  kladr_house X LEFT JOIN
  _kladr_house Y
    USING(code, codeExt)
WHERE
  (T.code, T.codeExt) = (X.code, X.codeExt) AND
  Y IS NULL;

-- обновляем оставшиеся
UPDATE
  kladr_house kl
SET
  (
    name,
    idx,
    ifns,
    ocato
  ) =
  (
    kli.name,
    kli.idx,
    kli.ifns,
    kli.ocato
  )
FROM
  _kladr_house kli
WHERE
  (kl.code, kl.codeExt) = (kli.code, kli.codeExt) AND
  (
    kl.name,
    kl.idx,
    kl.ifns,
    kl.ocato
  ) IS DISTINCT FROM
  (
    kli.name,
    kli.idx,
    kli.ifns,
    kli.ocato
  );

-- очищаем совпадающие
DELETE FROM
  _kladr_house kli
USING
  kladr_house kl
WHERE
  (kli.code, kli.codeExt) = (kl.code, kl.codeExt);

-- вставляем оставшиеся
INSERT INTO kladr_house
  SELECT
    *
  FROM
    _kladr_house;

-- обновляем поисковый кэш
DELETE FROM
  kladr_hs
WHERE
  (code) IN (
      SELECT
        (ro).code
      FROM
        kladr_house$log
      WHERE
        ro IS DISTINCT FROM NULL
    UNION ALL
      SELECT
        (rn).code
      FROM
        kladr_house$log
      WHERE
        rn IS DISTINCT FROM NULL
  );

-- заполняем преобразованными данными
CREATE TEMPORARY TABLE _kladr_hs0 AS
  SELECT DISTINCT ON(code, house)
    code,
    idx,
    ifns,
    ocato,
    unnest(houses) house
  FROM
    (
      SELECT
        *,
        CASE
          WHEN _range IS NULL AND name ~ E'_' THEN ARRAY[regexp_replace(name, '_', '-')]
          WHEN _range IS NULL THEN ARRAY[name]
          WHEN _range IS NOT NULL THEN ARRAY(
            SELECT
              i::text
            FROM
              generate_series(_range[1]::integer + CASE WHEN _range[4] IS NOT NULL THEN (_range[1]::integer + _range[4]::integer) % 2 ELSE 0 END, _range[2]::integer, _range[3]::integer) i
          )
          ELSE NULL
        END houses
      FROM
        (
          SELECT
            code,
            idx,
            ifns,
            ocato,
            name,
            CASE
              WHEN name ~ E'^Н\\(\\d+-\\d+\\)$' THEN regexp_split_to_array(substr(name, 3, length(name) - 3), '-') || '2'::text || '1'::text
              WHEN name ~ E'^Ч\\(\\d+-\\d+\\)$' THEN regexp_split_to_array(substr(name, 3, length(name) - 3), '-') || '2'::text || '0'::text
              WHEN name = 'Н' THEN '{1,999,2}'::text[]
              WHEN name = 'Ч' THEN '{2,998,2}'::text[]
              WHEN name ~ E'^\\d+-\\d+$' THEN regexp_split_to_array(name, '-') || '1'::text
              ELSE NULL
            END _range
          FROM
            (
              SELECT
                code,
                idx,
                ifns,
                ocato,
                unnest(regexp_split_to_array(upper(name), ',')) "name"
              FROM
                kladr_house
              WHERE
                (code) IN (
                    SELECT
                      (ro).code
                    FROM
                      kladr_house$log
                    WHERE
                      ro IS DISTINCT FROM NULL
                  UNION ALL
                    SELECT
                      (rn).code
                    FROM
                      kladr_house$log
                    WHERE
                      rn IS DISTINCT FROM NULL
                )
            ) T
        ) T
    ) T
ORDER BY
  code, house, (_range IS NULL) DESC;

CREATE INDEX ON _kladr_hs0(code, house, idx DESC NULLS LAST);

CREATE TEMPORARY TABLE _kladr_hs1 AS
  SELECT DISTINCT ON (code, house)
    code,
    idx,
    ifns,
    ocato,
    house
  FROM
    _kladr_hs0
  ORDER BY
    code, house, idx DESC NULLS LAST;

CREATE INDEX ON _kladr_hs1(code, house);

CREATE TEMPORARY TABLE _kladr_hs2 AS
  SELECT
    code,
    coalesce(
      idx, 
      coalesce(
        (
          SELECT
            idx
          FROM
            _kladr_hs1
          WHERE
            (code, house) = (T.code, regexp_replace(T.house, E'^(\\d+)(\\D)?.*$', E'\\1', 'ig'))
          LIMIT 1
        ), 
        coalesce(
          (
            SELECT
              idx
            FROM
              kladr
            WHERE
              code IN (
                substr(T.code, 1, 15),
                substr(T.code, 1, 11),
                substr(T.code, 1,  8),
                substr(T.code, 1,  5),
                substr(T.code, 1,  2)
              ) AND
--              status = '00' AND
              idx IS NOT NULL
            ORDER BY
              length(code) DESC
            LIMIT 1
          ),
          ''
        )
      )
    ) idx,
    ifns,
    ocato,
    house
  FROM
    _kladr_hs1 T;

CREATE INDEX ON _kladr_hs2(code, idx, ifns, ocato, house);

INSERT INTO kladr_hs(code, idx, ifns, ocato, houses)
  SELECT
    code,
    idx,
    ifns,
    ocato,
    array_agg(house ORDER BY house) houses
  FROM
    _kladr_hs2
  GROUP BY
    1, 2, 3, 4;

DELETE FROM kladr_house$log;

Здесь регулярными выражениями мы приводим форматы исходной базы в списки конкретных номеров домов:

  • 5-11 -> {5, 6, 7, 8, 9, 10, 11}

  • Н(5-11) -> {5, 7, 9, 11}

  • Ч(6-10) -> {6, 8, 10}

  • Н -> {1, 3, ..., 999}

  • Ч -> {2, 4, ..., 998}

kladr-repl
--// КЛАДР : замены

-- создаем временную таблицу с импортируемыми данными КЛАДР'а
CREATE TEMPORARY TABLE _kladr_repl(
  LIKE kladr_repl INCLUDING INDEXES
);

-- заполняем преобразованными из DBF данными
INSERT INTO _kladr_repl(oldCode, newCode)
  SELECT DISTINCT
    co,
    cn
  FROM
    (
      SELECT
        regexp_replace(rpad(co, 15, '0'), '^(.{0,}?)((((0{3})?0{3})?0{3})?0{4})?$', E'\\1', 'ig') co,
        so,
        regexp_replace(rpad(cn, 15, '0'), '^(.{0,}?)((((0{3})?0{3})?0{3})?0{4})?$', E'\\1', 'ig') cn,
        sn
      FROM
        (
          SELECT
            *,
            substr("OLDCODE", 1, length("OLDCODE") - 2) co,
            substr("OLDCODE", length("OLDCODE") - 1, 2) so,
            substr("NEWCODE", 1, length("NEWCODE") - 2) cn,
            substr("NEWCODE", length("NEWCODE") - 1, 2) sn
          FROM
            "ALTNAMES.DBF"
        ) T
    ) T;

-- удаляем отсутствующие
DELETE FROM
  kladr_repl T
USING
  kladr_repl X LEFT JOIN
  _kladr_repl Y
    USING(oldCode, newCode)
WHERE
  (T.oldCode, T.newCode) = (X.oldCode, X.newCode) AND
  Y IS NULL;

-- очищаем совпадающие
DELETE FROM
  _kladr_repl kli
USING
  kladr_repl kl
WHERE
  (kli.oldCode, kli.newCode) = (kl.oldCode, kl.newCode);

-- вставляем оставшиеся
INSERT INTO kladr_repl
  SELECT
    *
  FROM
    _kladr_repl;

Тут мы использовали еще две "производные" таблицы, которые выступают в качестве поисковых индексов - kladr_kw и kladr_hs:

-- ключевые слова адресных объектов
CREATE TABLE kladr_kw(
  code
    varchar,
  status
    varchar,
  keyword
    varchar
);
CREATE INDEX "kladr-kw-cd"
  ON kladr_kw(code, status);
CREATE INDEX "kladr-kw-kwcd"
  ON kladr_kw(keyword varchar_pattern_ops, code varchar_pattern_ops);
CREATE INDEX "kladr-kw-cdkw"
  ON kladr_kw(code varchar_pattern_ops, keyword varchar_pattern_ops);

-- группы номеров домов с одинаковыми признаками
CREATE TABLE kladr_hs(
  code
    varchar,
  idx
    varchar,
  ifns
    varchar,
  ocato
    varchar,
  houses
    varchar[]
);
CREATE INDEX "kladr-hs-code"
  ON kladr_hs(code);
CREATE INDEX "kladr-hs-idx"
  ON kladr_hs(idx);

-- специальные "уровневые" индексы
DO $$
DECLARE
  ln integer[] = '{2,5,8,11,15}'::integer[];
BEGIN
  FOR i IN 1..array_length(ln, 1) LOOP
    EXECUTE '
CREATE INDEX "kladr-' || lpad(ln[i]::text, 2, '0') || '"
  ON kladr(code varchar_pattern_ops, status)
WHERE
  length(code) = ' || ln[i] || ';
CREATE INDEX "kladr-kw-' || lpad(ln[i]::text, 2, '0') || '"
  ON kladr_kw(keyword varchar_pattern_ops)
WHERE
  length(code) = ' || ln[i] || ';
CREATE INDEX "kladr-kw-kwcd-' || lpad(ln[i]::text, 2, '0') || '"
  ON kladr_kw(keyword varchar_pattern_ops, code varchar_pattern_ops)
WHERE
  length(code) = ' || ln[i] || ';
    ';
  END LOOP;
END$$ LANGUAGE plpgsql;

Зачем нам понадобятся такие дополнительные структуры, и как их использовать для организации эффективного подстрочного поиска, рассмотрим в следующей части статьи, а эту мы закончим функцией нормализации названия объекта:

normalize.sql
CREATE OR REPLACE FUNCTION normalize(obj kladr) RETURNS text AS $$
  SELECT
    CASE
      -- регионы
      WHEN length(code) = 2 THEN
        regexp_replace(
          regexp_replace(
            (
              CASE
                WHEN abbr = 'Респ' AND name !~* '(ая)$' THEN
                  abbrfull || ' ' || name
                WHEN abbr = 'край' THEN
                  name || ' ' || abbr
                WHEN abbr = 'обл' THEN
                  name || ' ' || abbr || '.'
                WHEN abbr = 'г' THEN
                  abbr || '.' || name
                WHEN abbr IN ('АО', 'Аобл') THEN
                  regexp_replace(name, ' Автономный округ', '', 'ig') || ' АО'
                ELSE
                  name || ' ' || abbrfull
              END
            ),
            E'/([а-я]+)/',
            E'- \\1',
            'ig'
          ),
          E'^(.{0,}?) - ([а-я]+)(.{0,}?)$',
          E'\\1\\3 (\\2)',
          'ig'
        )
      -- районы
      WHEN length(code) = 5 THEN
        CASE
          -- (14xxx) улус
          WHEN abbr = 'у' THEN
            initcap(name) || ' ' || lower(abbrfull)
          -- (24048)
          WHEN abbr = 'АО' THEN
            initcap(name) || ' ' || abbr
          ELSE
            initcap(name) || ' ' || abbr || (CASE WHEN abbr ~ '(-|/)' THEN '' ELSE '.' END)
        END
      -- города и населенные пункты
      WHEN length(code) IN (8, 11) THEN
        CASE
          WHEN abbr IN ('тер', 'г', 'с/п', 'дп') THEN
            regexp_replace(
              regexp_replace(
                CASE
                  WHEN name ~* E'^ДНП ' THEN
                    regexp_replace(name, E'^ДНП ', 'днп ', 'ig')
                  WHEN name ~* E'^ДНТ ' THEN
                    regexp_replace(name, E'^ДНТ ', 'днт ', 'ig')
                  WHEN name ~* E'^ДСК ' THEN
                    regexp_replace(name, E'^ДСК ', 'дск ', 'ig')
                  WHEN name ~* E'^ДСПК ' THEN
                    regexp_replace(name, E'^ДСПК ', 'дспк ', 'ig')
                  WHEN name ~* E'^ДПК ' THEN
                    regexp_replace(name, E'^ДПК ', 'дпк ', 'ig')
                  WHEN name ~* E'^НСТ ' THEN
                    regexp_replace(name, E'^НСТ ', 'нст ', 'ig')
                  WHEN name ~* E'^СДТ ' THEN
                    regexp_replace(name, E'^СДТ ', 'сдт ', 'ig')
                  WHEN name ~* E'^СК ' THEN
                    regexp_replace(name, E'^СК ', 'ск ', 'ig')
                  WHEN name ~* E'^С/К ' THEN
                    regexp_replace(name, E'^С/К ', 'ск ', 'ig')
                  WHEN name ~* E'^СКТ ' THEN
                    regexp_replace(name, E'^СКТ ', 'скт ', 'ig')
                  WHEN name ~* E'^СТ ' THEN
                    regexp_replace(name, E'^СТ ', 'ст ', 'ig')
                  WHEN name ~* E'^С/Т ' THEN
                    regexp_replace(name, E'^С/Т ', 'ст ', 'ig')
                  WHEN name ~* E'^СНТ ' THEN
                    regexp_replace(name, E'^СНТ ', 'снт ', 'ig')
                  WHEN name ~* E'^ПКС ' THEN
                    regexp_replace(name, E'^ПКС ', 'пкс ', 'ig')
                  WHEN name ~* E'^ПКСТ ' THEN
                    regexp_replace(name, E'^ПКСТ ', 'пкст ', 'ig')
                  WHEN name ~* E'^ПОС ' THEN
                    regexp_replace(name, E'^ПОС ', 'пос ', 'ig')
                  WHEN name ~* E'^массив ' THEN
                    regexp_replace(name, E'^массив ', 'массив ', 'ig')
                  WHEN name ~* E' СНТ$' THEN
                    regexp_replace(name, E'^(.*) СНТ$', E'снт \\1', 'ig')
                  WHEN name ~* E' СТ$' THEN
                    regexp_replace(name, E'^(.*) СТ$', E'ст \\1', 'ig')
                  WHEN name ~* E' ДНТ$' THEN
                    regexp_replace(name, E'^(.*) ДНТ$', E'днт \\1', 'ig')
                  WHEN name ~* E' ДПК$' THEN
                    regexp_replace(name, E'^(.*) ДПК$', E'дпк \\1', 'ig')
                  WHEN name ~* E'^уч-к СДТ ' THEN
                    regexp_replace(name, E'^уч-к СДТ ', 'сдт ', 'ig')
                  WHEN name ~* E'^уч-к ' THEN
                    regexp_replace(name, E'^уч-к ', '', 'ig')
                  WHEN name ~* E'^СПК ' THEN
                    regexp_replace(name, E'^СПК ', 'спк ', 'ig')
                  WHEN name ~* E'^сот ' THEN
                    regexp_replace(name, E'^сот ', 'сот ', 'ig')
                  WHEN name ~* E'^урочище ' THEN
                    regexp_replace(name, E'^урочище ', 'урочище ', 'ig')
                  WHEN name ~* E'^КС ' THEN
                    regexp_replace(name, E'^КС ', 'кс ', 'ig')
                  WHEN name ~* E'^МО пгт.' THEN
                    regexp_replace(name, E'^МО пгт.', 'пгт ', 'ig')
                  WHEN name ~* E'^МО город ' THEN
                    regexp_replace(name, E'^МО город ', 'г.', 'ig')
                  WHEN name ~* E'^МО городское поселение ' THEN
                    regexp_replace(name, E'^МО городское поселение ', 'г/п ', 'ig')
                  WHEN name ~* E'^МО поселок ' THEN
                    regexp_replace(name, E'^МО поселок ', 'пос.', 'ig')
                  WHEN name ~* E'^МО СП ' THEN
                    regexp_replace(name, E'^МО СП (.*)', E'с/п \\1', 'ig')
                  WHEN name ~* E'^МО ГП поселок ' THEN
                    regexp_replace(name, E'^МО ГП поселок (.*)', E'пгт \\1', 'ig')
                  WHEN name ~* E'^МО сельское поселение ' THEN
                    regexp_replace(name, E'^МО сельское поселение (.*)', E'с/п \\1', 'ig')
                  WHEN name ~* E'^МО селское поселение ' THEN
                    regexp_replace(name, E'^МО селское поселение (.*)', E'с/п \\1', 'ig')
                  WHEN name ~* E'^МО (.*) сельское поселение$' THEN
                    regexp_replace(name, E'^МО (.*) сельское поселение$', E'с/п \\1', 'ig')
                  WHEN name ~* ' сельское поселение$' THEN
                    regexp_replace(name, '^(.*) сельское поселение$', E'с/п \\1', 'ig')
                  WHEN name ~* ' селское поселение$' THEN
                    regexp_replace(name, '^(.*) селское поселение$', E'с/п \\1', 'ig')
                  WHEN name ~* ' городское поселение$' THEN
                    regexp_replace(name, '^(.*) городское поселение$', E'г/п \\1', 'ig')
                  WHEN name ~* ' городской округ$' THEN
                    regexp_replace(name, '^(.*) городской округ$', E'\\1 г/о', 'ig')
                  WHEN name ~* '^Городское поселение ' THEN
                    regexp_replace(name, '^Городское поселение ', E'г/п ', 'ig')
                  ELSE
                    CASE
                      WHEN abbr IN ('с/п', 'дп') THEN abbr || ' ' || name
                      ELSE abbr || '.' || name
                    END
                END,
                E'^(г/п|с/п) (.*(?:ое))$',
                E'\\2 \\1',
                'ig'
              ),
              E'(\\S)\\(',
              E'\\1 (',
              'ig'
            )
          WHEN abbr IN ('высел') THEN
            CASE
              WHEN name ~ E'(.*((?:ие)))$' THEN
                name || ' выселки'
              ELSE
                'выселок ' || name
            END
          WHEN abbr IN ('городок') THEN
            CASE
              WHEN name ~* E'(^городок|городок$)' THEN
                name
              WHEN name ~* E'военный' THEN
                regexp_replace(name, E'(военный)(\\s?)', E'\\1 городок\\2', 'ig')
              ELSE
                abbr || ' ' || name
            END
          WHEN abbr IN ('с/а', 'с/мо') THEN
            regexp_replace(abbr || ' ' || name, E'^(' || abbr || ') (.*((?:ая)|(?:ое)|(?:ий)|(?:ый)))$', E'\\2 \\1', 'ig')
          WHEN abbr IN ('п/о', 'пгт', 'массив') THEN abbr || ' ' || name
          WHEN abbr = 'с/с' THEN (
            CASE
              WHEN name ~* E'^Рабочий поселок' THEN regexp_replace(name, E'^Рабочий поселок', 'рабочий поселок', 'ig')
              WHEN name ~* E'^р\\.п\\.' THEN regexp_replace(name, E'^р\\.п\\.', 'рабочий поселок', 'ig')
              ELSE name || ' ' || abbr
            END
          )
          WHEN abbr = 'ж/д_будка' THEN
            'ж/д будка ' || regexp_replace(name, E'(будка|будка железной дороги|железнодорожная будка)\\s*', E'', 'ig')
          WHEN abbr = 'ж/д_казарм' THEN
            'ж/д казарма ' || regexp_replace(name, E'(казарма)\\s*', E'', 'ig')
          WHEN abbr = 'ж/д_оп' THEN
            'ж/д о/п ' || name
          WHEN abbr = 'ж/д_платф' THEN
            'ж/д платформа ' || regexp_replace(name, E'(платформа|платформа железной дороги|железнодорожная платформа)\\s*', E'', 'ig')
          WHEN abbr = 'ж/д_пост' THEN
            CASE
              WHEN name ~* E'^Блокпост' THEN
                name
              WHEN name ~* E'^(железнодорожная )?будка' THEN
                regexp_replace(name, E'^(железнодорожная )?будка', E'ж/д будка', 'ig')
              WHEN name ~* E'^разъезд ' THEN
                regexp_replace(name, E'^разъезд', E'ж/д разъезд', 'ig')
              ELSE
                'ж/д пост ' || name
            END
          WHEN abbr = 'ж/д_рзд' THEN
            'ж/д разъезд ' || name
          WHEN abbr = 'ж/д_ст' THEN
            'ж/д станция ' || name
          WHEN abbr = 'казарма' THEN
            CASE
              WHEN name ~* E'^Блокпост' THEN
                name
              WHEN name ~* E'^железнодорожная\\s+' THEN
                regexp_replace(name, E'^железнодорожная\\s+', E'ж/д казарма ', 'ig')
              ELSE
                'казарма ' || regexp_replace(name, E'^казарма\\s*', E'', 'ig')
            END
          WHEN abbr = 'мкр' THEN
            CASE
              WHEN name ~* E'^(.*-й|\\d+)$' THEN
                name || ' ' || lower(abbrfull)
              WHEN name ~* E'^(\\d+[а-я])$' THEN
                lower(abbrfull) || ' ' || name
              ELSE
                abbr || '.' || name
            END
          WHEN abbr IN ('нп','с','д','тер','п','жилзона','жилрайон','кв-л','х','ст-ца','рп','рзд','м','ст','у','снт','сл') THEN
            CASE
              WHEN name ~* E'^остров\\s+' THEN
                regexp_replace(name, E'^остров', E'остров', 'ig')
              WHEN name ~* E'^казарма\\s+' THEN
                regexp_replace(name, E'^казарма', E'казарма', 'ig')
              WHEN name ~* E'^кордон\\s+' THEN
                regexp_replace(name, E'^кордон', E'кордон', 'ig')
              WHEN name ~* E'^жилой район ' THEN
                regexp_replace(name, E'^жилой район ', 'жилой район ', 'ig')
              WHEN name ~* E' жилой район$' THEN
                regexp_replace(name, E'^(.*) жилой район$', E'жилой район \\1', 'ig')
              WHEN name ~* E'^жилая зона ' THEN
                regexp_replace(name, E'^жилая зона ', 'жилая зона ', 'ig')
              WHEN name ~* E' жилая зона$' THEN
                regexp_replace(name, E'^(.*) жилая зона$', E'жилая зона \\1', 'ig')
              WHEN name ~* E'^д/о ' THEN
                regexp_replace(name, E'^д/о ', 'д/о ', 'ig')
              WHEN name ~* E'^п/о ' THEN
                regexp_replace(name, E'^п/о ', 'п/о ', 'ig')
              WHEN name ~* E'^п/л ' THEN
                regexp_replace(name, E'^п/л ', 'п/л ', 'ig')
              WHEN name ~* E'^т/б ' THEN
                regexp_replace(name, E'^т/б ', 'т/б ', 'ig')
              WHEN name ~* E'^СНТ ' THEN
                regexp_replace(name, E'^СНТ ', 'снт ', 'ig')
              WHEN name ~* E' СНТ$' THEN
                regexp_replace(name, E'^(.*) СНТ$', E'снт \\1', 'ig')
              WHEN name ~* E'^СДТ ' THEN
                regexp_replace(name, E'^СДТ ', 'сдт ', 'ig')
              WHEN name ~* E' СДТ$' THEN
                regexp_replace(name, E'^(.*) СДТ$', E'сдт \\1', 'ig')
              WHEN name ~* E'^СТ ' THEN
                regexp_replace(name, E'^СТ ', 'ст ', 'ig')
              WHEN name ~* E' СТ$' THEN
                regexp_replace(name, E'^(.*) СТ$', E'ст \\1', 'ig')
              WHEN name ~* E'^ДНТ ' THEN
                regexp_replace(name, E'^ДНТ ', 'днт ', 'ig')
              WHEN name ~* E' ДНТ$' THEN
                regexp_replace(name, E'^(.*) ДНТ$', E'днт \\1', 'ig')
              WHEN name ~* E'^ДНП ' THEN
                regexp_replace(name, E'^ДНП ', 'днп ', 'ig')
              WHEN name ~* E' ДНП$' THEN
                regexp_replace(name, E'^(.*) ДНП$', E'днп \\1', 'ig')
              WHEN name ~* E'^будка( (ж/д|железной дороги))?' THEN
                regexp_replace(name, E'^будка( (ж/д|железной дороги))?', E'ж/д будка', 'ig')
              WHEN name ~* E'^железнодорожная будка' THEN
                regexp_replace(name, E'^железнодорожная будка', E'ж/д будка', 'ig')
              WHEN name ~* E'^платф.+' THEN
                regexp_replace(name, E'^(платф)\\s+', E'платформа ', 'ig')
              WHEN name ~* E'^поселок.+' THEN
                regexp_replace(name, E'^(поселок)\\s+', E'п.', 'ig')
              WHEN name ~* E'^(рзд|разъезд).+' THEN
                regexp_replace(name, E'^(рзд|разъезд)\\s+', E'разъезд ', 'ig')
              WHEN name ~* E'^участок.+' THEN
                regexp_replace(name, E'^(участок)\\s+', E'участок ', 'ig')
              WHEN name ~* E'^(урочище|ур-ще) ' THEN
                regexp_replace(name, E'^(урочище|ур-ще) ', 'урочище ', 'ig')
--              WHEN name ~* E'^ж/д' THEN
--                name
              ELSE
                CASE
                  WHEN abbr IN ('с', 'д', 'тер', 'п') THEN
                    abbr || '.' || name
                  WHEN abbr IN ('жилзона','жилрайон','кв-л','х','ст-ца','рп','рзд','м','ст','у') THEN
                    lower(abbrfull) || ' ' || name
                  WHEN abbr IN ('снт') THEN
                    abbr || ' ' || name
                  WHEN abbr IN ('сл') THEN
                    regexp_replace(lower(abbrfull) || ' ' || name, E'^(' || lower(abbrfull) || ') (.*((?:ая)))$', E'\\2 \\1', 'ig')
                  WHEN abbr IN ('нп') THEN
                    'н/п ' || name
                  ELSE
                    name
                END
            END
          ELSE
            CASE
              WHEN length(code) = 8 THEN
                name || ' ' || abbr
              ELSE
                abbr || ' ' || name
            END
        END
      -- улицы
      WHEN length(code) = 15 THEN
        CASE
          WHEN abbr IN ('высел') THEN
            CASE
              WHEN name ~ E'(.*((?:ие)))$' THEN
                name || ' выселки'
              ELSE
                'выселок ' || name
            END
          WHEN abbr IN ('гск') THEN
            CASE
              WHEN name ~* E'^ГСК ' THEN
                regexp_replace(name, E'^ГСК ', 'гск ', 'ig')
              WHEN name ~* E'^Г СК ' THEN
                regexp_replace(name, E'^Г СК ', 'гск ', 'ig')
              ELSE
                lower(abbr) || ' ' || name
            END
          WHEN abbr IN ('ж/д_будка') THEN
            CASE
              WHEN name ~* E'^((железнодорожная )?будка|ждб|ж-д будка)' THEN
                regexp_replace(name, E'^((железнодорожная )?будка|ждб|ж-д будка)', E'ж/д будка', 'ig')
              WHEN name ~* E'^железнодорожная\\s+' THEN
                regexp_replace(name, E'^железнодорожная', E'ж/д будка', 'ig')
              WHEN name ~* E'^железнодорожная$' THEN
                regexp_replace(name, E'^железнодорожная$', E'Железнодорожная будка', 'ig')
              ELSE
                'ж/д будка ' || name
            END
          WHEN abbr IN ('ж/д_казарм') THEN
            CASE
              WHEN name ~* E'^казарма\\s+' THEN
                regexp_replace(name, E'^казарма', E'ж/д казарма', 'ig')
              ELSE
                'ж/д казарма ' || name
            END
          WHEN abbr IN ('ж/д_оп') THEN
            'ж/д о/п ' || name
          WHEN abbr = 'ж/д_платф' THEN
            'ж/д платформа ' || regexp_replace(name, E'(платформа|платформа железной дороги|железнодорожная платформа)\\s*', E'', 'ig')
          WHEN abbr IN ('аал','аул') THEN
            lower(abbrfull) || ' ' || name
          WHEN abbr = 'ж/д_пост' THEN
            CASE
              WHEN name ~* E'^Блокпост' THEN
                name
              WHEN name ~* E'^(железнодорожная )?будка' THEN
                regexp_replace(name, E'^(железнодорожная )?будка', E'ж/д будка', 'ig')
              WHEN name ~* E'^разъезд ' THEN
                regexp_replace(name, E'^разъезд', E'ж/д разъезд', 'ig')
              ELSE
                'ж/д пост ' || name
            END
          WHEN abbr = 'ж/д_рзд' THEN
            'ж/д разъезд ' || name
          WHEN abbr = 'ж/д_ст' THEN
            'ж/д станция ' || name
          WHEN abbr = 'жт' THEN
            CASE
              WHEN name ~* '^животноводческая' THEN
                name
              WHEN name ~* E'^точка\\s' THEN
                lower(abbrfull) || ' ' || regexp_replace(name, E'^точка ', E'', 'ig')
               WHEN name ~* E'^ферма ' THEN
                 regexp_replace(name, E'^ферма ', 'ферма ', 'ig')
               WHEN name ~* E' ферма$' THEN
                 regexp_replace(name, E'^(.*) ферма$', E'ферма \\1', 'ig')
              ELSE
                regexp_replace(lower(abbrfull) || ' ' || name, E'^(' || lower(abbrfull) || ') (.*((?:ая)|(?:-я)))$', E'\\2 \\1', 'ig')
            END
          WHEN abbr = 'зона' THEN
            regexp_replace(
              CASE
                 WHEN name ~* E'^зона ' THEN
                   regexp_replace(name, E'^зона ', 'зона ', 'ig')
                 WHEN name ~* E' зона$' THEN
                   name
                 WHEN name ~* E'^промзона$' THEN
                   name
                 WHEN name ~* E' промзона$' THEN
                   name
                ELSE
                  regexp_replace(lower(abbrfull) || ' ' || name, E'^(' || lower(abbrfull) || ') (.*((?:ая)|(?:-я)))$', E'\\2 \\1', 'ig')
              END,
              E' промышленная зона$',
              E' промзона',
              'ig'
            )
          WHEN abbr = 'казарма' THEN
            CASE
              WHEN name ~* E'^Блокпост' THEN
                name
              WHEN name ~* E'^железнодорожная\\s+' THEN
                regexp_replace(name, E'^железнодорожная\\s+', E'ж/д казарма ', 'ig')
              ELSE
                'казарма ' || regexp_replace(name, E'^казарма\\s*', E'', 'ig')
            END
          WHEN abbr = 'км' THEN
            CASE
              WHEN name ~* E'(\\d|-й)$' THEN
                name || ' ' || abbr
              WHEN name ~* E'^([0-9\\.]+)( )?' THEN
                regexp_replace(name, E'^([0-9\\.]+)( )?(км( |$))?', E'\\1 км ', 'ig')
              ELSE
                name
            END
          WHEN abbr = 'мкр' THEN
            CASE
              WHEN name ~* E'^(.*-й|\\d+)$' THEN
                name || ' ' || lower(abbrfull)
              WHEN name ~* E'^(\\d+[а-я])$' THEN
                lower(abbrfull) || ' ' || name
              ELSE
                abbr || '.' || name
            END
          WHEN abbr = 'остров' THEN
            CASE
              WHEN name ~* E'(.*-й$|\\d+)$' THEN
                name || ' ' || lower(abbrfull)
              ELSE
                lower(abbrfull) || ' ' || name
            END
          WHEN abbr IN ('лпх','п/о','п/р','п/ст','промзона') THEN
            regexp_replace(lower(abbr) || ' ' || name, E'^(' || lower(abbr) || ') (.*((?:ая)|(?:ое)|(?:ой)|(?:ий)|(?:ый)|(?:-й)|(?:-я)))$', E'\\2 \\1', 'ig')
          WHEN abbr IN ('сад','б-р') THEN
            CASE
              WHEN name ~* E'^(.*-й|.*-я|\\d+)$' THEN
                name || ' ' || lower(abbrfull)
              ELSE
                regexp_replace(lower(abbr) || ' ' || name, E'^(' || lower(abbr) || ') (.*((?:ой)|(?:ий)|(?:ый)|(?:-й)))$', E'\\2 \\1', 'ig')
            END
          WHEN abbr IN ('пер','пл','туп') THEN
            CASE
              WHEN name ~* E'^(.*-й|.*-я|\\d+)$' THEN
                name || ' ' || lower(abbrfull)
              ELSE
                regexp_replace(lower(abbr) || '.' || name, E'^(' || lower(abbr) || E'\\.)(.*((?:ая)|(?:ое)|(?:ой)|(?:ий)|(?:ый)|(?:-й)|(?:-я)))$', E'\\2 \\1', 'ig')
            END
          WHEN abbr IN ('х','ш') THEN
              CASE
                WHEN name ~* E'^(.*-й|.*-я|\\d+)$' THEN
                  name || ' ' || lower(abbrfull)
                ELSE
                  regexp_replace(lower(abbr) || '.' || regexp_replace(name, E' ' || lower(abbrfull) || '$', E'', 'ig'), E'^(' || lower(abbr) || E'\\.)(.*((?:ая)|(?:ое)|(?:ой)|(?:ий)|(?:ый)|(?:-й)|(?:-я)))$', E'\\2 ' || lower(abbrfull), 'ig')
              END
          WHEN abbr IN ('пр-кт') THEN
            CASE
              WHEN name ~* E'^(.*-й|.*-я|\\d+)$' THEN
                name || ' ' || lower(abbrfull)
              ELSE
                regexp_replace(lower('пр-т') || ' ' || name, E'^(' || lower('пр-т') || E')\\s(.*((?:ой)|(?:ий)|(?:ый)|(?:-й)|(?:-я)))$', E'\\2 \\1', 'ig')
            END
          WHEN abbr IN ('проезд') THEN
            CASE
              WHEN name ~* E'^(.*-й|.*-я|\\d+)$' THEN
                name || ' ' || lower(abbrfull)
              ELSE
                regexp_replace(lower('пр-д') || ' ' || name, E'^(' || lower('пр-д') || E')\\s(.*((?:ой)|(?:ий)|(?:ый)|(?:-й)|(?:-я)))$', E'\\2 \\1', 'ig')
            END
          WHEN abbr IN ('стр') THEN
            CASE
              WHEN name ~* E'^\\d+-е$' THEN
                name || ' ' || lower(abbrfull)
              WHEN name ~* E'^\\d+(-[а-я])?$' THEN
                lower(abbrfull) || ' ' || name
              ELSE
                name
            END
          WHEN abbr IN ('аллея','берег','бугор','вал','въезд','городок','дор','заезд','канал','кв-л','кольцо','коса','линия','мост','парк','переезд','платф','пл-ка','полустанок','просек','просека','проселок','проток','проулок','рзд','ряды','сквер','спуск','тракт','уч-к','ферма') THEN
            CASE
              WHEN abbr = 'дор' AND name ~* '^(а/д|автодорога)' THEN
                regexp_replace(name, E'^(а/д|автодорога)\\s', E'а/д ', 'ig')
              WHEN name ~* E'^(.*-й|.*-я|\\d+)$' THEN
                name || ' ' || lower(abbrfull)
              ELSE
                regexp_replace(lower(abbrfull) || ' ' || name, E'^(' || lower(abbrfull) || ') (.*((?:ая)|(?:ое)|(?:ой)|(?:ий)|(?:ый)|(?:ые)|(?:-й)|(?:-я)))$', E'\\2 \\1', 'ig')
            END
          WHEN abbr IN ('ул') THEN
            replace(abbr, '_', ' ') || (CASE WHEN abbr ~ '(-|/)' THEN '' ELSE '.' END) || '' || name
          WHEN abbr IN ('нп','с','д','тер','п','жилзона','жилрайон','х','ст-ца','рп','рзд','м','ст','у','снт','сл') THEN
            CASE
              WHEN name ~* E'^остров\\s+' THEN
                regexp_replace(name, E'^остров', E'остров', 'ig')
              WHEN name ~* E'^казарма\\s+' THEN
                regexp_replace(name, E'^казарма', E'казарма', 'ig')
              WHEN name ~* E'^кордон\\s+' THEN
                regexp_replace(name, E'^кордон', E'кордон', 'ig')
              WHEN name ~* E'^жилой район ' THEN
                regexp_replace(name, E'^жилой район ', 'жилой район ', 'ig')
              WHEN name ~* E' жилой район$' THEN
                regexp_replace(name, E'^(.*) жилой район$', E'жилой район \\1', 'ig')
              WHEN name ~* E'^жилая зона ' THEN
                regexp_replace(name, E'^жилая зона ', 'жилая зона ', 'ig')
              WHEN name ~* E' жилая зона$' THEN
                regexp_replace(name, E'^(.*) жилая зона$', E'жилая зона \\1', 'ig')
              WHEN name ~* E'^д/о ' THEN
                regexp_replace(name, E'^д/о ', 'д/о ', 'ig')
              WHEN name ~* E'^п/о ' THEN
                regexp_replace(name, E'^п/о ', 'п/о ', 'ig')
              WHEN name ~* E'^п/л ' THEN
                regexp_replace(name, E'^п/л ', 'п/л ', 'ig')
              WHEN name ~* E'^т/б ' THEN
                regexp_replace(name, E'^т/б ', 'т/б ', 'ig')
              WHEN name ~* E'^СНТ ' THEN
                regexp_replace(name, E'^СНТ ', 'снт ', 'ig')
              WHEN name ~* E' СНТ$' THEN
                regexp_replace(name, E'^(.*) СНТ$', E'снт \\1', 'ig')
              WHEN name ~* E'^СДТ ' THEN
                regexp_replace(name, E'^СДТ ', 'сдт ', 'ig')
              WHEN name ~* E' СДТ$' THEN
                regexp_replace(name, E'^(.*) СДТ$', E'сдт \\1', 'ig')
              WHEN name ~* E'^СТ ' THEN
                regexp_replace(name, E'^СТ ', 'ст ', 'ig')
              WHEN name ~* E' СТ$' THEN
                regexp_replace(name, E'^(.*) СТ$', E'ст \\1', 'ig')
              WHEN name ~* E'^ДНТ ' THEN
                regexp_replace(name, E'^ДНТ ', 'днт ', 'ig')
              WHEN name ~* E' ДНТ$' THEN
                regexp_replace(name, E'^(.*) ДНТ$', E'днт \\1', 'ig')
              WHEN name ~* E'^ДНП ' THEN
                regexp_replace(name, E'^ДНП ', 'днп ', 'ig')
              WHEN name ~* E' ДНП$' THEN
                regexp_replace(name, E'^(.*) ДНП$', E'днп \\1', 'ig')
              WHEN name ~* E'^будка( (ж/д|железной дороги))?' THEN
                regexp_replace(name, E'^будка( (ж/д|железной дороги))?', E'ж/д будка', 'ig')
              WHEN name ~* E'^железнодорожная будка' THEN
                regexp_replace(name, E'^железнодорожная будка', E'ж/д будка', 'ig')
              WHEN name ~* E'^платф.+' THEN
                regexp_replace(name, E'^(платф)\\s+', E'платформа ', 'ig')
              WHEN name ~* E'^поселок.+' THEN
                regexp_replace(name, E'^(поселок)\\s+', E'п.', 'ig')
              WHEN name ~* E'^(рзд|разъезд).+' THEN
                regexp_replace(name, E'^(рзд|разъезд)\\s+', E'разъезд ', 'ig')
              WHEN name ~* E'^участок.+' THEN
                regexp_replace(name, E'^(участок)\\s+', E'участок ', 'ig')
              WHEN name ~* E'^(урочище|ур-ще) ' THEN
                regexp_replace(name, E'^(урочище|ур-ще) ', 'урочище ', 'ig')
--              WHEN name ~* E'^ж/д' THEN
--                name
              ELSE
                CASE
                  WHEN abbr IN ('с', 'д', 'тер', 'п') THEN
                    abbr || '.' || name
                  WHEN abbr IN ('жилзона','жилрайон','кв-л','х','ст-ца','рп','рзд','м','ст','у') THEN
                    lower(abbrfull) || ' ' || name
                  WHEN abbr IN ('снт') THEN
                    abbr || ' ' || name
                  WHEN abbr IN ('сл') THEN
                    regexp_replace(lower(abbrfull) || ' ' || name, E'^(' || lower(abbrfull) || ') (.*((?:ая)))$', E'\\2 \\1', 'ig')
                  WHEN abbr IN ('нп') THEN
                    'н/п ' || name
                  ELSE
                    name
                END
            END
          ELSE
            CASE
              WHEN name ~* '((?:ая)|(?:ое)|(?:ой)|(?:ий)|(?:ый)|(?:-й)|(?:-я))$' THEN
                name || ' ' || replace(abbr, '_', ' ') || (CASE WHEN abbr ~ '(-|/)' THEN '' ELSE '.' END)
              ELSE
                replace(abbr, '_', ' ') || (CASE WHEN abbr ~ '(-|/)' THEN '' ELSE '.' END) || name
            END
        END
    END
  FROM
    (
      SELECT
        kl.*,
        ab.name abbrfull
      FROM
        (
          SELECT ($1).*
        ) kl LEFT JOIN
        kladr_abbr ab ON
          ab.code = kl.abbr AND (
            (ab.lvl / 100, length(kl.code)) = (5, 15) OR
            (ab.lvl / 100, length(kl.code)) = (4, 11) OR
            (ab.lvl / 100, length(kl.code)) = (3,  8) OR
            (ab.lvl / 100, length(kl.code)) = (2,  5) OR
            (ab.lvl / 100, length(kl.code)) = (1,  2)
          )
    ) T
$$ LANGUAGE sql;

CREATE OR REPLACE FUNCTION kladr$norm() RETURNS trigger AS $$
BEGIN
  -- проверяем необходимость логгирования при обновлении записи
  NEW.norm = normalize(NEW);
  RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER norm
  BEFORE INSERT OR UPDATE OF code, name, abbr
  ON kladr
    FOR EACH ROW
      EXECUTE PROCEDURE kladr$norm();
Tags:
Hubs:
Total votes 11: ↑10 and ↓1+9
Comments10
Тензор
Разработчик системы СБИС
Website
183
Karma
36
Rating
Боровиков Кирилл @Kilor

Архитектура ИС: PostgreSQL, Node.js и highload

Send message
WebsiteWebsite

Articles

Information

Website
sbis.ru
Registered
Founded
Employees
1,001–5,000 employees
Location
Россия

Your account

Sections

Information

Services

Support
© 2006–2024, Habr