Рецепты PostgreSQL: загрузка Государственного Адресного Реестра

Для приготовления загрузки Государственного Адресного Реестра в PostgreSQL нам понадобится сам PostgreSQL, bash, sh, curl, wget, xml2csv, jq, или можно воспользоваться готовым образом.

ОСТОРОЖНО! Может потребоваться много дискового пространства! Терабайта должно хватить, может, даже пол-терабайта хватит.

Первым делом создадим файлы

# показываем, что выполняем
# выходим при ошибках
# отсутствие переменной окружения - это ошибка
set -eux 
SELF="$$" # запоминаем свой pid
touch deltaVersionId.txt # дата последней загруженной дельты
touch fullVersionId.txt # дата загрузки полного ГАРа
touch state.txt # текущее состояние

Далее, в зависимости от состояния будем делать различную работу

state="$(cat state.txt)" # читаем текущее состояние
while # пока
    case "$state" in # при состоянии
        ... # таком-то делаем то-то
    esac
    state="$(cat state.txt)" # читаем текущее состояние
    test "$state" != "done" # выполняем пока состояние не готово
do true; done 2>&1 | tee cron.log # пишем в лог и консоль

Итак, изначально не было никакого состояния (а также в случае состояния готово)

* ) # в остальных случаях
echo "done" >state.txt # запишем состояние готово
deltaVersionId="$(cat deltaVersionId.txt)" # прочитаем дату дельты
fullVersionId="$(cat fullVersionId.txt)" # прочитаем дату полного
if [ -z "$fullVersionId" ]; then # если ещё ни разу не загружали полный
    echo sql2pg >state.txt # то переходми к состоянию инициализации базы
elif [ -z "$deltaVersionId" ]; then # иначе если ни разу не загружали дельту, то
		# скачиваем информацию о последних доступных выгрузках
    wget --output-document=GetLastDownloadFileInfo.json https://fias.nalog.ru/WebServices/Public/GetLastDownloadFileInfo
 		# извлекаем из неё дату последней версии   
    lastVersionId="$(jq --raw-output .VersionId <GetLastDownloadFileInfo.json)"
		# и адрес последней выгрузки    
    URL="$(jq --raw-output .GarXMLFullURL <GetLastDownloadFileInfo.json)"
    # задаём имя скачиваемого файла
    ZIP="$lastVersionId.zip"
    # скачиваем выгрузку (полная выгрузка скачивается ОЧЕНЬ долго, часов 5)
    wget --continue --output-document="$ZIP" "$URL"
    echo "$lastVersionId" >deltaVersionId.txt # сохраняем дату последней выгрузки как дельты
    echo "$lastVersionId" >fullVersionId.txt # так и полной
    echo unzip >state.txt # переходим к состоянию распаковки
else # иначе (база проинициализирована и уже скачивали полную выгрузку или дельту)
		# скачиваем информацию обо всех доступных выгрузках
    wget --output-document=GetAllDownloadFileInfo.json https://fias.nalog.ru/WebServices/Public/GetAllDownloadFileInfo
		# скачиваем все выгрузки после даты последней дельты    
    jq --raw-output "sort_by(.VersionId) | .[] | select(.VersionId > $deltaVersionId) \
    		# параллельно в несколько потоков (по количеству доступных процессоров)
    		| .GarXMLDeltaURL" <GetAllDownloadFileInfo.json | xargs --verbose --no-run-if-empty --max-procs="$(nproc)" --replace=URL bash /usr/local/bin/wget.sh "URL"
    test $? -eq 0 || kill -SIGINT "$SELF" # выходим при ошибке
    # обновляем соответственно дату последней дельты
    jq --raw-output "sort_by(.VersionId) | .[] | select(.VersionId > $deltaVersionId) \
    		# последовательно (и при успехе переходим к состоянию распаковки)
    		| .GarXMLDeltaURL" <GetAllDownloadFileInfo.json | xargs --verbose --no-run-if-empty --replace=URL bash /usr/local/bin/version.sh "URL"
    test $? -eq 0 || kill -SIGINT "$SELF" # выходим при ошибке
fi
;;

файлы скачивания дельты и обновления версии ничего интересного из себя не представляют, кроме разве что в файле скачивания

# вычисляем размер скачиваемого файла
SIZE="$(curl -Is "$URL" | grep 'Content-Length' | grep -oP '\d+')"
test "$SIZE" -lt 1073741824 # если размер больше 1ГБ, то выходим

далее

"sql2pg" ) # в случае инициализации базы
# исполняем все sql-скрипты
find /usr/local/sql2pg -type f -name "*.sql" | sort -u \
		# последовательно
		| xargs --verbose --no-run-if-empty --replace=SQL cat "SQL" | psql --variable=ON_ERROR_STOP=1
test $? -eq 0 || kill -SIGINT "$SELF" # выходим при ошибке
echo region2pg >state.txt # переходим к состоянию инициализации для регионов
;;

sql-скрипты представляют из себя создание таблиц с комментариями и необходимых в дальнейшем индексов

далее

"region2pg" ) # в случае инициализации базы для регионов
# для каждого региона от 01 до 99 (для простоты взят весь диапазон чисел)
seq --format "%02.0f" 1 99 \
		# последовательно добавляем соответствующую схему в базу
		| xargs --verbose --no-run-if-empty --replace=REGION echo "CREATE SCHEMA IF NOT EXISTS \"REGION\";" | psql --variable=ON_ERROR_STOP=1
test $? -eq 0 || kill -SIGINT "$SELF" # выходим при ошибке
# для всех sh-скриптов
find /usr/local/region2pg -type f -name "*.sh" | sort -u | while read -r SH; do
    seq --format "%02.0f" 1 99 \ # для каждого региона
    		# последовательно выполняем скрипт, который генерирует sql-код
        # для создания таблиц и индексов
    		| xargs --verbose --no-run-if-empty --replace=REGION sh "$SH" "REGION" | psql --variable=ON_ERROR_STOP=1
    test $? -eq 0 || kill -SIGINT "$SELF" # выходим при ошибке
done
echo wget >fullVersionId.txt # зануляем полную версию
echo wget >state.txt # переходим к состоянию скачивания
# но т.к. отдельной обработки такого состояния нет, то оно
# обрабатывается дефолтным, описанным выше самым первым
;;

sh-скрипты представляют из себя генерацию sql-кода создания таблиц и необходимых в дальнейшем индексов для каждого заданного региона

далее

"unzip" ) # в случае распаковки
find . -type f -name "*.zip" | sort -u \ # все архивы
		# распаковываем параллельно 
    # (делая названия файлов маленькими буквами)
    # и при успехе удаляя архив
		| xargs --verbose --no-run-if-empty --max-procs="$(nproc)" --replace=ZIP bash /usr/local/bin/unzip.sh "ZIP"
test $? -eq 0 || kill -SIGINT "$SELF" # выходим при ошибке
echo xml2csv >state.txt # переходим к состоянию конвертации
;;

далее

"xml2csv" ) # в случае конвертации
# для всех sh-скриптов
find /usr/local/xml2csv -type f -name "*.sh" | sort -u | while read -r SH; do
    TABLE="$(basename -- "${SH%.*}")" # извлекаем название таблицы
    # для всех соответствующих xml-файлов
    find . -type f -name "as_${TABLE}_2*.xml" | sort -u \
    		# параллельно конвертируем их в csv-файлы
        # и при успехе удаляем xml-файл
    		| xargs --verbose --no-run-if-empty --max-procs="$(nproc)" --replace=XML bash "$SH" "XML"
    test $? -eq 0 || kill -SIGINT "$SELF" # выходим при ошибке
done
deltaVersionId="$(cat deltaVersionId.txt)" # читаем сохранённую дату дельты
fullVersionId="$(cat fullVersionId.txt)" # и полной выгрузки
# если они разные
if [ "$deltaVersionId" != "$fullVersionId" ]; then
    echo delta2pg >state.txt # то переходим к состоянию загрузки дельты
else # иначе (если они одинаковые)
    echo full2pg >state.txt # то переходим к состоянию загрузки полной
fi
;;

sh-скрипты представляют из себя команды для конвертации

сначала происходит

"full2pg" ) # в случае загрузки полной выгрузки
# для всех sh-скриптов
find /usr/local/full2pg -type f -name "*.sh" | sort -u | while read -r SH; do
    TABLE="$(basename -- "${SH%.*}")" # извлекаем название таблицы
    # для всех csv-файлов
    find . -type f -name "as_${TABLE}_2*.csv" | sort -u \
    		# параллельно загружаем их в базу
        # и при успехе удаляем csv-файл
    		| xargs --verbose --no-run-if-empty --max-procs="$(nproc)" --replace=CSV bash "$SH" "CSV"
    test $? -eq 0 || kill -SIGINT "$SELF" # выходим при ошибке
done
echo update >state.txt # переходим к состоянию обновления
;;

sh-скрипты представляют из себя генерацию sql-кода для загрузки соответствующей таблицы в базу и выполнение загрузки с помощью оператора COPY только актуальных данных

или же (во все следующие разы) происходит

"delta2pg" ) # в случае загрузки дельты
# для всех sh-скриптов
find /usr/local/delta2pg -type f -name "*.sh" | sort -u | while read -r SH; do
    TABLE="$(basename -- "${SH%.*}")" # извлекаем название таблицы
    # для всех скачанных дельт (если их было несколько с предыдущего раза)
    find -mindepth 1 -maxdepth 1 -type d | sort -u | while read -r DIR; do
    		# для всех csv-файлов
        find "$DIR" -type f -name "as_${TABLE}_2*.csv" | sort -u \
        		# параллельно загружаем их в базу
            # и успехе удаляем csv-файл
        		| xargs --verbose --no-run-if-empty --max-procs="$(nproc)" --replace=CSV bash "$SH" "CSV"
        test $? -eq 0 || kill -SIGINT "$SELF" # выходим при ошибке
        rmdir "$DIR"/* "$DIR" # удаляем директорию дельты
    done
done
echo update >state.txt # переходим к состоянию обновления
;;

sh-скрипты представляют из себя генерацию sql-кода для загрузки соответствующей таблицы в базу и выполнение загрузки с помощью оператора COPY во временную таблицу всех данных, а затем вставку из временной таблицы в основную с обновлением при кофликте, а в конце неактуальные данные удаляются из основной таблицы

таким образом, запуская главный скрипт в докере по крону каждый день получаем актуальное состояние Государственного Адресного Реестра в PostgreSQL

Про состояние обновления и дальнейшее использование Государственного Адресного Реестра будет в следующей статье.