File Coverage

blib/lib/FIAS/SQL.pm

Criterion	Covered	Total	%
statement	29	259	11.2
branch	0	108	0.0
condition	0	43	0.0
subroutine	10	39	25.6
pod	18	23	78.2
total	57	472	12.0

package FIAS::SQL;
$FIAS::SQL::VERSION = '0.08';
# ABSTRACT: Модуль для минимальной работы с данными из базы ФИАC https://fias.nalog.ru/FiasInfo.aspx

use strict;
use warnings;
use utf8;

use DBI;
use XBase;
use LWP::UserAgent;
use Carp qw ( confess );
use Encode qw ( decode );
use Readonly;
use v5.10;


# Описываем уровни адресных объектов в человекочитаемом виде.
# Уровень объекта может принимать только нижеперечисленные значения
# Условно выделены следующие уровни адресных объектов:
Readonly our $LEVELS => {
   # 1 – уровень региона
   region                    => '1',
   # 3 – уровень района
   district                  => '3',
   # 35 – уровень городских и сельских поселений
   settlement => '35',
   # 4 – уровень города
   town                      => '4',
   # 6 – уровень населенного пункта
   inhabitet_locality        => '6',
   # 65 – планировочная структура
   planning_structure        => '65',
   # 7 – уровень улицы
   street                    => '7',
   # 75 – земельный участок
   stead                     => '75',
   # 8 – здания, сооружения, объекта незавершенного строительства
   structure                 => '8',
   # 9 – уровень помещения в пределах здания, сооружения
   premises                  => '9'
};


sub new {
    my ( $class, %params ) = @_;

    # параметры подключения к базе
    my $db_connection = delete $params{db_connection};

    my $dsn = delete $db_connection->{dsn}
        or confess 'no dsn!';

    my $login    = delete $db_connection->{login};
    my $password = delete $db_connection->{password};

    my $self;

    $self->{dbh} =  DBI->connect( $dsn, $login, $password, $params{additional_connection_params} )
       or confess $DBI::errstr;

    bless $self, $class;

    return $self;
}

sub load_files {
    my ( $self, %params ) = @_;

    my $directory   = delete $params{directory};
    my $update_flag = delete $params{update};

    opendir( my $dh, $directory ) or confess $!;

    # Получаем все файлы из папки $directory
    my @files;
    while ( my $file = readdir( $dh ) ) {
         push @files, $file
            if $file =~ /\.[Dd][Bb][Ff]$/;
    }
    closedir( $dh );

    # хендлер базы
    my $dbh = $self->{dbh};

    # TODO надо будет в следующей версии продумать этот момент, когда обновление будем делать
    # перед заполнением базы хорошо бы её дропнуть
    #$dbh->do("DROP DATABASE IF EXISTS $basename");
    #$dbh->do("CREATE DATABASE $basename");
    #$dbh->do("USE $basename");

    # Хеш полей базы( используем при обновлении )
    my %sql_tables_names;
    # перебираем все файлы
    for my $dbf_file_name ( @files ) {

        # Получаем имя таблицы для SQL без номера региона и расширения
        # HOUSE89.DBF превращается в HOUSE
        ( my $sql_table_name = $dbf_file_name ) =~ s/\d*\.[Dd][Bb][Ff]$//;

        # Информация в данных таблицах не нужна
        next if $dbf_file_name =~ /^D/
            || $sql_table_name eq 'NORDOC'
            || $sql_table_name eq 'STEAD'
            || $sql_table_name eq 'NDOCTYPE';

        # Получаем доступ к таблице dbf
        my $table = XBase->new( "$directory/$dbf_file_name" ) or confess XBase->errstr;

        # индекс последней записи в dbf
        my $index_of_last_record  = $table->last_record;
        # массив типов записей в таблице dbf
        my @type  = $table->field_types;
        # массив имён полей в таблице dbf
        my @name  = $table->field_names;
        # массив длин полей в таблице dbf
        my @len   = $table->field_lengths;
        # массив количества знаков после запятой полей 'N' в таблице dbf
        my @dec   = $table->field_decimals;

        # сохраняем имена таблиц для последующей обработки при обновлении
        if ( $update_flag ) {
            $sql_tables_names{ lc $sql_table_name} = 1;
            $sql_table_name .= '_';
        }
        $dbh->do(_create_table( $sql_table_name, $table ) );

        # счётчик записей
        my $all_records = -1;
        # буфер записей
        my @sqldata;

        my $fields_for_insert = '(' . lc( join( ',', _get_table_fields( $sql_table_name, $table ) ) ) . ')';

        # выдёргиваем из файла по одной записи
        my $cursor = $table->prepare_select();
        while ( my $record = $cursor->fetch_hashref() ) {
            $all_records++;
            # Забираем только актуальные записи
            push @sqldata,  '('. _convert_data( $record, $table, $sql_table_name ) . ')'
                unless _check_record_actuality( $record );

            # Если есть актуальные записи и их накопилось 5000 или считали последнюю запись файла,
            # необходимо закоммитить значения в таблицу, если конечно есть что коммитить
            if ( scalar @sqldata && !( scalar @sqldata % 5000) || $all_records == $index_of_last_record ) {
                my $sql_table = lc $sql_table_name;
                $dbh->do( "REPLACE INTO $sql_table $fields_for_insert VALUES " . join(',', @sqldata ) )
                    or confess $DBI::errstr;
                undef @sqldata;
            }
        }
       print "Table $dbf_file_name copied\n";
       $table->close;
    }

    # после обработки файлов, при обновлении заменяем файлы
    if ( $update_flag ){
	   for my $table ( keys %sql_tables_names ) {
	       my $tmp_table = $table . '_';
	       $dbh->do( "DROP TABLE $table" ) or confess $DBI::errstr;
	       $dbh->do( "ALTER TABLE $tmp_table RENAME $table") or confess $DBI::errstr;
	   }
    }
    # Доработка таблиц для уменьшения и убыстрания
    # TODO перенести в функции создания таблиц
    $dbh->do('ALTER TABLE addrob DROP COLUMN nextid');
    $dbh->do('ALTER TABLE house  DROP COLUMN enddate');
    $dbh->do('ALTER TABLE room   DROP COLUMN nextid');
}

sub get_address_objects {
    my ( $self, %params ) = @_;

    my $parentguid = $params{parentguid};

    # Разрешаются только вышеперечисленные уровни
    confess 'level is incorrect'
        if $params{aolevel} && !$LEVELS->{ $params{aolevel} };

    # Забираем числовое значения уровня для выборкиж
    my $aolevel;
    $aolevel = $LEVELS->{ $params{aolevel} }
        if $params{aolevel};

    my $sqlquery = 'SELECT * FROM addrob';
    my ( @where, @binds );

    $sqlquery .= ' WHERE '
        if ( $aolevel || $parentguid);

    # Добавляем WHERE в зависимости от пришедших параметров
    if ( $aolevel ) {
        push @where, 'aolevel = ?';
        push @binds, $aolevel;
    }

    if ( $parentguid ) {
        # если приходит мультизначение
        if ( ref $parentguid eq 'ARRAY' ) {
            push @where, 'parentguid in (' . join (',', ('?') x  @$parentguid) . ')';
            push @binds, @$parentguid;
        }
        else {
            push @where, 'parentguid = ?';
            push @binds, $parentguid;
        }
    }

    $sqlquery .= join( ' AND ', @where );

    return ref $parentguid eq 'ARRAY'
           ? $self->{dbh}->selectall_arrayref( $sqlquery, undef, @binds )
           : $self->{dbh}->selectall_arrayref( $sqlquery, { Slice => {} }, @binds )
}

sub get_address_objects_guids_only {
    my ( $self, %params ) = @_;

    return [map{ $_->{aoguid} } @{ $self->get_address_objects( %params ) } ];
}


sub get_sublevels_for_objects {
    my ( $self, $parentguid ) = @_;

    # Должен быть id родительского объекта
    confess 'parentguid is empty!'
        unless $parentguid;

    my $sublevels = $self->{dbh}->selectcol_arrayref(
        'SELECT DISTINCT aolevel FROM addrob WHERE parentguid = ? ORDER BY aolevel',
        undef,  $parentguid);

    my %reversed_LEVELS = reverse %$LEVELS;

    return [ map { $reversed_LEVELS{ $_ } } @$sublevels ];
}
sub get_data_for_object_by_aoguid {
    my ( $self, $aoguid ) = @_;

    # Должен быть id  объекта
    confess 'aoguid is empty!'
        unless $aoguid;

    return $self->{dbh}->selectrow_hashref(
        'SELECT aoguid, aolevel, offname, shortname, parentguid, postalcode FROM addrob WHERE aoguid = ?',
        undef,  $aoguid );
}
sub get_data_for_object_by_aoguid_array{
    my ( $self, $aoguid ) = @_;

    # Должен быть id  объекта
    confess 'aoguid is empty!'
        unless $aoguid;

    return $self->{dbh}->selectrow_array("SELECT * FROM addrob where aoguid = '$aoguid'");
}
sub add_object_to_base {
    my ( $self, $table, @values ) = @_;

    # эскейпим данные
    $_ = "\'$_\'" for @values;

   # собираем данные региона в строку
   my $values_string = join(',', @values );

   $self->{dbh}->do( "INSERT INTO $table VALUES($values_string)")
       or confess $DBI::errstr; ;

}

sub get_houses_of_address_objects {
    my ( $self, $aoguid ) =  @_;

    my $sqlquery = 'SELECT * FROM house ';

    $sqlquery .= ' WHERE aoguid '
        if $aoguid;

    my @binds;
    if ( $aoguid ) {
        if ( ref $aoguid eq 'ARRAY' ){
            $sqlquery .= ' in (' . join (',', ('?') x @$aoguid) . ')';
            push @binds, @$aoguid;
        }
        else {

            $sqlquery .= ' = ?';
            push @binds, $aoguid;
        }
    }

    return ref $aoguid eq 'ARRAY'
           ? $self->{dbh}->selectall_arrayref( $sqlquery, undef, @binds )
           : $self->{dbh}->selectall_arrayref( $sqlquery, { Slice => {} }, @binds )

}

sub get_data_for_house_by_houseguid {
    my ( $self, $houseguid ) =  @_;

    # Должен быть id  дома
    confess 'houseguid is empty!'
        unless $houseguid;

    return $self->{dbh}->selectrow_hashref(
        'SELECT housenum, buildnum, strucnum, aoguid, houseguid, postalcode FROM house WHERE houseguid = ?',
        undef,  $houseguid );
}

sub get_houses_guids_only {
    my ( $self, %params ) = @_;

    return [map{ $_->{houseguid} } @{ $self->get_houses_of_address_objects() } ];
}

sub get_rooms_of_address_objects {
    my ( $self, $houseguid ) =  @_;

    # ID дома не может быть пустым
    confess 'houseguid is empty!' unless $houseguid;

    my $sqlquery = 'SELECT * FROM room WHERE houseguid ';

    my @binds;

    if ( ref $houseguid eq 'ARRAY' ){
        $sqlquery .= ' in (' . join (',', ('?') x @$houseguid) . ')';
        push @binds, @$houseguid;
    }
    else {

        $sqlquery .= ' = ?';
        push @binds, $houseguid;
    }

    return ref $houseguid eq 'ARRAY'
           ? $self->{dbh}->selectall_arrayref( $sqlquery, undef, @binds )
           : $self->{dbh}->selectall_arrayref( $sqlquery, { Slice => {} }, @binds );
}

sub get_data_for_room_by_roomguid {
    my ( $self, $roomguid ) =  @_;

    # Должен быть id  дома
    confess 'roomguid is empty!'
        unless $roomguid;

    return $self->{dbh}->selectrow_hashref(
        'SELECT flatnumber, flattype.shortname AS flattype, houseguid, roomguid, postalcode, roomnumber, roomtype.shortname AS roomtype FROM room
                    JOIN flattype ON flattype.fltypeid = room.flattype
                    JOIN roomtype ON roomtype.rmtypeid = room.roomtype
                    WHERE roomguid = ?',
        undef,  $roomguid );
}

sub get_data_for_table {
    my ( $self, $table_name ) =  @_;

    return $self->{dbh}->selectall_arrayref( "select * from $table_name" );
}

sub get_parent_record_chain_by_aoguid {
    my ( $self, $aoguid ) =  @_;

    # Должен быть id
    confess 'aoguid is empty!'
        unless $aoguid;

    my $levels_chain;
    my %reversed_LEVELS = reverse %$LEVELS;

    my $ao_data = $self->get_data_for_object_by_aoguid( $aoguid );
    confess 'wrong aoguid!' unless $ao_data;
    push @$levels_chain, { guid => $ao_data->{aoguid}, level => $reversed_LEVELS{ $ao_data->{aolevel} } };

    while( $reversed_LEVELS{ $ao_data->{aolevel} } ne 'region' ) {
        $ao_data = $self->get_data_for_object_by_aoguid( $ao_data->{parentguid} );
        push @$levels_chain, { guid => $ao_data->{aoguid}, level => $reversed_LEVELS{ $ao_data->{aolevel} } };
    }

    return $levels_chain;
}


sub get_parent_record_chain_by_houseguid {
    my ( $self, $houseguid ) =  @_;

    # Должен быть id
    confess 'houseguid is empty!'
        unless $houseguid;

    my $levels_chain;

    my $house_data = $self->get_data_for_house_by_houseguid( $houseguid );
    confess 'wrong houseguid' unless $house_data;
    push @$levels_chain, { guid => $house_data->{houseguid}, level => 'house' };
    push @$levels_chain, @{ $self->get_parent_record_chain_by_aoguid( $house_data->{aoguid} ) };

    return $levels_chain;
}


sub get_parent_record_chain_by_roomguid {
    my ( $self, $roomguid ) =  @_;

    # Должен быть id
    confess 'roomguid is empty!'
        unless $roomguid;

    my $levels_chain;

    my $room_data = $self->get_data_for_room_by_roomguid( $roomguid );
    confess 'wrong roomguid' unless $room_data;
    push @$levels_chain, { guid => $room_data->{roomguid}, level => 'room' };
    push @$levels_chain, @{ $self->get_parent_record_chain_by_houseguid( $room_data->{houseguid} ) };

    return $levels_chain;
}
sub set_database_version {
     my ( $self, $version ) = @_;

    # хендлер базы
    my $dbh = $self->{dbh};

    # создаём индексы
    $dbh->do( 'CREATE TABLE IF NOT EXISTS version ( database_version INT (8), restriction ENUM(\'\') PRIMARY KEY )' )
                    or confess $DBI::errstr;
    $dbh->do( "REPLACE INTO version (database_version) VALUES ( $version )" )
                    or confess $DBI::errstr;
}

sub get_database_version {
    my ( $self ) = @_;

    # хендлер базы
    my $dbh = $self->{dbh};

    return $dbh->selectrow_arrayref('SELECT database_version FROM version')->[0];
}
sub get_link_for_full_database_dbf{
    return 'http://fias.nalog.ru/Public/Downloads/Actual/fias_dbf.rar';
}
sub get_actual_base_version {
    my $ua = LWP::UserAgent->new;

    my $get = $ua->get( 'http://fias.nalog.ru/Public/Downloads/Actual/VerDate.txt', ':content_file'=>'version.txt' );

    open my $file, '<', "version.txt";
    my $firstline = <$file>;
    $firstline =~ s/\.//g;
    close $file;
    unlink 'version.txt';

    return $firstline;
}

sub clone_base_structure {
    my ( $self, $destination_object ) = @_;

    # Забираем список таблиц из ФИАС
    my @fias_tables = grep { $_ !~ /\_$/} @{ $self->show_tables() };
    # Создаём таблицы в тестовой базе
    for my $table_name ( @fias_tables ) {
        my $cr_statement = $self->{dbh}->selectcol_arrayref( "SHOW CREATE TABLE $table_name", { Columns => [2] } )->[0];
        $destination_object->{dbh}->do( $cr_statement );
    }
}

sub show_tables {
    my ( $self ) = @_;

    return $self->{dbh}->selectcol_arrayref("show tables");
}

sub _check_record_actuality {
    my ( $record ) = @_;

    # если есть поля свидетельствующие о неактуальности, то возвращаем 1.
    return 1
        if exists $record->{ACTSTATUS} && !$record->{ACTSTATUS}
        || exists $record->{LIVESTATUS} && !$record->{LIVESTATUS}
        || $record->{NEXTID}
        || exists $record->{ENDDATE} && $record->{ENDDATE} ne '20790606';

    return 0;
}

sub _get_table_fields {
    my ( $sql_table_name, $dbf_table ) = @_;

    # Изначально FIAS содержит много подробной, но не всегда необходимой информации
    # для того, чтобы не перегружать базу, записываем только необходимые поля, которые перечислены в этом хеше
    state $table_config = {
        ADDROB => {
            fields => [ qw/AOGUID AOLEVEL OFFNAME SHORTNAME PARENTGUID POSTALCODE NEXTID/ ],
            #where  => 'WHERE ACTSTATUS = 1 AND LIVESTATUS = 1 AND CURRSTATUS = 0'
        },
        HOUSE  =>{
            fields   => [ qw/HOUSENUM BUILDNUM STRUCNUM AOGUID HOUSEGUID POSTALCODE STATSTATUS STRSTATUS ENDDATE/ ]
        },
        ROOM   => {
            fields => [ qw/FLATNUMBER FLATTYPE HOUSEGUID ROOMGUID POSTALCODE ROOMNUMBER ROOMTYPE NEXTID/ ],
            #where => 'WHERE LIVESTATUS = 1',
        },
    };

    # При апдейте приходят подчёркивания, это лишннее
    $sql_table_name =~ s/\_$//a;

    # Возвращаем только необходимые поля
    return @{ $table_config->{ $sql_table_name }{fields} }
        if $table_config->{ $sql_table_name };

    # Возвращаем все поля
    return $dbf_table->field_names;
}

sub _create_table {
    my ( $sql_table_name, $dbf_table ) = @_;

    my @sqlcommand;

    # Получаем имена полей в таблице
    my @field_names = _get_table_fields( $sql_table_name, $dbf_table );

    # Выясняем для каждого поля дополнительные параметры.
    for my $name ( @field_names ) {
        # Длина поля
        my $len  = $dbf_table->field_length( $name );
        # Количество знаков после запятой для полей 'NUMERIC'
        my $dec  = $dbf_table->field_decimal( $name );
        # Тип поля
        my $type = $dbf_table->field_type( $name );
        # Переводим имя поля в нижний регистр
        $name = lc $name;

        # DBF 'C' переходит в 'CHAR'
        if ( $type eq 'C' ) {
            push @sqlcommand, "\`$name\` CHAR($len)";
        }
        # DBF 'D' переходит в 'DATE'
        elsif ( $type eq 'D' ) {
            push @sqlcommand, "\`$name\` DATE";
        }
        # DBF 'N' переходит в 'NUMERIC'
        elsif ( $type eq 'N' ) {
            push @sqlcommand, "\`$name\` NUMERIC($len, $dec)";
        }
        # Пришло что-то неизведанное!
        else { confess "unknown $type type" }
    }

    # имя таблицы тоже переводим в нижний регистр
    $sql_table_name = lc $sql_table_name;

    my $create_string = "CREATE TABLE IF NOT EXISTS $sql_table_name (" . join( ', ', @sqlcommand );

    if ( $sql_table_name =~ /^house\_?$/ ) {
	    $create_string .= ", PRIMARY KEY \`houseguid\` (\`houseguid\`), KEY \`aoguid\`(\`aoguid\`)";
    }
    elsif (  $sql_table_name =~ /^room\_?$/ ) {
	    $create_string .= ", PRIMARY KEY \`roomguid\` (\`roomguid\`), KEY \`houseguid\`(\`houseguid\`)";
    }
    elsif (  $sql_table_name =~ /^addrob\_?$/ ) {
	    $create_string .= ", PRIMARY KEY \`aoguid\` (\`aoguid\`), KEY \`aolevel\`(\`aolevel\`), KEY \`aoguid\`(\`aoguid\`)";
    }
    $create_string .=  ') MAX_ROWS=1000000000';
    return $create_string;
}

sub _convert_data {
    my ( $record, $dbf_table, $sql_table_name ) = @_;

    my @sqlcommand;

    # Получаем имена полей в таблице
    my @field_names = _get_table_fields( $sql_table_name, $dbf_table );

    # Выясняем для каждого поля дополнительные параметры.
    for my $name ( @field_names ) {
        # Тип поля
        my $type = $dbf_table->field_type( $name );
        my $cell;
        # Экранируем текст
        if ( $type eq 'C' ) {
            $cell = "'" . _get_quoted_text( $record->{ $name } ) . "'" // '\N';
        }
        # Превращаем дату DBF в дату SQL
        elsif ( $type eq 'D' ) {
            $cell =
              ( $record->{ $name } )
              ? _get_formatted_date( $record->{ $name } )
              : '\N';
        }
        # Получаем численное значение из DBF
        elsif ( $type eq 'N' ) {
            $cell = $record->{ $name } // 0;
        }

        push @sqlcommand, $cell;
   }
   # Склеиваем данные запятыми для инсёрта
   return join( ',', @sqlcommand );
}

sub _get_formatted_date {
    my ( $date )  = @_;

    $date = sprintf( '%08d', $date ) if ( length($date) < 8 );
    $date =~ s/(\d{4})(\d{2})(\d{2})/\'$1-$2-$3\'/;

    return $date;
}

sub _get_quoted_text {
    my ( $text ) = @_;

    # Экранируем
    $text =~ s/\\/\\\\/g;
    $text =~ s/\'/\\\'/g;
    # Декодируем
    $text = decode( 'cp866', $text );

    return $text;
}
1;

__END__