TT#18886 implement teletek importer #7
+ parse permanent registrations file Change-Id: If82fbe133b2645e760fc7243ab572c823b804fa9changes/61/15561/1
Rene Krenn
9 years ago
parent
5b04e7653a
commit
0a8d34836c
@ -0,0 +1,282 @@
|
||||
package NGCP::BulkProcessor::Projects::Migration::Teletek::Dao::import::Registration;
|
||||
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_sipusername
|
||||
countby_sipcontact
|
||||
|
||||
update_delta
|
||||
findby_delta
|
||||
countby_delta
|
||||
|
||||
$deleted_delta
|
||||
$updated_delta
|
||||
$added_delta
|
||||
|
||||
);
|
||||
|
||||
my $tablename = 'registration';
|
||||
my $get_db = \&get_import_db;
|
||||
#my $get_tablename = \&import_db_tableidentifier;
|
||||
|
||||
our @fieldnames = (
|
||||
"sip_username",
|
||||
"domain",
|
||||
"sip_contact",
|
||||
|
||||
#calculated fields at the end!
|
||||
'rownum',
|
||||
'filename',
|
||||
);
|
||||
|
||||
my $expected_fieldnames = [
|
||||
@fieldnames,
|
||||
'delta',
|
||||
];
|
||||
|
||||
# table creation:
|
||||
my $primarykey_fieldnames = [ 'sip_username' ];
|
||||
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_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)->[0];
|
||||
|
||||
}
|
||||
|
||||
sub countby_sipcontact {
|
||||
|
||||
my ($sip_contact) = @_;
|
||||
|
||||
check_table();
|
||||
my $db = &$get_db();
|
||||
my $table = $db->tableidentifier($tablename);
|
||||
|
||||
my $stmt = 'SELECT COUNT(*) FROM ' . $table;
|
||||
my @params = ();
|
||||
if (defined $sip_contact) {
|
||||
$stmt .= ' WHERE ' .
|
||||
$db->columnidentifier('sip_contact') . ' = ?';
|
||||
push(@params,$sip_contact);
|
||||
}
|
||||
|
||||
return $db->db_get_value($stmt,@params);
|
||||
|
||||
}
|
||||
|
||||
sub update_delta {
|
||||
|
||||
my ($sip_username,$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 $sip_username) {
|
||||
$stmt .= ' WHERE ' .
|
||||
$db->columnidentifier('sip_username') . ' = ?';
|
||||
push(@params,$sip_username);
|
||||
}
|
||||
|
||||
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('sip_username') . ' = ?';
|
||||
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;
|
||||
Loading…
Reference in new issue