MT#18663 row bulk processing framework WIP #8
+delta calculation for lnp,features,featureoptions +skip prepaid subscribers +import username+password.txt -a full file with plaintext instead of hashes is needed to test.. +import order forced +import "batch.txt" -prepared for a single column containing the numbers to migrate Change-Id: Icfedda64b635fe872f6f24c005228773d9869b21changes/69/6969/3
Rene Krenn
10 years ago
parent
b4f244a9f1
commit
8eece8ecd1
@ -0,0 +1,267 @@
|
||||
package NGCP::BulkProcessor::Projects::Migration::IPGallery::Dao::import::Batch;
|
||||
use strict;
|
||||
|
||||
## no critic
|
||||
|
||||
#use File::Basename;
|
||||
#use Cwd;
|
||||
#use lib Cwd::abs_path(File::Basename::dirname(__FILE__) . '/../../../../');
|
||||
|
||||
use NGCP::BulkProcessor::Projects::Migration::IPGallery::ProjectConnectorPool qw(
|
||||
get_import_db
|
||||
|
||||
);
|
||||
#import_db_tableidentifier
|
||||
|
||||
use NGCP::BulkProcessor::SqlProcessor qw(
|
||||
registertableinfo
|
||||
create_targettable
|
||||
checktableinfo
|
||||
copy_row
|
||||
|
||||
insert_stmt
|
||||
);
|
||||
use NGCP::BulkProcessor::SqlRecord qw();
|
||||
|
||||
use NGCP::BulkProcessor::Projects::Migration::IPGallery::Dao::import::Subscriber qw();
|
||||
|
||||
require Exporter;
|
||||
our @ISA = qw(Exporter NGCP::BulkProcessor::SqlRecord);
|
||||
our @EXPORT_OK = qw(
|
||||
create_table
|
||||
gettablename
|
||||
check_table
|
||||
getinsertstatement
|
||||
getupsertstatement
|
||||
|
||||
findby_number
|
||||
countby_number
|
||||
|
||||
update_delta
|
||||
findby_delta
|
||||
countby_delta
|
||||
|
||||
$deleted_delta
|
||||
$updated_delta
|
||||
$added_delta
|
||||
|
||||
);
|
||||
|
||||
my $tablename = 'batch';
|
||||
my $get_db = \&get_import_db;
|
||||
#my $get_tablename = \&import_db_tableidentifier;
|
||||
|
||||
|
||||
my $expected_fieldnames = [
|
||||
'number',
|
||||
'delta',
|
||||
];
|
||||
|
||||
# table creation:
|
||||
my $primarykey_fieldnames = [ 'number' ];
|
||||
my $indexes = { $tablename . '_delta' => [ 'delta(7)' ]};
|
||||
#my $fixtable_statements = [];
|
||||
|
||||
our $deleted_delta = 'DELETED';
|
||||
our $updated_delta = 'UPDATED';
|
||||
our $added_delta = 'ADDED';
|
||||
|
||||
sub new {
|
||||
|
||||
my $class = shift;
|
||||
my $self = NGCP::BulkProcessor::SqlRecord->new($get_db,
|
||||
$tablename,
|
||||
$expected_fieldnames,$indexes);
|
||||
|
||||
bless($self,$class);
|
||||
|
||||
copy_row($self,shift,$expected_fieldnames);
|
||||
|
||||
return $self;
|
||||
|
||||
}
|
||||
|
||||
sub create_table {
|
||||
|
||||
my ($truncate) = @_;
|
||||
|
||||
my $db = &$get_db();
|
||||
|
||||
registertableinfo($db,$tablename,$expected_fieldnames,$indexes,$primarykey_fieldnames);
|
||||
return create_targettable($db,$tablename,$db,$tablename,$truncate,0,undef);
|
||||
|
||||
}
|
||||
|
||||
sub findby_delta {
|
||||
|
||||
my ($delta,$load_recursive) = @_;
|
||||
|
||||
check_table();
|
||||
my $db = &$get_db();
|
||||
my $table = $db->tableidentifier($tablename);
|
||||
|
||||
return [] unless defined $delta;
|
||||
|
||||
my $rows = $db->db_get_all_arrayref(
|
||||
'SELECT * FROM ' .
|
||||
$table .
|
||||
' WHERE ' .
|
||||
$db->columnidentifier('delta') . ' = ?'
|
||||
,$delta);
|
||||
|
||||
return buildrecords_fromrows($rows,$load_recursive);
|
||||
|
||||
}
|
||||
|
||||
sub findby_number {
|
||||
|
||||
my ($number,$load_recursive) = @_;
|
||||
|
||||
check_table();
|
||||
my $db = &$get_db();
|
||||
my $table = $db->tableidentifier($tablename);
|
||||
|
||||
my $rows = $db->db_get_all_arrayref(
|
||||
'SELECT * FROM ' .
|
||||
$table .
|
||||
' WHERE ' .
|
||||
$db->columnidentifier('number') . ' = ?'
|
||||
,$number);
|
||||
|
||||
return buildrecords_fromrows($rows,$load_recursive);
|
||||
|
||||
}
|
||||
|
||||
sub update_delta {
|
||||
|
||||
my ($number,$delta) = @_;
|
||||
|
||||
check_table();
|
||||
my $db = &$get_db();
|
||||
my $table = $db->tableidentifier($tablename);
|
||||
|
||||
my $stmt = 'UPDATE ' . $table . ' SET delta = ?';
|
||||
my @params = ();
|
||||
push(@params,$delta);
|
||||
if (defined $number) {
|
||||
$stmt .= ' WHERE ' . $db->columnidentifier('number') . ' = ?';
|
||||
push(@params,$number);
|
||||
}
|
||||
|
||||
return $db->db_do($stmt,@params);
|
||||
|
||||
}
|
||||
|
||||
sub countby_number {
|
||||
|
||||
my ($number) = @_;
|
||||
|
||||
check_table();
|
||||
my $db = &$get_db();
|
||||
my $table = $db->tableidentifier($tablename);
|
||||
|
||||
my $stmt = 'SELECT COUNT(*) FROM ' . $table;
|
||||
my @params = ();
|
||||
if (defined $number) {
|
||||
$stmt .= ' WHERE ' . $db->columnidentifier('number') . ' = ?';
|
||||
push(@params,$number);
|
||||
}
|
||||
|
||||
return $db->db_get_value($stmt,@params);
|
||||
|
||||
}
|
||||
|
||||
sub countby_delta {
|
||||
|
||||
my ($delta) = @_;
|
||||
|
||||
check_table();
|
||||
my $db = &$get_db();
|
||||
my $table = $db->tableidentifier($tablename);
|
||||
|
||||
my $stmt = 'SELECT COUNT(*) FROM ' . $table;
|
||||
my @params = ();
|
||||
if (defined $delta) {
|
||||
$stmt .= ' WHERE ' .
|
||||
$db->columnidentifier('delta') . ' = ?';
|
||||
push(@params,$delta);
|
||||
}
|
||||
|
||||
return $db->db_get_value($stmt,@params);
|
||||
|
||||
}
|
||||
|
||||
sub buildrecords_fromrows {
|
||||
|
||||
my ($rows,$load_recursive) = @_;
|
||||
|
||||
my @records = ();
|
||||
my $record;
|
||||
|
||||
if (defined $rows and ref $rows eq 'ARRAY') {
|
||||
foreach my $row (@$rows) {
|
||||
$record = __PACKAGE__->new($row);
|
||||
|
||||
# transformations go here ...
|
||||
if ($load_recursive) {
|
||||
$record->{_subscriber} = NGCP::BulkProcessor::Projects::Migration::IPGallery::Dao::import::Subscriber::findby_subscribernumber(
|
||||
$record->{number},
|
||||
$load_recursive,
|
||||
);
|
||||
}
|
||||
|
||||
push @records,$record;
|
||||
}
|
||||
}
|
||||
|
||||
return \@records;
|
||||
|
||||
}
|
||||
|
||||
sub getinsertstatement {
|
||||
|
||||
my ($insert_ignore) = @_;
|
||||
check_table();
|
||||
return insert_stmt($get_db,$tablename,$insert_ignore);
|
||||
|
||||
}
|
||||
|
||||
sub getupsertstatement {
|
||||
|
||||
my ($exists_delta,$new_delta) = @_;
|
||||
check_table();
|
||||
my $db = &$get_db();
|
||||
my $table = $db->tableidentifier($tablename);
|
||||
my $upsert_stmt = 'INSERT OR REPLACE INTO ' . $table . ' (' .
|
||||
join(', ',map { local $_ = $_; $_ = $db->columnidentifier($_); $_; } @$expected_fieldnames) . ')';
|
||||
my @values = ();
|
||||
foreach my $fieldname (@$expected_fieldnames) {
|
||||
if ('delta' eq $fieldname) {
|
||||
my $stmt = 'SELECT \'' . $exists_delta . '\' FROM ' . $table . ' WHERE ' .
|
||||
$db->columnidentifier('number') . ' = ?';
|
||||
push(@values,'COALESCE((' . $stmt . '), \'' . $new_delta . '\')');
|
||||
} else {
|
||||
push(@values,'?');
|
||||
}
|
||||
}
|
||||
$upsert_stmt .= ' VALUES (' . join(',',@values) . ')';
|
||||
return $upsert_stmt;
|
||||
|
||||
}
|
||||
|
||||
sub gettablename {
|
||||
|
||||
return $tablename;
|
||||
|
||||
}
|
||||
|
||||
sub check_table {
|
||||
|
||||
return checktableinfo($get_db,
|
||||
$tablename,
|
||||
$expected_fieldnames,
|
||||
$indexes);
|
||||
|
||||
}
|
||||
|
||||
1;
|
||||
@ -0,0 +1,262 @@
|
||||
package NGCP::BulkProcessor::Projects::Migration::IPGallery::Dao::import::UsernamePassword;
|
||||
use strict;
|
||||
|
||||
## no critic
|
||||
|
||||
#use File::Basename;
|
||||
#use Cwd;
|
||||
#use lib Cwd::abs_path(File::Basename::dirname(__FILE__) . '/../../../../');
|
||||
|
||||
use NGCP::BulkProcessor::Projects::Migration::IPGallery::ProjectConnectorPool qw(
|
||||
get_import_db
|
||||
|
||||
);
|
||||
#import_db_tableidentifier
|
||||
|
||||
use NGCP::BulkProcessor::SqlProcessor qw(
|
||||
registertableinfo
|
||||
create_targettable
|
||||
checktableinfo
|
||||
copy_row
|
||||
|
||||
insert_stmt
|
||||
);
|
||||
use NGCP::BulkProcessor::SqlRecord qw();
|
||||
|
||||
require Exporter;
|
||||
our @ISA = qw(Exporter NGCP::BulkProcessor::SqlRecord);
|
||||
our @EXPORT_OK = qw(
|
||||
create_table
|
||||
gettablename
|
||||
check_table
|
||||
getinsertstatement
|
||||
getupsertstatement
|
||||
|
||||
findby_fqdn
|
||||
countby_fqdn
|
||||
|
||||
update_delta
|
||||
findby_delta
|
||||
countby_delta
|
||||
|
||||
$deleted_delta
|
||||
$updated_delta
|
||||
$added_delta
|
||||
|
||||
);
|
||||
|
||||
my $tablename = 'user_password';
|
||||
my $get_db = \&get_import_db;
|
||||
#my $get_tablename = \&import_db_tableidentifier;
|
||||
|
||||
|
||||
my $expected_fieldnames = [
|
||||
'fqdn',
|
||||
'username',
|
||||
'password',
|
||||
'auth_level',
|
||||
'delta',
|
||||
];
|
||||
|
||||
# table creation:
|
||||
my $primarykey_fieldnames = [ 'fqdn' ];
|
||||
my $indexes = { $tablename . '_delta' => [ 'delta(7)' ]};
|
||||
#my $fixtable_statements = [];
|
||||
|
||||
our $deleted_delta = 'DELETED';
|
||||
our $updated_delta = 'UPDATED';
|
||||
our $added_delta = 'ADDED';
|
||||
|
||||
sub new {
|
||||
|
||||
my $class = shift;
|
||||
my $self = NGCP::BulkProcessor::SqlRecord->new($get_db,
|
||||
$tablename,
|
||||
$expected_fieldnames,$indexes);
|
||||
|
||||
bless($self,$class);
|
||||
|
||||
copy_row($self,shift,$expected_fieldnames);
|
||||
|
||||
return $self;
|
||||
|
||||
}
|
||||
|
||||
sub create_table {
|
||||
|
||||
my ($truncate) = @_;
|
||||
|
||||
my $db = &$get_db();
|
||||
|
||||
registertableinfo($db,$tablename,$expected_fieldnames,$indexes,$primarykey_fieldnames);
|
||||
return create_targettable($db,$tablename,$db,$tablename,$truncate,0,undef);
|
||||
|
||||
}
|
||||
|
||||
sub findby_delta {
|
||||
|
||||
my ($delta,$load_recursive) = @_;
|
||||
|
||||
check_table();
|
||||
my $db = &$get_db();
|
||||
my $table = $db->tableidentifier($tablename);
|
||||
|
||||
return [] unless defined $delta;
|
||||
|
||||
my $rows = $db->db_get_all_arrayref(
|
||||
'SELECT * FROM ' .
|
||||
$table .
|
||||
' WHERE ' .
|
||||
$db->columnidentifier('delta') . ' = ?'
|
||||
,$delta);
|
||||
|
||||
return buildrecords_fromrows($rows,$load_recursive);
|
||||
|
||||
}
|
||||
|
||||
sub findby_fqdn {
|
||||
|
||||
my ($fqdn,$load_recursive) = @_;
|
||||
|
||||
check_table();
|
||||
my $db = &$get_db();
|
||||
my $table = $db->tableidentifier($tablename);
|
||||
|
||||
my $rows = $db->db_get_all_arrayref(
|
||||
'SELECT * FROM ' .
|
||||
$table .
|
||||
' WHERE ' .
|
||||
$db->columnidentifier('fqdn') . ' = ?'
|
||||
,$fqdn);
|
||||
|
||||
return buildrecords_fromrows($rows,$load_recursive);
|
||||
|
||||
}
|
||||
|
||||
sub update_delta {
|
||||
|
||||
my ($fqdn,$delta) = @_;
|
||||
|
||||
check_table();
|
||||
my $db = &$get_db();
|
||||
my $table = $db->tableidentifier($tablename);
|
||||
|
||||
my $stmt = 'UPDATE ' . $table . ' SET delta = ?';
|
||||
my @params = ();
|
||||
push(@params,$delta);
|
||||
if (defined $fqdn) {
|
||||
$stmt .= ' WHERE ' . $db->columnidentifier('fqdn') . ' = ?';
|
||||
push(@params,$fqdn);
|
||||
}
|
||||
|
||||
return $db->db_do($stmt,@params);
|
||||
|
||||
}
|
||||
|
||||
sub countby_fqdn {
|
||||
|
||||
my ($fqdn) = @_;
|
||||
|
||||
check_table();
|
||||
my $db = &$get_db();
|
||||
my $table = $db->tableidentifier($tablename);
|
||||
|
||||
my $stmt = 'SELECT COUNT(*) FROM ' . $table;
|
||||
my @params = ();
|
||||
if (defined $fqdn) {
|
||||
$stmt .= ' WHERE ' . $db->columnidentifier('fqdn') . ' = ?';
|
||||
push(@params,$fqdn);
|
||||
}
|
||||
|
||||
return $db->db_get_value($stmt,@params);
|
||||
|
||||
}
|
||||
|
||||
sub countby_delta {
|
||||
|
||||
my ($delta) = @_;
|
||||
|
||||
check_table();
|
||||
my $db = &$get_db();
|
||||
my $table = $db->tableidentifier($tablename);
|
||||
|
||||
my $stmt = 'SELECT COUNT(*) FROM ' . $table;
|
||||
my @params = ();
|
||||
if (defined $delta) {
|
||||
$stmt .= ' WHERE ' .
|
||||
$db->columnidentifier('delta') . ' = ?';
|
||||
push(@params,$delta);
|
||||
}
|
||||
|
||||
return $db->db_get_value($stmt,@params);
|
||||
|
||||
}
|
||||
|
||||
sub buildrecords_fromrows {
|
||||
|
||||
my ($rows,$load_recursive) = @_;
|
||||
|
||||
my @records = ();
|
||||
my $record;
|
||||
|
||||
if (defined $rows and ref $rows eq 'ARRAY') {
|
||||
foreach my $row (@$rows) {
|
||||
$record = __PACKAGE__->new($row);
|
||||
|
||||
# transformations go here ...
|
||||
|
||||
push @records,$record;
|
||||
}
|
||||
}
|
||||
|
||||
return \@records;
|
||||
|
||||
}
|
||||
|
||||
sub getinsertstatement {
|
||||
|
||||
my ($insert_ignore) = @_;
|
||||
check_table();
|
||||
return insert_stmt($get_db,$tablename,$insert_ignore);
|
||||
|
||||
}
|
||||
|
||||
sub getupsertstatement {
|
||||
|
||||
my ($exists_delta,$new_delta) = @_;
|
||||
check_table();
|
||||
my $db = &$get_db();
|
||||
my $table = $db->tableidentifier($tablename);
|
||||
my $upsert_stmt = 'INSERT OR REPLACE INTO ' . $table . ' (' .
|
||||
join(', ',map { local $_ = $_; $_ = $db->columnidentifier($_); $_; } @$expected_fieldnames) . ')';
|
||||
my @values = ();
|
||||
foreach my $fieldname (@$expected_fieldnames) {
|
||||
if ('delta' eq $fieldname) {
|
||||
my $stmt = 'SELECT \'' . $exists_delta . '\' FROM ' . $table . ' WHERE ' .
|
||||
$db->columnidentifier('fqdn') . ' = ?';
|
||||
push(@values,'COALESCE((' . $stmt . '), \'' . $new_delta . '\')');
|
||||
} else {
|
||||
push(@values,'?');
|
||||
}
|
||||
}
|
||||
$upsert_stmt .= ' VALUES (' . join(',',@values) . ')';
|
||||
return $upsert_stmt;
|
||||
|
||||
}
|
||||
|
||||
sub gettablename {
|
||||
|
||||
return $tablename;
|
||||
|
||||
}
|
||||
|
||||
sub check_table {
|
||||
|
||||
return checktableinfo($get_db,
|
||||
$tablename,
|
||||
$expected_fieldnames,
|
||||
$indexes);
|
||||
|
||||
}
|
||||
|
||||
1;
|
||||
@ -0,0 +1,68 @@
|
||||
package NGCP::BulkProcessor::Projects::Migration::IPGallery::FileProcessors::BatchFile;
|
||||
use strict;
|
||||
|
||||
## no critic
|
||||
|
||||
#use File::Basename;
|
||||
#use Cwd;
|
||||
#use lib Cwd::abs_path(File::Basename::dirname(__FILE__) . '/../../../../');
|
||||
|
||||
use NGCP::BulkProcessor::Logging qw(
|
||||
getlogger
|
||||
);
|
||||
use NGCP::BulkProcessor::LogError qw(
|
||||
fileprocessingerror
|
||||
fileprocessingwarn
|
||||
);
|
||||
|
||||
use NGCP::BulkProcessor::FileProcessor;
|
||||
|
||||
require Exporter;
|
||||
our @ISA = qw(Exporter NGCP::BulkProcessor::FileProcessor);
|
||||
our @EXPORT_OK = qw();
|
||||
|
||||
my $lineseparator = '\\n';
|
||||
my $fieldseparator = " +";
|
||||
my $encoding = 'UTF-8';
|
||||
|
||||
my $buffersize = 100 * 1024;
|
||||
my $threadqueuelength = 10;
|
||||
my $default_numofthreads = 3;
|
||||
#my $multithreading = 0;
|
||||
my $blocksize = 500;
|
||||
|
||||
sub new {
|
||||
|
||||
my $class = shift;
|
||||
|
||||
my $self = NGCP::BulkProcessor::FileProcessor->new(@_);
|
||||
|
||||
$self->{numofthreads} = shift // $default_numofthreads;
|
||||
$self->{line_separator} = $lineseparator;
|
||||
$self->{field_separator} = $fieldseparator;
|
||||
$self->{encoding} = $encoding;
|
||||
$self->{buffersize} = $buffersize;
|
||||
$self->{threadqueuelength} = $threadqueuelength;
|
||||
#$self->{multithreading} = $multithreading;
|
||||
$self->{blocksize} = $blocksize;
|
||||
|
||||
bless($self,$class);
|
||||
|
||||
#restdebug($self,__PACKAGE__ . ' file processor created',getlogger(__PACKAGE__));
|
||||
|
||||
return $self;
|
||||
|
||||
}
|
||||
|
||||
sub extractfields {
|
||||
my ($context,$line_ref) = @_;
|
||||
my $separator = $context->{instance}->{field_separator};
|
||||
$$line_ref =~ s/^ +//;
|
||||
$$line_ref =~ s/ +$//;
|
||||
return undef if length($$line_ref) == 0;
|
||||
return undef if $$line_ref =~ /^#/;
|
||||
my @fields = split(/$separator/,$$line_ref,-1);
|
||||
return \@fields;
|
||||
}
|
||||
|
||||
1;
|
||||
@ -0,0 +1,68 @@
|
||||
package NGCP::BulkProcessor::Projects::Migration::IPGallery::FileProcessors::UserPasswordFile;
|
||||
use strict;
|
||||
|
||||
## no critic
|
||||
|
||||
#use File::Basename;
|
||||
#use Cwd;
|
||||
#use lib Cwd::abs_path(File::Basename::dirname(__FILE__) . '/../../../../');
|
||||
|
||||
use NGCP::BulkProcessor::Logging qw(
|
||||
getlogger
|
||||
);
|
||||
use NGCP::BulkProcessor::LogError qw(
|
||||
fileprocessingerror
|
||||
fileprocessingwarn
|
||||
);
|
||||
|
||||
use NGCP::BulkProcessor::FileProcessor;
|
||||
|
||||
require Exporter;
|
||||
our @ISA = qw(Exporter NGCP::BulkProcessor::FileProcessor);
|
||||
our @EXPORT_OK = qw();
|
||||
|
||||
my $lineseparator = '\\r\\n';
|
||||
my $fieldseparator = " +";
|
||||
my $encoding = 'UTF-8';
|
||||
|
||||
my $buffersize = 100 * 1024;
|
||||
my $threadqueuelength = 10;
|
||||
my $default_numofthreads = 3;
|
||||
#my $multithreading = 0;
|
||||
my $blocksize = 500;
|
||||
|
||||
sub new {
|
||||
|
||||
my $class = shift;
|
||||
|
||||
my $self = NGCP::BulkProcessor::FileProcessor->new(@_);
|
||||
|
||||
$self->{numofthreads} = shift // $default_numofthreads;
|
||||
$self->{line_separator} = $lineseparator;
|
||||
$self->{field_separator} = $fieldseparator;
|
||||
$self->{encoding} = $encoding;
|
||||
$self->{buffersize} = $buffersize;
|
||||
$self->{threadqueuelength} = $threadqueuelength;
|
||||
#$self->{multithreading} = $multithreading;
|
||||
$self->{blocksize} = $blocksize;
|
||||
|
||||
bless($self,$class);
|
||||
|
||||
#restdebug($self,__PACKAGE__ . ' file processor created',getlogger(__PACKAGE__));
|
||||
|
||||
return $self;
|
||||
|
||||
}
|
||||
|
||||
sub extractfields {
|
||||
my ($context,$line_ref) = @_;
|
||||
my $separator = $context->{instance}->{field_separator};
|
||||
$$line_ref =~ s/^ +//;
|
||||
$$line_ref =~ s/ +$//;
|
||||
return undef if length($$line_ref) == 0;
|
||||
return undef if $$line_ref =~ /^#/;
|
||||
my @fields = split(/$separator/,$$line_ref,-1);
|
||||
return \@fields;
|
||||
}
|
||||
|
||||
1;
|
||||
Loading…
Reference in new issue