TT#18882 implement teletek importer #6

+ parse callforwards file
+ populate sip username

Change-Id: I1a53b343e23dbf3b6745eefdfa5228a9fba57ae7
changes/60/15560/1
Rene Krenn 9 years ago
parent 52ad964c53
commit 5b04e7653a

@ -42,6 +42,7 @@ use NGCP::BulkProcessor::Dao::Trunk::kamailio::voicemail_users qw();
use NGCP::BulkProcessor::Projects::Migration::Teletek::Dao::import::Subscriber qw();
use NGCP::BulkProcessor::Projects::Migration::Teletek::Dao::import::AllowedCli qw();
use NGCP::BulkProcessor::Projects::Migration::Teletek::Dao::import::Clir qw();
use NGCP::BulkProcessor::Projects::Migration::Teletek::Dao::import::CallForward qw();
use NGCP::BulkProcessor::RestRequests::Trunk::Resellers qw();
use NGCP::BulkProcessor::RestRequests::Trunk::Domains qw();
@ -137,6 +138,9 @@ sub check_import_db_tables {
($check_result,$message) = _check_table($message_prefix,'NGCP::BulkProcessor::Projects::Migration::Teletek::Dao::import::Clir');
$result &= $check_result; push(@$messages,$message);
($check_result,$message) = _check_table($message_prefix,'NGCP::BulkProcessor::Projects::Migration::Teletek::Dao::import::CallForward');
$result &= $check_result; push(@$messages,$message);
return $result;
}

@ -0,0 +1,335 @@
package NGCP::BulkProcessor::Projects::Migration::Teletek::Dao::import::CallForward;
use strict;
## no critic
use NGCP::BulkProcessor::Projects::Migration::Teletek::ProjectConnectorPool qw(
get_import_db
destroy_all_dbs
);
#import_db_tableidentifier
use NGCP::BulkProcessor::SqlProcessor qw(
registertableinfo
create_targettable
checktableinfo
copy_row
insert_stmt
);
#process_table
use NGCP::BulkProcessor::SqlRecord qw();
#use NGCP::BulkProcessor::Projects::Migration::Teletek::Dao::import::Subscriber qw();
require Exporter;
our @ISA = qw(Exporter NGCP::BulkProcessor::SqlRecord);
our @EXPORT_OK = qw(
create_table
gettablename
check_table
getinsertstatement
getupsertstatement
@fieldnames
findby_ccacsntypedestination
findby_sipusername
countby_ccacsntype
update_delta
findby_delta
countby_delta
$deleted_delta
$updated_delta
$added_delta
);
#findby_sipusername
#countby_clir
my $tablename = 'callforward';
my $get_db = \&get_import_db;
#my $get_tablename = \&import_db_tableidentifier;
our @fieldnames = (
"cc",
"ac",
"sn",
"type",
"destination",
"priority",
"timeout",
"ringtimeout",
#calculated fields at the end!
"sip_username",
'rownum',
'filename',
);
my $expected_fieldnames = [
@fieldnames,
'delta',
];
# table creation:
my $primarykey_fieldnames = [ "cc", "ac", "sn", "type", "destination" ];
my $indexes = {
$tablename . '_rownum' => [ 'rownum(11)' ],
$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($class,$get_db,
$tablename,$expected_fieldnames,$indexes);
copy_row($self,shift,$expected_fieldnames);
return $self;
}
sub create_table {
my ($truncate) = @_;
my $db = &$get_db();
registertableinfo($db,__PACKAGE__,$tablename,$expected_fieldnames,$indexes,$primarykey_fieldnames);
return create_targettable($db,__PACKAGE__,$db,__PACKAGE__,$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_ccacsntypedestination {
my ($cc,$ac,$sn,$type,$destination,$load_recursive) = @_;
check_table();
my $db = &$get_db();
my $table = $db->tableidentifier($tablename);
return [] unless (defined $cc or defined $ac or defined $sn or defined $type or defined $destination);
my $rows = $db->db_get_all_arrayref(
'SELECT * FROM ' .
$table .
' WHERE ' .
$db->columnidentifier('cc') . ' = ?' .
' AND ' . $db->columnidentifier('ac') . ' = ?' .
' AND ' . $db->columnidentifier('sn') . ' = ?' .
' AND ' . $db->columnidentifier('type') . ' = ?' .
' AND ' . $db->columnidentifier('destination') . ' = ?'
,$cc,$ac,$sn,$type,$destination);
return buildrecords_fromrows($rows,$load_recursive)->[0];
}
sub findby_sipusername {
my ($sip_username,$load_recursive) = @_;
check_table();
my $db = &$get_db();
my $table = $db->tableidentifier($tablename);
#return [] unless (defined $cc or defined $ac or defined $sn);
my $rows = $db->db_get_all_arrayref(
'SELECT * FROM ' .
$table .
' WHERE ' .
$db->columnidentifier('sip_username') . ' = ?'
,$sip_username);
return buildrecords_fromrows($rows,$load_recursive);
}
sub countby_ccacsntype {
my ($cc,$ac,$sn,$type) = @_;
check_table();
my $db = &$get_db();
my $table = $db->tableidentifier($tablename);
my $stmt = 'SELECT COUNT(*) FROM ' . $table;
my @params = ();
my @terms = ();
if (defined $cc or defined $ac or defined $sn) {
push(@terms,
$db->columnidentifier('cc') . ' = ?',
$db->columnidentifier('ac') . ' = ?',
$db->columnidentifier('sn') . ' = ?');
push(@params,$cc,$ac,$sn);
}
if (defined $type) {
push(@terms,
$db->columnidentifier('type') . ' = ?');
push(@params,$type);
}
if ((scalar @terms) > 0) {
$stmt .= ' WHERE ' . join(' AND ',@terms);
}
return $db->db_get_value($stmt,@params);
}
sub update_delta {
my ($cc,$ac,$sn,$type,$destination,$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 $cc or defined $ac or defined $sn or defined $type or defined $destination) {
$stmt .= ' WHERE ' .
$db->columnidentifier('cc') . ' = ?' .
' AND ' . $db->columnidentifier('ac') . ' = ?' .
' AND ' . $db->columnidentifier('sn') . ' = ?' .
' AND ' . $db->columnidentifier('type') . ' = ?' .
' AND ' . $db->columnidentifier('destination') . ' = ?';
push(@params,$cc,$ac,$sn,$type,$destination);
}
return $db->db_do($stmt,@params);
}
sub countby_delta {
my ($deltas) = @_;
check_table();
my $db = &$get_db();
my $table = $db->tableidentifier($tablename);
my $stmt = 'SELECT COUNT(*) FROM ' . $table . ' WHERE 1=1';
my @params = ();
if (defined $deltas and 'HASH' eq ref $deltas) {
foreach my $in (keys %$deltas) {
my @values = (defined $deltas->{$in} and 'ARRAY' eq ref $deltas->{$in} ? @{$deltas->{$in}} : ($deltas->{$in}));
$stmt .= ' AND ' . $db->columnidentifier('delta') . ' ' . $in . ' (' . substr(',?' x scalar @values,1) . ')';
push(@params,@values);
}
} elsif (defined $deltas and length($deltas) > 0) {
$stmt .= ' AND ' . $db->columnidentifier('delta') . ' = ?';
push(@params,$deltas);
}
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,__PACKAGE__,$insert_ignore);
}
sub getupsertstatement {
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 \'' . $updated_delta . '\' FROM ' . $table . ' WHERE ' .
$db->columnidentifier('cc') . ' = ?' .
' AND ' . $db->columnidentifier('ac') . ' = ?' .
' AND ' . $db->columnidentifier('sn') . ' = ?' .
' AND ' . $db->columnidentifier('type') . ' = ?' .
' AND ' . $db->columnidentifier('destination') . ' = ?';
push(@values,'COALESCE((' . $stmt . '), \'' . $added_delta . '\')');
} else {
push(@values,'?');
}
}
$upsert_stmt .= ' VALUES (' . join(',',@values) . ')';
return $upsert_stmt;
}
sub gettablename {
return $tablename;
}
sub check_table {
return checktableinfo($get_db,
__PACKAGE__,$tablename,
$expected_fieldnames,
$indexes);
}
1;

@ -21,6 +21,10 @@ use NGCP::BulkProcessor::Projects::Migration::Teletek::Settings qw(
$ignore_clir_unique
$clir_import_single_row_txn
$callforward_import_numofthreads
$ignore_callforward_unique
$callforward_import_single_row_txn
$skip_errors
);
use NGCP::BulkProcessor::Logging qw (
@ -54,6 +58,7 @@ our @EXPORT_OK = qw(
import_subscriber
import_allowedcli
import_clir
import_callforward
);
sub import_subscriber {
@ -87,9 +92,6 @@ sub import_subscriber {
$record->{cc} //= '';
$record->{ac} //= '';
$record->{sn} //= '';
$record->{cc} = trim($record->{cc});
$record->{ac} = trim($record->{ac});
$record->{sn} = trim($record->{sn});
$record->{rownum} = $rownum;
$record->{filename} = $file;
my %r = %$record;
@ -237,9 +239,6 @@ sub import_allowedcli {
$record->{cc} //= '';
$record->{ac} //= '';
$record->{sn} //= '';
$record->{cc} = trim($record->{cc});
$record->{ac} = trim($record->{ac});
$record->{sn} = trim($record->{sn});
$record->{rownum} = $rownum;
$record->{filename} = $file;
@ -604,6 +603,184 @@ sub _insert_clir_rows {
}
sub import_callforward {
my (@files) = @_;
my $result = NGCP::BulkProcessor::Projects::Migration::Teletek::Dao::import::CallForward::create_table(0);
foreach my $file (@files) {
$result &= _import_callforward_checks($file);
}
my $importer = NGCP::BulkProcessor::Projects::Migration::Teletek::FileProcessors::CSVFile->new($callforward_import_numofthreads);
my $upsert = _import_callforward_reset_delta();
destroy_all_dbs(); #close all db connections before forking..
my $warning_count :shared = 0;
foreach my $file (@files) {
$result &= $importer->process(
file => $file,
process_code => sub {
my ($context,$rows,$row_offset) = @_;
my $rownum = $row_offset;
my @callforward_rows = ();
foreach my $row (@$rows) {
$rownum++;
next if (scalar @$row) == 0;
$row = [ map { local $_ = $_; trim($_); } @$row ];
my $record = NGCP::BulkProcessor::Projects::Migration::Teletek::Dao::import::CallForward->new($row);
$record->{cc} //= '';
$record->{ac} //= '';
$record->{sn} //= '';
$record->{rownum} = $rownum;
$record->{filename} = $file;
if (my $subscriber = NGCP::BulkProcessor::Projects::Migration::Teletek::Dao::import::Subscriber::findby_ccacsn($record->{cc},$record->{ac},$record->{sn})) {
$record->{sip_username} = $subscriber->{sip_username};
} elsif (my $allowedcli = NGCP::BulkProcessor::Projects::Migration::Teletek::Dao::import::AllowedCli::findby_ccacsn($record->{cc},$record->{ac},$record->{sn})) {
$record->{sip_username} = $subscriber->{sip_username};
} else {
my $number = $record->{cc} . $record->{ac} . $record->{sn};
if ($skip_errors) {
_warn($context,"no sip username found for number $number");
} else {
_error($context,"no sip username found for number $number");
}
next;
}
my %r = %$record; my @callforward_row = @r{@NGCP::BulkProcessor::Projects::Migration::Teletek::Dao::import::CallForward::fieldnames};
if ($context->{upsert}) {
push(@callforward_row,$record->{cc},$record->{ac},$record->{sn},$record->{type},$record->{destination});
} else {
push(@callforward_row,$NGCP::BulkProcessor::Projects::Migration::Teletek::Dao::import::CallForward::added_delta);
}
push(@callforward_rows,\@callforward_row);
if ($callforward_import_single_row_txn and (scalar @callforward_rows) > 0) {
while (defined (my $callforward_row = shift @callforward_rows)) {
if ($skip_errors) {
eval { _insert_callforward_rows($context,[$callforward_row]); };
_warn($context,$@) if $@;
} else {
_insert_callforward_rows($context,[$callforward_row]);
}
}
}
}
if (not $callforward_import_single_row_txn and (scalar @callforward_rows) > 0) {
if ($skip_errors) {
eval { _insert_callforward_rows($context,\@callforward_rows); };
_warn($context,$@) if $@;
} else {
_insert_callforward_rows($context,\@callforward_rows);
}
}
#use Data::Dumper;
#print Dumper(\@subscriber_rows);
return 1;
},
init_process_context_code => sub {
my ($context)= @_;
$context->{db} = &get_import_db(); # keep ref count low..
$context->{upsert} = $upsert;
$context->{error_count} = 0;
$context->{warning_count} = 0;
},
uninit_process_context_code => sub {
my ($context)= @_;
undef $context->{db};
destroy_all_dbs();
{
lock $warning_count;
$warning_count += $context->{warning_count};
}
},
multithreading => $import_multithreading
);
}
return ($result,$warning_count);
}
sub _import_callforward_checks {
my ($file) = @_;
my $result = 1;
my $subscribercount = 0;
eval {
$subscribercount = NGCP::BulkProcessor::Projects::Migration::Teletek::Dao::import::Subscriber::countby_ccacsn();
};
if ($@ or $subscribercount == 0) {
fileprocessingerror($file,'please import subscribers first',getlogger(__PACKAGE__));
$result = 0; #even in skip-error mode..
}
my $allowedclicount = 0;
eval {
$allowedclicount = NGCP::BulkProcessor::Projects::Migration::Teletek::Dao::import::AllowedCli::countby_ccacsn();
};
if ($@ or $allowedclicount == 0) {
fileprocessingerror($file,'please import allowedclis first',getlogger(__PACKAGE__));
$result = 0; #even in skip-error mode..
}
return $result;
}
sub _import_callforward_reset_delta {
my $upsert = 0;
if (NGCP::BulkProcessor::Projects::Migration::Teletek::Dao::import::CallForward::countby_ccacsntype() > 0) {
processing_info(threadid(),'resetting delta of ' .
NGCP::BulkProcessor::Projects::Migration::Teletek::Dao::import::CallForward::update_delta(undef,
$NGCP::BulkProcessor::Projects::Migration::Teletek::Dao::import::CallForward::deleted_delta) .
' callforward records',getlogger(__PACKAGE__));
$upsert |= 1;
}
return $upsert;
}
sub _insert_callforward_rows {
my ($context,$callforward_rows) = @_;
$context->{db}->db_do_begin(
($context->{upsert} ?
NGCP::BulkProcessor::Projects::Migration::Teletek::Dao::import::CallForward::getupsertstatement()
: NGCP::BulkProcessor::Projects::Migration::Teletek::Dao::import::CallForward::getinsertstatement($ignore_callforward_unique)),
#NGCP::BulkProcessor::Projects::Migration::Teletek::Dao::import::Subscriber::gettablename(),
#lock
);
eval {
$context->{db}->db_do_rowblock($callforward_rows);
$context->{db}->db_finish();
};
my $err = $@;
if ($err) {
eval {
$context->{db}->db_finish(1);
};
die($err);
}
}
sub _error {
my ($context,$message) = @_;

@ -71,6 +71,11 @@ our @EXPORT_OK = qw(
$ignore_clir_unique
$clir_import_single_row_txn
@callforward_filenames
$callforward_import_numofthreads
$ignore_callforward_unique
$callforward_import_single_row_txn
$provision_subscriber_multithreading
$provision_subscriber_numofthreads
$webpassword_length
@ -134,6 +139,11 @@ our $clir_import_numofthreads = $cpucount;
our $ignore_clir_unique = 0;
our $clir_import_single_row_txn = 1;
our @callforward_filenames = ();
our $callforward_import_numofthreads = $cpucount;
our $ignore_callforward_unique = 0;
our $callforward_import_single_row_txn = 1;
our $provision_subscriber_multithreading = $enablemultithreading;
our $provision_subscriber_numofthreads = $cpucount;
our $webpassword_length = 8;
@ -197,6 +207,11 @@ sub update_settings {
$ignore_clir_unique = $data->{ignore_clir_unique} if exists $data->{ignore_clir_unique};
$clir_import_single_row_txn = $data->{clir_import_single_row_txn} if exists $data->{clir_import_single_row_txn};
@callforward_filenames = _get_import_filenames(\@callforward_filenames,$data,'callforward_filenames');
$callforward_import_numofthreads = _get_numofthreads($cpucount,$data,'callforward_import_numofthreads');
$ignore_callforward_unique = $data->{ignore_callforward_unique} if exists $data->{ignore_callforward_unique};
$callforward_import_single_row_txn = $data->{callforward_import_single_row_txn} if exists $data->{callforward_import_single_row_txn};
$provision_subscriber_multithreading = $data->{provision_subscriber_multithreading} if exists $data->{provision_subscriber_multithreading};
$provision_subscriber_numofthreads = _get_numofthreads($cpucount,$data,'provision_subscriber_numofthreads');
$webpassword_length = $data->{webpassword_length} if exists $data->{webpassword_length};

@ -29,6 +29,8 @@ use NGCP::BulkProcessor::Projects::Migration::Teletek::Settings qw(
@allowedcli_filenames
@clir_filenames
@callforward_filenames
);
#$allowed_ips
@ -68,6 +70,7 @@ use NGCP::BulkProcessor::Projects::Migration::Teletek::ProjectConnectorPool qw(d
use NGCP::BulkProcessor::Projects::Migration::Teletek::Dao::import::Subscriber qw();
use NGCP::BulkProcessor::Projects::Migration::Teletek::Dao::import::AllowedCli qw();
use NGCP::BulkProcessor::Projects::Migration::Teletek::Dao::import::Clir qw();
use NGCP::BulkProcessor::Projects::Migration::Teletek::Dao::import::CallForward qw();
use NGCP::BulkProcessor::Dao::Trunk::billing::contracts qw();
use NGCP::BulkProcessor::Dao::Trunk::billing::voip_subscribers qw();
@ -90,6 +93,7 @@ use NGCP::BulkProcessor::Projects::Migration::Teletek::Import qw(
import_subscriber
import_allowedcli
import_clir
import_callforward
);
use NGCP::BulkProcessor::Projects::Migration::Teletek::Provisioning qw(
@ -135,6 +139,11 @@ push(@TASK_OPTS,$import_clir_task_opt);
my $import_truncate_clir_task_opt = 'truncate_clir';
push(@TASK_OPTS,$import_truncate_clir_task_opt);
my $import_callforward_task_opt = 'import_callforward';
push(@TASK_OPTS,$import_callforward_task_opt);
my $import_truncate_callforward_task_opt = 'truncate_callforward';
push(@TASK_OPTS,$import_truncate_callforward_task_opt);
my $create_subscriber_task_opt = 'create_subscriber';
push(@TASK_OPTS,$create_subscriber_task_opt);
@ -206,6 +215,11 @@ sub main() {
} elsif (lc($import_truncate_clir_task_opt) eq lc($task)) {
$result &= import_truncate_clir_task(\@messages) if taskinfo($import_truncate_clir_task_opt,$result);
} elsif (lc($import_callforward_task_opt) eq lc($task)) {
$result &= import_callforward_task(\@messages) if taskinfo($import_callforward_task_opt,$result);
} elsif (lc($import_truncate_callforward_task_opt) eq lc($task)) {
$result &= import_truncate_callforward_task(\@messages) if taskinfo($import_truncate_callforward_task_opt,$result);
} elsif (lc($create_subscriber_task_opt) eq lc($task)) {
if (taskinfo($create_subscriber_task_opt,$result,1)) {
next unless check_dry();
@ -475,6 +489,68 @@ sub import_truncate_clir_task {
sub import_callforward_task {
my ($messages) = @_;
my ($result,$warning_count) = (0,0);
eval {
($result,$warning_count) = import_callforward(@callforward_filenames);
};
my $err = $@;
my $stats = ": $warning_count warnings";
eval {
$stats .= "\n total callforward records: " .
NGCP::BulkProcessor::Projects::Migration::Teletek::Dao::import::CallForward::countby_ccacsntype() . ' rows';
my $added_count = NGCP::BulkProcessor::Projects::Migration::Teletek::Dao::import::CallForward::countby_delta(
$NGCP::BulkProcessor::Projects::Migration::Teletek::Dao::import::CallForward::added_delta
);
$stats .= "\n new: $added_count rows";
my $existing_count = NGCP::BulkProcessor::Projects::Migration::Teletek::Dao::import::CallForward::countby_delta(
$NGCP::BulkProcessor::Projects::Migration::Teletek::Dao::import::CallForward::updated_delta
);
$stats .= "\n existing: $existing_count rows";
my $deleted_count = NGCP::BulkProcessor::Projects::Migration::Teletek::Dao::import::CallForward::countby_delta(
$NGCP::BulkProcessor::Projects::Migration::Teletek::Dao::import::CallForward::deleted_delta
);
$stats .= "\n removed: $deleted_count rows";
};
if ($err or !$result) {
push(@$messages,"importing callforwards INCOMPLETE$stats");
} else {
push(@$messages,"importing callforwards completed$stats");
}
destroy_all_dbs(); #every task should leave with closed connections.
return $result;
}
sub import_truncate_callforward_task {
my ($messages) = @_;
my $result = 0;
eval {
$result = NGCP::BulkProcessor::Projects::Migration::Teletek::Dao::import::CallForward::create_table(1);
};
my $err = $@;
my $stats = '';
eval {
$stats .= "\n total callforward records: " .
NGCP::BulkProcessor::Projects::Migration::Teletek::Dao::import::CallForward::countby_ccacsntype() . ' rows';
};
if ($err or !$result) {
push(@$messages,"truncating imported callforwards INCOMPLETE$stats");
} else {
push(@$messages,"truncating imported callforwards completed$stats");
}
destroy_all_dbs(); #every task should leave with closed connections.
return $result;
}
sub create_subscriber_task {
my ($messages) = @_;

@ -29,15 +29,17 @@ clir_import_numofthreads = 2
ignore_clir_unique = 0
clir_import_single_row_txn = 1
callforward_filenames = /home/rkrenn/temp/teletek/export_Call_forwards_170823.csv
callforward_import_numofthreads = 2
ignore_callforward_unique = 0
callforward_import_single_row_txn = 1
provision_subscriber_multithreading = 1
#provision_subscriber_numofthreads = 6
webpassword_length = 8
webusername_length = 8
set_allowed_ips_multithreading = 1
#set_allowed_ips_numofthreads = 6
allowed_ips=127.0.0.1/24,127.0.0.2/24
set_call_forwards_multithreading = 1
#set_call_forwards_numofthreads = 6
@ -52,7 +54,3 @@ cfna_timeouts = 300
ringtimeout = 20
#cfnumber_exclude_pattern =
cfnumber_trim_pattern = ^05\d{2}
set_preference_bulk_multithreading = 1
#set_preference_bulk_numofthreads = 6
concurrent_max_total = 2

Loading…
Cancel
Save