fix tool to update acc_trash and re-enqueue records accumulated during 3.x->4.x updates due to introduction of cdr lnp prefix fields. Change-Id: Iea6140edccc1aeb456f6584b7d8335e18ba1b69dchanges/00/12500/6
Rene Krenn
9 years ago
parent
fbbe82204a
commit
37787469c1
@ -0,0 +1,203 @@
|
||||
package NGCP::BulkProcessor::Dao::Trunk::accounting::cdr;
|
||||
use strict;
|
||||
|
||||
## no critic
|
||||
|
||||
use NGCP::BulkProcessor::Logging qw(
|
||||
getlogger
|
||||
rowsdeleted
|
||||
);
|
||||
|
||||
use NGCP::BulkProcessor::ConnectorPool qw(
|
||||
get_accounting_db
|
||||
);
|
||||
|
||||
use NGCP::BulkProcessor::SqlProcessor qw(
|
||||
checktableinfo
|
||||
|
||||
copy_row
|
||||
);
|
||||
use NGCP::BulkProcessor::SqlRecord qw();
|
||||
|
||||
require Exporter;
|
||||
our @ISA = qw(Exporter NGCP::BulkProcessor::SqlRecord);
|
||||
our @EXPORT_OK = qw(
|
||||
gettablename
|
||||
check_table
|
||||
|
||||
delete_callids
|
||||
|
||||
);
|
||||
#process_records
|
||||
#delete_ids
|
||||
|
||||
my $tablename = 'cdr';
|
||||
my $get_db = \&get_accounting_db;
|
||||
|
||||
my $expected_fieldnames = [
|
||||
"id",
|
||||
"update_time",
|
||||
"source_user_id",
|
||||
"source_provider_id",
|
||||
"source_external_subscriber_id",
|
||||
"source_external_contract_id",
|
||||
"source_account_id",
|
||||
"source_user",
|
||||
"source_domain",
|
||||
"source_cli",
|
||||
"source_clir",
|
||||
"source_ip",
|
||||
"source_gpp0",
|
||||
"source_gpp1",
|
||||
"source_gpp2",
|
||||
"source_gpp3",
|
||||
"source_gpp4",
|
||||
"source_gpp5",
|
||||
"source_gpp6",
|
||||
"source_gpp7",
|
||||
"source_gpp8",
|
||||
"source_gpp9",
|
||||
"source_lnp_prefix",
|
||||
"destination_user_id",
|
||||
"destination_provider_id",
|
||||
"destination_external_subscriber_id",
|
||||
"destination_external_contract_id",
|
||||
"destination_account_id",
|
||||
"destination_user",
|
||||
"destination_domain",
|
||||
"destination_user_dialed",
|
||||
"destination_user_in",
|
||||
"destination_domain_in",
|
||||
"destination_gpp0",
|
||||
"destination_gpp1",
|
||||
"destination_gpp2",
|
||||
"destination_gpp3",
|
||||
"destination_gpp4",
|
||||
"destination_gpp5",
|
||||
"destination_gpp6",
|
||||
"destination_gpp7",
|
||||
"destination_gpp8",
|
||||
"destination_gpp9",
|
||||
"destination_lnp_prefix",
|
||||
"peer_auth_user",
|
||||
"peer_auth_realm",
|
||||
"call_type",
|
||||
"call_status",
|
||||
"call_code",
|
||||
"init_time",
|
||||
"start_time",
|
||||
"duration",
|
||||
"call_id",
|
||||
"source_carrier_cost",
|
||||
"source_reseller_cost",
|
||||
"source_customer_cost",
|
||||
"source_carrier_free_time",
|
||||
"source_reseller_free_time",
|
||||
"source_customer_free_time",
|
||||
"source_carrier_billing_fee_id",
|
||||
"source_reseller_billing_fee_id",
|
||||
"source_customer_billing_fee_id",
|
||||
"source_carrier_billing_zone_id",
|
||||
"source_reseller_billing_zone_id",
|
||||
"source_customer_billing_zone_id",
|
||||
"destination_carrier_cost",
|
||||
"destination_reseller_cost",
|
||||
"destination_customer_cost",
|
||||
"destination_carrier_free_time",
|
||||
"destination_reseller_free_time",
|
||||
"destination_customer_free_time",
|
||||
"destination_carrier_billing_fee_id",
|
||||
"destination_reseller_billing_fee_id",
|
||||
"destination_customer_billing_fee_id",
|
||||
"destination_carrier_billing_zone_id",
|
||||
"destination_reseller_billing_zone_id",
|
||||
"destination_customer_billing_zone_id",
|
||||
"frag_carrier_onpeak",
|
||||
"frag_reseller_onpeak",
|
||||
"frag_customer_onpeak",
|
||||
"is_fragmented",
|
||||
"split",
|
||||
"rated_at",
|
||||
"rating_status",
|
||||
"exported_at",
|
||||
"export_status",
|
||||
];
|
||||
|
||||
my $indexes = {};
|
||||
|
||||
my $insert_unique_fields = [];
|
||||
|
||||
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 delete_callids {
|
||||
|
||||
my ($xa_db,$callids) = @_;
|
||||
|
||||
check_table();
|
||||
my $db = &$get_db();
|
||||
$xa_db //= $db;
|
||||
my $table = $db->tableidentifier($tablename);
|
||||
|
||||
my $stmt = 'DELETE FROM ' . $table . ' WHERE ' .
|
||||
$db->columnidentifier('call_id') . ' IN (' . substr(',?' x scalar @$callids,1) . ')';
|
||||
my @params = @$callids;
|
||||
|
||||
my $count;
|
||||
if ($count = $xa_db->db_do($stmt,@params)) {
|
||||
rowsdeleted($db,$tablename,$count,$count,getlogger(__PACKAGE__));
|
||||
return 1;
|
||||
} else {
|
||||
rowsdeleted($db,$tablename,0,0,getlogger(__PACKAGE__));
|
||||
return 0;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
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 gettablename {
|
||||
|
||||
return $tablename;
|
||||
|
||||
}
|
||||
|
||||
sub check_table {
|
||||
|
||||
return checktableinfo($get_db,
|
||||
__PACKAGE__,$tablename,
|
||||
$expected_fieldnames,
|
||||
$indexes);
|
||||
|
||||
}
|
||||
|
||||
1;
|
||||
@ -0,0 +1,152 @@
|
||||
package NGCP::BulkProcessor::Dao::Trunk::kamailio::acc;
|
||||
use strict;
|
||||
|
||||
## no critic
|
||||
|
||||
use NGCP::BulkProcessor::Logging qw(
|
||||
getlogger
|
||||
rowinserted
|
||||
);
|
||||
|
||||
use NGCP::BulkProcessor::ConnectorPool qw(
|
||||
get_kamailio_db
|
||||
);
|
||||
|
||||
use NGCP::BulkProcessor::SqlProcessor qw(
|
||||
checktableinfo
|
||||
insert_record
|
||||
copy_row
|
||||
);
|
||||
use NGCP::BulkProcessor::SqlRecord qw();
|
||||
|
||||
require Exporter;
|
||||
our @ISA = qw(Exporter NGCP::BulkProcessor::SqlRecord);
|
||||
our @EXPORT_OK = qw(
|
||||
gettablename
|
||||
check_table
|
||||
|
||||
insert_row
|
||||
count
|
||||
count_callids
|
||||
);
|
||||
|
||||
my $tablename = 'acc';
|
||||
my $get_db = \&get_kamailio_db;
|
||||
|
||||
my $expected_fieldnames = [
|
||||
"id",
|
||||
"method",
|
||||
"from_tag",
|
||||
"to_tag",
|
||||
"callid",
|
||||
"sip_code",
|
||||
"sip_reason",
|
||||
"time",
|
||||
"time_hires",
|
||||
"src_leg",
|
||||
"dst_leg",
|
||||
"dst_user",
|
||||
"dst_ouser",
|
||||
"dst_domain",
|
||||
"src_user",
|
||||
"src_domain",
|
||||
];
|
||||
|
||||
my $indexes = {};
|
||||
|
||||
my $insert_unique_fields = [];
|
||||
|
||||
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 insert_row {
|
||||
|
||||
my $db = &$get_db();
|
||||
my $xa_db = shift // $db;
|
||||
#if ('HASH' eq ref $_[0]) {
|
||||
my ($data,$insert_ignore) = @_;
|
||||
check_table();
|
||||
if (insert_record($db,$xa_db,__PACKAGE__,$data,$insert_ignore,$insert_unique_fields)) {
|
||||
return $xa_db->db_last_insert_id();
|
||||
}
|
||||
#}
|
||||
return undef;
|
||||
|
||||
}
|
||||
|
||||
sub count {
|
||||
|
||||
my ($xa_db) = @_;
|
||||
|
||||
check_table();
|
||||
my $db = &$get_db();
|
||||
$xa_db //= $db;
|
||||
my $table = $db->tableidentifier($tablename);
|
||||
|
||||
my $stmt = 'SELECT COUNT(*) FROM ' . $table;
|
||||
|
||||
return $db->db_get_value($stmt);
|
||||
|
||||
}
|
||||
|
||||
sub count_callids {
|
||||
|
||||
my ($xa_db) = @_;
|
||||
|
||||
check_table();
|
||||
my $db = &$get_db();
|
||||
$xa_db //= $db;
|
||||
my $table = $db->tableidentifier($tablename);
|
||||
|
||||
my $stmt = 'SELECT COUNT(DISTINCT ' . $db->columnidentifier('callid') . ') FROM ' . $table;
|
||||
|
||||
return $db->db_get_value($stmt);
|
||||
|
||||
}
|
||||
|
||||
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 gettablename {
|
||||
|
||||
return $tablename;
|
||||
|
||||
}
|
||||
|
||||
sub check_table {
|
||||
|
||||
return checktableinfo($get_db,
|
||||
__PACKAGE__,$tablename,
|
||||
$expected_fieldnames,
|
||||
$indexes);
|
||||
|
||||
}
|
||||
|
||||
1;
|
||||
@ -0,0 +1,228 @@
|
||||
package NGCP::BulkProcessor::Dao::Trunk::kamailio::acc_trash;
|
||||
use strict;
|
||||
|
||||
## no critic
|
||||
|
||||
use NGCP::BulkProcessor::Logging qw(
|
||||
getlogger
|
||||
rowsdeleted
|
||||
);
|
||||
|
||||
use NGCP::BulkProcessor::ConnectorPool qw(
|
||||
get_kamailio_db
|
||||
destroy_dbs
|
||||
);
|
||||
|
||||
use NGCP::BulkProcessor::SqlProcessor qw(
|
||||
checktableinfo
|
||||
process_table
|
||||
copy_row
|
||||
);
|
||||
use NGCP::BulkProcessor::SqlRecord qw();
|
||||
|
||||
require Exporter;
|
||||
our @ISA = qw(Exporter NGCP::BulkProcessor::SqlRecord);
|
||||
our @EXPORT_OK = qw(
|
||||
gettablename
|
||||
check_table
|
||||
|
||||
process_records
|
||||
delete_callids
|
||||
findby_callids
|
||||
count
|
||||
count_callids
|
||||
);
|
||||
|
||||
my $tablename = 'acc_trash';
|
||||
my $get_db = \&get_kamailio_db;
|
||||
|
||||
my $expected_fieldnames = [
|
||||
"id",
|
||||
"method",
|
||||
"from_tag",
|
||||
"to_tag",
|
||||
"callid",
|
||||
"sip_code",
|
||||
"sip_reason",
|
||||
"time",
|
||||
"time_hires",
|
||||
"src_leg",
|
||||
"dst_leg",
|
||||
"dst_user",
|
||||
"dst_ouser",
|
||||
"dst_domain",
|
||||
"src_user",
|
||||
"src_domain",
|
||||
];
|
||||
|
||||
my $indexes = {};
|
||||
|
||||
my $insert_unique_fields = [];
|
||||
|
||||
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 delete_callids {
|
||||
|
||||
my ($xa_db,$callids) = @_;
|
||||
|
||||
check_table();
|
||||
my $db = &$get_db();
|
||||
$xa_db //= $db;
|
||||
my $table = $db->tableidentifier($tablename);
|
||||
|
||||
my $stmt = 'DELETE FROM ' . $table . ' WHERE ' .
|
||||
$db->columnidentifier('callid') . ' IN (' . substr(',?' x scalar @$callids,1) . ')';
|
||||
my @params = @$callids;
|
||||
|
||||
my $count;
|
||||
if ($count = $xa_db->db_do($stmt,@params)) {
|
||||
rowsdeleted($db,$tablename,$count,$count,getlogger(__PACKAGE__));
|
||||
return 1;
|
||||
} else {
|
||||
rowsdeleted($db,$tablename,0,0,getlogger(__PACKAGE__));
|
||||
return 0;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
sub count {
|
||||
|
||||
my ($xa_db) = @_;
|
||||
|
||||
check_table();
|
||||
my $db = &$get_db();
|
||||
$xa_db //= $db;
|
||||
my $table = $db->tableidentifier($tablename);
|
||||
|
||||
my $stmt = 'SELECT COUNT(*) FROM ' . $table;
|
||||
|
||||
return $db->db_get_value($stmt);
|
||||
|
||||
}
|
||||
|
||||
sub count_callids {
|
||||
|
||||
my ($xa_db) = @_;
|
||||
|
||||
check_table();
|
||||
my $db = &$get_db();
|
||||
$xa_db //= $db;
|
||||
my $table = $db->tableidentifier($tablename);
|
||||
|
||||
my $stmt = 'SELECT COUNT(DISTINCT ' . $db->columnidentifier('callid') . ') FROM ' . $table;
|
||||
|
||||
return $db->db_get_value($stmt);
|
||||
|
||||
}
|
||||
|
||||
sub findby_callids {
|
||||
|
||||
my ($xa_db,$callids,$load_recursive) = @_;
|
||||
|
||||
check_table();
|
||||
my $db = &$get_db();
|
||||
$xa_db //= $db;
|
||||
my $table = $db->tableidentifier($tablename);
|
||||
|
||||
my $stmt = 'SELECT * FROM ' . $table . ' WHERE ' .
|
||||
$db->columnidentifier('callid') . ' IN (' . substr(',?' x scalar @$callids,1) . ')';
|
||||
my @params = @$callids;
|
||||
my $rows = $xa_db->db_get_all_arrayref($stmt,@params);
|
||||
|
||||
return buildrecords_fromrows($rows,$load_recursive);
|
||||
|
||||
}
|
||||
|
||||
|
||||
sub process_records {
|
||||
|
||||
my %params = @_;
|
||||
my ($process_code,
|
||||
$static_context,
|
||||
$init_process_context_code,
|
||||
$uninit_process_context_code,
|
||||
$multithreading,
|
||||
$blocksize,
|
||||
$numofthreads,
|
||||
$load_recursive) = @params{qw/
|
||||
process_code
|
||||
static_context
|
||||
init_process_context_code
|
||||
uninit_process_context_code
|
||||
multithreading
|
||||
blocksize
|
||||
numofthreads
|
||||
load_recursive
|
||||
/};
|
||||
|
||||
check_table();
|
||||
my $db = &$get_db();
|
||||
my $table = $db->tableidentifier($tablename);
|
||||
|
||||
return process_table(
|
||||
get_db => $get_db,
|
||||
class => __PACKAGE__,
|
||||
process_code => sub {
|
||||
my ($context,$rowblock,$row_offset) = @_;
|
||||
return &$process_code($context,buildrecords_fromrows($rowblock,$load_recursive),$row_offset);
|
||||
},
|
||||
static_context => $static_context,
|
||||
init_process_context_code => $init_process_context_code,
|
||||
uninit_process_context_code => $uninit_process_context_code,
|
||||
destroy_reader_dbs_code => \&destroy_dbs,
|
||||
multithreading => $multithreading,
|
||||
blocksize => $blocksize,
|
||||
tableprocessing_threads => $numofthreads,
|
||||
'select' => 'SELECT * FROM ' . $table . ' GROUP BY ' . $db->columnidentifier('callid'),
|
||||
'selectcount' => 'SELECT COUNT(DISTINCT ' . $db->columnidentifier('callid') . ') FROM ' . $table,
|
||||
);
|
||||
}
|
||||
|
||||
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 gettablename {
|
||||
|
||||
return $tablename;
|
||||
|
||||
}
|
||||
|
||||
sub check_table {
|
||||
|
||||
return checktableinfo($get_db,
|
||||
__PACKAGE__,$tablename,
|
||||
$expected_fieldnames,
|
||||
$indexes);
|
||||
|
||||
}
|
||||
|
||||
1;
|
||||
@ -0,0 +1,352 @@
|
||||
package NGCP::BulkProcessor::Projects::Disaster::Acc::AccTrash;
|
||||
use strict;
|
||||
|
||||
## no critic
|
||||
|
||||
use threads::shared qw();
|
||||
#use List::Util qw();
|
||||
use DateTime qw();
|
||||
use Time::HiRes qw(sleep);
|
||||
|
||||
use NGCP::BulkProcessor::Projects::Disaster::Acc::Settings qw(
|
||||
$dry
|
||||
$skip_errors
|
||||
|
||||
$process_acc_trash_multithreading
|
||||
$process_acc_trash_numofthreads
|
||||
$process_acc_trash_blocksize
|
||||
|
||||
$delete_cdr
|
||||
|
||||
$sleep_secs
|
||||
$acc_record_limit
|
||||
);
|
||||
#$set_preference_bulk_numofthreads
|
||||
|
||||
use NGCP::BulkProcessor::Logging qw (
|
||||
getlogger
|
||||
processing_info
|
||||
processing_debug
|
||||
);
|
||||
use NGCP::BulkProcessor::LogError qw(
|
||||
rowprocessingerror
|
||||
rowprocessingwarn
|
||||
);
|
||||
|
||||
use NGCP::BulkProcessor::Dao::Trunk::accounting::cdr qw();
|
||||
use NGCP::BulkProcessor::Dao::Trunk::kamailio::acc qw();
|
||||
use NGCP::BulkProcessor::Dao::Trunk::kamailio::acc_trash qw();
|
||||
|
||||
|
||||
|
||||
use NGCP::BulkProcessor::ConnectorPool qw(
|
||||
|
||||
get_kamailio_db
|
||||
get_accounting_db
|
||||
);
|
||||
#get_xa_db
|
||||
|
||||
use NGCP::BulkProcessor::ConnectorPool qw(
|
||||
destroy_dbs
|
||||
);
|
||||
|
||||
use NGCP::BulkProcessor::Utils qw(threadid);
|
||||
|
||||
require Exporter;
|
||||
our @ISA = qw(Exporter);
|
||||
our @EXPORT_OK = qw(
|
||||
process_acc_trash
|
||||
);
|
||||
|
||||
sub process_acc_trash {
|
||||
|
||||
my ($process_acc_trash,$fix_lnp_prefix_tokens) = @_;
|
||||
my $static_context = {
|
||||
delete_cdr => $delete_cdr,
|
||||
sleep_secs => $sleep_secs,
|
||||
acc_record_limit => $acc_record_limit,
|
||||
process_acc_trash => $process_acc_trash,
|
||||
fix_lnp_prefix_tokens => $fix_lnp_prefix_tokens,
|
||||
};
|
||||
my $result = _process_acc_trash_checks($static_context);
|
||||
|
||||
destroy_dbs();
|
||||
my $warning_count :shared = 0;
|
||||
return ($result && NGCP::BulkProcessor::Dao::Trunk::kamailio::acc_trash::process_records(
|
||||
static_context => $static_context,
|
||||
process_code => sub {
|
||||
my ($context,$records,$row_offset) = @_;
|
||||
my $rownum = $row_offset;
|
||||
if ($context->{sleep_secs} > 0.0 and $context->{acc_record_limit} > 0) {
|
||||
my $sleep = $context->{sleep_secs};
|
||||
while (((my $acc_count = NGCP::BulkProcessor::Dao::Trunk::kamailio::acc::count($context->{acc_db})) + scalar @$records) > $context->{acc_record_limit}) {
|
||||
_info($context,"$acc_count acc records (limit $context->{acc_record_limit}), sleep for $sleep secs ...");
|
||||
sleep($sleep);
|
||||
$sleep *= 2.0 if $sleep < 30.0; #manchester
|
||||
}
|
||||
}
|
||||
foreach my $acc_trash (@$records) {
|
||||
$rownum++;
|
||||
next unless _reset_process_acc_trash_context($context,$acc_trash,$rownum);
|
||||
#if (_delete_cdr($context)) {
|
||||
# _process_acc_trash($context);
|
||||
#}
|
||||
_delete_cdr($context) if $context->{delete_cdr};
|
||||
_process_acc_trash($context) if $context->{process_acc_trash};
|
||||
}
|
||||
|
||||
#return 0;
|
||||
return 1;
|
||||
},
|
||||
init_process_context_code => sub {
|
||||
my ($context)= @_;
|
||||
$context->{acc_db} = &get_kamailio_db();
|
||||
$context->{cdr_db} = &get_accounting_db();
|
||||
$context->{error_count} = 0;
|
||||
$context->{warning_count} = 0;
|
||||
# below is not mandatory..
|
||||
_check_insert_tables();
|
||||
},
|
||||
uninit_process_context_code => sub {
|
||||
my ($context)= @_;
|
||||
undef $context->{acc_db};
|
||||
undef $context->{cdr_db};
|
||||
destroy_dbs();
|
||||
{
|
||||
lock $warning_count;
|
||||
$warning_count += $context->{warning_count};
|
||||
}
|
||||
},
|
||||
load_recursive => 0,
|
||||
multithreading => $process_acc_trash_multithreading,
|
||||
numofthreads => $process_acc_trash_numofthreads,
|
||||
blocksize => $process_acc_trash_blocksize,
|
||||
),$warning_count);
|
||||
}
|
||||
|
||||
|
||||
sub _check_insert_tables {
|
||||
|
||||
#NGCP::BulkProcessor::Dao::mr38::provisioning::voip_usr_preferences::check_table();
|
||||
|
||||
}
|
||||
|
||||
sub _process_acc_trash {
|
||||
my ($context) = @_;
|
||||
|
||||
eval {
|
||||
$context->{acc_db}->db_begin();
|
||||
|
||||
my $incomplete = 0;
|
||||
foreach my $acc_trash (@{$context->{acc_trash}}) {
|
||||
my $src_leg = $acc_trash->{src_leg};
|
||||
#my $move = 1;
|
||||
if ($context->{fix_lnp_prefix_tokens}) {
|
||||
if (defined $src_leg and length($src_leg) > 0) {
|
||||
my @tokens = split(/\|/,$src_leg,-1);
|
||||
my $tokencount = scalar @tokens;
|
||||
if ($tokencount == 24) {
|
||||
$acc_trash->{src_leg} .= '|';
|
||||
} elsif ($tokencount == 25) {
|
||||
_info($context,"($context->{rownum}) " . 'acc trash record id ' . $acc_trash->{id} . " seems to be correct (lnp_prefix), src_leg token count = $tokencount");
|
||||
} elsif ($tokencount > 0) {
|
||||
if ($skip_errors) {
|
||||
_warn($context,"($context->{rownum}) " . 'acc trash record id ' . $acc_trash->{id} . " cannot be fixed (lnp_prefix), src_leg token count = $tokencount");
|
||||
} else {
|
||||
_error($context,"($context->{rownum}) " . 'acc trash record id ' . $acc_trash->{id} . " cannot be fixed (lnp_prefix), src_leg token count = $tokencount");
|
||||
}
|
||||
$incomplete = 1;
|
||||
#$move = 0;
|
||||
#last;
|
||||
}
|
||||
}
|
||||
my $dst_leg = $acc_trash->{dst_leg};
|
||||
if (defined $dst_leg and length($dst_leg) > 0) {
|
||||
my @tokens = split(/\|/,$dst_leg,-1);
|
||||
my $tokencount = scalar @tokens;
|
||||
if ($tokencount == 22) {
|
||||
$acc_trash->{dst_leg} .= '|';
|
||||
} elsif ($tokencount == 23) {
|
||||
_info($context,"($context->{rownum}) " . 'acc trash record id ' . $acc_trash->{id} . " seems to be correct (lnp_prefix), dst_leg token count = $tokencount");
|
||||
} elsif ($tokencount > 0) {
|
||||
if ($skip_errors) {
|
||||
_warn($context,"($context->{rownum}) " . 'acc trash record id ' . $acc_trash->{id} . " cannot be fixed (lnp_prefix), dst_leg token count = $tokencount");
|
||||
} else {
|
||||
_error($context,"($context->{rownum}) " . 'acc trash record id ' . $acc_trash->{id} . " cannot be fixed (lnp_prefix), dst_leg token count = $tokencount");
|
||||
}
|
||||
$incomplete = 1;
|
||||
#$move = 0;
|
||||
#last;
|
||||
}
|
||||
}
|
||||
}
|
||||
unless ($incomplete) {
|
||||
NGCP::BulkProcessor::Dao::Trunk::kamailio::acc::insert_row($context->{acc_db},$acc_trash);
|
||||
_info($context,"($context->{rownum}) " . 'acc trash record id ' . $acc_trash->{id} . " copied",1);
|
||||
}
|
||||
}
|
||||
unless ($incomplete) {
|
||||
NGCP::BulkProcessor::Dao::Trunk::kamailio::acc_trash::delete_callids($context->{acc_db},[ $context->{call_id} ]);
|
||||
_info($context,"($context->{rownum}) " . 'acc trash records for callid ' . $context->{call_id} . " deleted",1);
|
||||
}
|
||||
|
||||
if ($dry or $incomplete) {
|
||||
$context->{acc_db}->db_rollback(0);
|
||||
} else {
|
||||
$context->{acc_db}->db_commit();
|
||||
}
|
||||
|
||||
};
|
||||
my $err = $@;
|
||||
if ($err) {
|
||||
eval {
|
||||
$context->{acc_db}->db_rollback(1);
|
||||
};
|
||||
if ($skip_errors) {
|
||||
_warn($context,"($context->{rownum}) " . 'kamailio database error with acc trash callid ' . $context->{call_id} . ': ' . $err);
|
||||
} else {
|
||||
_error($context,"($context->{rownum}) " . 'kamailio database error with acc trash callid ' . $context->{call_id} . ': ' . $err);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
sub _delete_cdr {
|
||||
my ($context) = @_;
|
||||
|
||||
eval {
|
||||
$context->{cdr_db}->db_begin();
|
||||
|
||||
if (NGCP::BulkProcessor::Dao::Trunk::accounting::cdr::delete_callids($context->{cdr_db},[ $context->{call_id} ])) {
|
||||
_info($context,'cdr id ' . $context->{call_id} . " deleted");
|
||||
} else {
|
||||
_info($context,'no cdr id ' . $context->{call_id} . " to delete",1);
|
||||
}
|
||||
|
||||
if ($dry) {
|
||||
$context->{cdr_db}->db_rollback(0);
|
||||
} else {
|
||||
$context->{cdr_db}->db_commit();
|
||||
}
|
||||
|
||||
};
|
||||
my $err = $@;
|
||||
if ($err) {
|
||||
eval {
|
||||
$context->{cdr_db}->db_rollback(1);
|
||||
};
|
||||
if ($skip_errors) {
|
||||
_warn($context,"($context->{rownum}) " . 'accounting database error with acc trash callid ' . $context->{call_id} . ': ' . $err);
|
||||
} else {
|
||||
_error($context,"($context->{rownum}) " . 'accounting database error with acc trash callid ' . $context->{call_id} . ': ' . $err);
|
||||
}
|
||||
#return 0;
|
||||
}
|
||||
#return 1;
|
||||
|
||||
}
|
||||
|
||||
sub _process_acc_trash_checks {
|
||||
my ($context) = @_;
|
||||
|
||||
my $result = _checks($context);
|
||||
|
||||
return $result;
|
||||
}
|
||||
|
||||
sub _reset_process_acc_trash_context {
|
||||
|
||||
my ($context,$acc_trash,$rownum) = @_;
|
||||
|
||||
my $result = _reset_context($context,$acc_trash,$rownum);
|
||||
|
||||
$context->{call_id} = $acc_trash->{callid};
|
||||
$context->{acc_trash} = NGCP::BulkProcessor::Dao::Trunk::kamailio::acc_trash::findby_callids($context->{acc_db},[ $context->{call_id} ]);
|
||||
|
||||
##$context->{barring_profile} = $imported_subscriber->{barring_profile};
|
||||
##$context->{ncos_level} = $context->{ncos_level_map}->{$context->{barring_profile}};
|
||||
|
||||
##delete $context->{adm_ncos_id_preference_id};
|
||||
|
||||
return $result;
|
||||
|
||||
}
|
||||
|
||||
|
||||
sub _checks {
|
||||
|
||||
my ($context) = @_;
|
||||
|
||||
my $result = 1;
|
||||
my $acctrashcount = 0;
|
||||
eval {
|
||||
$acctrashcount = NGCP::BulkProcessor::Dao::Trunk::kamailio::acc_trash::count();
|
||||
};
|
||||
if ($@ or $acctrashcount == 0) {
|
||||
rowprocessingerror(threadid(),'no acc trash records',getlogger(__PACKAGE__));
|
||||
$result = 0; #even in skip-error mode..
|
||||
}
|
||||
#my $userpasswordcount = 0;
|
||||
#eval {
|
||||
# $userpasswordcount = NGCP::BulkProcessor::Projects::Migration::IPGallery::Dao::import::UsernamePassword::countby_fqdn();
|
||||
#};
|
||||
#if ($@ or $userpasswordcount == 0) {
|
||||
# rowprocessingerror(threadid(),'please import user passwords first',getlogger(__PACKAGE__));
|
||||
# $result = 0; #even in skip-error mode..
|
||||
#}
|
||||
#my $subscribercount = 0;
|
||||
#my $subscriber_barring_profiles = [];
|
||||
#eval {
|
||||
# $subscribercount = NGCP::BulkProcessor::Projects::Migration::IPGallery::Dao::import::Subscriber::countby_subscribernumber();
|
||||
# $subscriber_barring_profiles = NGCP::BulkProcessor::Projects::Migration::IPGallery::Dao::import::Subscriber::list_barringprofiles();
|
||||
#};
|
||||
#if ($@ or $subscribercount == 0) {
|
||||
# rowprocessingerror(threadid(),'please import subscribers first',getlogger(__PACKAGE__));
|
||||
# $result = 0; #even in skip-error mode..
|
||||
#}
|
||||
|
||||
return $result;
|
||||
|
||||
}
|
||||
|
||||
sub _reset_context {
|
||||
|
||||
my ($context,$acc_trash,$rownum) = @_;
|
||||
|
||||
my $result = 1;
|
||||
|
||||
$context->{rownum} = $rownum;
|
||||
|
||||
undef $context->{call_id};
|
||||
undef $context->{acc_trash}; # = $acc_trash;
|
||||
|
||||
return $result;
|
||||
|
||||
}
|
||||
|
||||
sub _error {
|
||||
|
||||
my ($context,$message) = @_;
|
||||
$context->{error_count} = $context->{error_count} + 1;
|
||||
rowprocessingerror($context->{tid},$message,getlogger(__PACKAGE__));
|
||||
|
||||
}
|
||||
|
||||
sub _warn {
|
||||
|
||||
my ($context,$message) = @_;
|
||||
$context->{warning_count} = $context->{warning_count} + 1;
|
||||
rowprocessingwarn($context->{tid},$message,getlogger(__PACKAGE__));
|
||||
|
||||
}
|
||||
|
||||
sub _info {
|
||||
|
||||
my ($context,$message,$debug) = @_;
|
||||
if ($debug) {
|
||||
processing_debug($context->{tid},$message,getlogger(__PACKAGE__));
|
||||
} else {
|
||||
processing_info($context->{tid},$message,getlogger(__PACKAGE__));
|
||||
}
|
||||
}
|
||||
|
||||
1;
|
||||
@ -0,0 +1,118 @@
|
||||
package NGCP::BulkProcessor::Projects::Disaster::Acc::Check;
|
||||
use strict;
|
||||
|
||||
## no critic
|
||||
|
||||
no strict 'refs';
|
||||
|
||||
use NGCP::BulkProcessor::Dao::Trunk::accounting::cdr qw();
|
||||
use NGCP::BulkProcessor::Dao::Trunk::kamailio::acc qw();
|
||||
use NGCP::BulkProcessor::Dao::Trunk::kamailio::acc_trash qw();
|
||||
|
||||
require Exporter;
|
||||
our @ISA = qw(Exporter);
|
||||
our @EXPORT_OK = qw(
|
||||
check_accounting_db_tables
|
||||
check_kamailio_db_tables
|
||||
);
|
||||
#check_rest_get_items
|
||||
|
||||
my $NOK = 'NOK';
|
||||
my $OK = 'ok';
|
||||
|
||||
sub check_kamailio_db_tables {
|
||||
|
||||
my ($messages) = @_;
|
||||
|
||||
my $result = 1;
|
||||
my $check_result;
|
||||
my $message;
|
||||
|
||||
my $message_prefix = 'NGCP kamailio db tables - ';
|
||||
|
||||
($check_result,$message) = _check_table($message_prefix,'NGCP::BulkProcessor::Dao::Trunk::kamailio::acc');
|
||||
$result &= $check_result; push(@$messages,$message);
|
||||
|
||||
($check_result,$message) = _check_table($message_prefix,'NGCP::BulkProcessor::Dao::Trunk::kamailio::acc_trash');
|
||||
$result &= $check_result; push(@$messages,$message);
|
||||
|
||||
return $result;
|
||||
|
||||
}
|
||||
|
||||
sub check_accounting_db_tables {
|
||||
|
||||
my ($messages) = @_;
|
||||
|
||||
my $result = 1;
|
||||
my $check_result;
|
||||
my $message;
|
||||
|
||||
my $message_prefix = 'NGCP accounting db tables - ';
|
||||
|
||||
($check_result,$message) = _check_table($message_prefix,'NGCP::BulkProcessor::Dao::Trunk::accounting::cdr');
|
||||
$result &= $check_result; push(@$messages,$message);
|
||||
|
||||
return $result;
|
||||
|
||||
}
|
||||
|
||||
sub _check_table {
|
||||
|
||||
my ($message_prefix,$module) = @_;
|
||||
my $result = 0;
|
||||
my $message = ($message_prefix // '') . &{$module . '::gettablename'}() . ': ';
|
||||
eval {
|
||||
$result = &{$module . '::check_table'}();
|
||||
};
|
||||
if (@$ or not $result) {
|
||||
return (0,$message . $NOK);
|
||||
} else {
|
||||
return (1,$message . $OK);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
#sub check_rest_get_items {
|
||||
#
|
||||
# my ($messages) = @_;
|
||||
#
|
||||
# my $result = 1;
|
||||
# my $check_result;
|
||||
# my $message;
|
||||
#
|
||||
# my $message_prefix = 'NGCP id\'s/constants - ';
|
||||
#
|
||||
# ($check_result,$message, my $reseller) = _check_rest_get_item($message_prefix,
|
||||
# 'NGCP::BulkProcessor::RestRequests::mr38::Resellers',
|
||||
# $reseller_id,
|
||||
# 'name');
|
||||
# $result &= $check_result; push(@$messages,$message);
|
||||
#
|
||||
#
|
||||
# return $result;
|
||||
#
|
||||
#}
|
||||
|
||||
sub _check_rest_get_item {
|
||||
|
||||
my ($message_prefix,$module,$id,$item_name_field,$get_method,$item_path_method) = @_;
|
||||
my $item = undef;
|
||||
$get_method //= 'get_item';
|
||||
$item_path_method //= 'get_item_path';
|
||||
my $message = ($message_prefix // '') . &{$module . '::' . $item_path_method}($id) . ': ';
|
||||
return (0,$message . $NOK,$item) unless $id;
|
||||
eval {
|
||||
$item = &{$module . '::' . $get_method}($id);
|
||||
};
|
||||
|
||||
if (@$ or not defined $item or ('ARRAY' eq ref $item and (scalar @$item) != 1)) {
|
||||
return (0,$message . $NOK,$item);
|
||||
} else {
|
||||
$item = $item->[0] if ('ARRAY' eq ref $item and (scalar @$item) == 1);
|
||||
return (1,$message . "'" . $item->{$item_name_field} . "' " . $OK,$item);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
1;
|
||||
@ -0,0 +1,134 @@
|
||||
package NGCP::BulkProcessor::Projects::Disaster::Acc::Settings;
|
||||
use strict;
|
||||
|
||||
## no critic
|
||||
|
||||
use NGCP::BulkProcessor::Globals qw(
|
||||
$enablemultithreading
|
||||
$cpucount
|
||||
);
|
||||
#$working_path
|
||||
#create_path
|
||||
|
||||
use NGCP::BulkProcessor::Logging qw(
|
||||
getlogger
|
||||
scriptinfo
|
||||
configurationinfo
|
||||
);
|
||||
|
||||
use NGCP::BulkProcessor::LogError qw(
|
||||
fileerror
|
||||
configurationwarn
|
||||
configurationerror
|
||||
);
|
||||
|
||||
use NGCP::BulkProcessor::LoadConfig qw(
|
||||
split_tuple
|
||||
parse_regexp
|
||||
);
|
||||
use NGCP::BulkProcessor::Utils qw(prompt);
|
||||
#format_number check_ipnet
|
||||
|
||||
require Exporter;
|
||||
our @ISA = qw(Exporter);
|
||||
our @EXPORT_OK = qw(
|
||||
update_settings
|
||||
check_dry
|
||||
|
||||
$defaultsettings
|
||||
$defaultconfig
|
||||
|
||||
$dry
|
||||
$skip_errors
|
||||
$force
|
||||
|
||||
$process_acc_trash_multithreading
|
||||
$process_acc_trash_numofthreads
|
||||
$process_acc_trash_blocksize
|
||||
|
||||
$delete_cdr
|
||||
|
||||
$sleep_secs
|
||||
$acc_record_limit
|
||||
);
|
||||
|
||||
our $defaultconfig = 'config.cfg';
|
||||
our $defaultsettings = 'settings.cfg';
|
||||
|
||||
our $force = 0;
|
||||
our $dry = 0;
|
||||
our $skip_errors = 0;
|
||||
|
||||
our $process_acc_trash_multithreading = $enablemultithreading;
|
||||
our $process_acc_trash_numofthreads = $cpucount;
|
||||
our $process_acc_trash_blocksize = 100;
|
||||
our $delete_cdr = 1;
|
||||
our $sleep_secs = 0.5;
|
||||
our $acc_record_limit = 1000;
|
||||
|
||||
sub update_settings {
|
||||
|
||||
my ($data,$configfile) = @_;
|
||||
|
||||
if (defined $data) {
|
||||
|
||||
my $result = 1;
|
||||
#my $regexp_result;
|
||||
|
||||
#&$configurationinfocode("testinfomessage",$configlogger);
|
||||
|
||||
#$result &= _prepare_working_paths(1);
|
||||
|
||||
$dry = $data->{dry} if exists $data->{dry};
|
||||
$skip_errors = $data->{skip_errors} if exists $data->{skip_errors};
|
||||
|
||||
$process_acc_trash_multithreading = $data->{process_acc_trash_multithreading} if exists $data->{process_acc_trash_multithreading};
|
||||
$process_acc_trash_numofthreads = _get_numofthreads($cpucount,$data,'process_acc_trash_numofthreads');
|
||||
|
||||
$process_acc_trash_blocksize = $data->{process_acc_trash_blocksize} if exists $data->{process_acc_trash_blocksize};
|
||||
$delete_cdr = $data->{delete_cdr} if exists $data->{delete_cdr};
|
||||
|
||||
$sleep_secs = $data->{sleep_secs} if exists $data->{sleep_secs};
|
||||
$acc_record_limit = $data->{acc_record_limit} if exists $data->{acc_record_limit};
|
||||
|
||||
#if (defined $acc_record_limit and defined ) {
|
||||
# configurationerror()
|
||||
#}
|
||||
|
||||
return $result;
|
||||
|
||||
}
|
||||
return 0;
|
||||
|
||||
}
|
||||
|
||||
sub check_dry {
|
||||
|
||||
if ($dry) {
|
||||
scriptinfo('running in dry mode - NGCP databases will not be modified',getlogger(__PACKAGE__));
|
||||
return 1;
|
||||
} else {
|
||||
scriptinfo('NO DRY MODE - NGCP DATABASES WILL BE MODIFIED!',getlogger(__PACKAGE__));
|
||||
if (!$force) {
|
||||
if ('yes' eq lc(prompt("Type 'yes' to proceed: "))) {
|
||||
return 1;
|
||||
} else {
|
||||
return 0;
|
||||
}
|
||||
} else {
|
||||
scriptinfo('force option applied',getlogger(__PACKAGE__));
|
||||
return 1;
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
sub _get_numofthreads {
|
||||
my ($default_value,$data,$key) = @_;
|
||||
my $_numofthreads = $default_value;
|
||||
$_numofthreads = $data->{$key} if exists $data->{$key};
|
||||
$_numofthreads = $cpucount if $_numofthreads > $cpucount;
|
||||
return $_numofthreads;
|
||||
}
|
||||
|
||||
1;
|
||||
@ -0,0 +1,62 @@
|
||||
##general settings:
|
||||
working_path = /var/sipwise
|
||||
cpucount = 4
|
||||
enablemultithreading = 1
|
||||
|
||||
##gearman/service listener config:
|
||||
jobservers = 127.0.0.1:4730
|
||||
|
||||
#provisioning_conf = /etc/ngcp-panel/provisioning.conf
|
||||
|
||||
##NGCP MySQL connectivity - "accounting" db:
|
||||
accounting_host = 192.168.0.84
|
||||
accounting_port = 3306
|
||||
accounting_databasename = accounting
|
||||
accounting_username = root
|
||||
accounting_password =
|
||||
|
||||
##NGCP MySQL connectivity - "billing" db:
|
||||
billing_host = 192.168.0.74
|
||||
billing_port = 3306
|
||||
billing_databasename = billing
|
||||
billing_username = root
|
||||
billing_password =
|
||||
|
||||
##NGCP MySQL connectivity - "provisioning" db:
|
||||
provisioning_host = 192.168.0.74
|
||||
provisioning_port = 3306
|
||||
provisioning_databasename = provisioning
|
||||
provisioning_username = root
|
||||
provisioning_password =
|
||||
|
||||
##NGCP MySQL connectivity - "kamailio" db:
|
||||
kamailio_host = 192.168.0.84
|
||||
kamailio_port = 3306
|
||||
kamailio_databasename = kamailio
|
||||
kamailio_username = root
|
||||
kamailio_password =
|
||||
|
||||
##NGCP MySQL connectivity - default db for distributed transactions (XA) to connect to:
|
||||
xa_host = 192.168.0.84
|
||||
xa_port = 3306
|
||||
xa_databasename = ngcp
|
||||
xa_username = root
|
||||
xa_password =
|
||||
|
||||
##NGCP REST-API connectivity:
|
||||
ngcprestapi_uri = https://127.0.0.1:1443
|
||||
ngcprestapi_username = administrator
|
||||
ngcprestapi_password = administrator
|
||||
ngcprestapi_realm = api_admin_http
|
||||
|
||||
##sending email:
|
||||
emailenable = 0
|
||||
erroremailrecipient =
|
||||
warnemailrecipient =
|
||||
completionemailrecipient = rkrenn@sipwise.com
|
||||
doneemailrecipient =
|
||||
|
||||
##logging:
|
||||
fileloglevel = DEBUG
|
||||
screenloglevel = INFO
|
||||
emailloglevel = OFF
|
||||
@ -0,0 +1,250 @@
|
||||
use strict;
|
||||
|
||||
## no critic
|
||||
|
||||
use File::Basename;
|
||||
use Cwd;
|
||||
use lib Cwd::abs_path(File::Basename::dirname(__FILE__) . '/../../../../../');
|
||||
|
||||
use Getopt::Long qw(GetOptions);
|
||||
use Fcntl qw(LOCK_EX LOCK_NB);
|
||||
|
||||
use NGCP::BulkProcessor::Globals qw();
|
||||
use NGCP::BulkProcessor::Projects::Disaster::Acc::Settings qw(
|
||||
update_settings
|
||||
check_dry
|
||||
|
||||
$defaultsettings
|
||||
$defaultconfig
|
||||
$dry
|
||||
$skip_errors
|
||||
$force
|
||||
|
||||
);
|
||||
|
||||
use NGCP::BulkProcessor::Logging qw(
|
||||
init_log
|
||||
getlogger
|
||||
$attachmentlogfile
|
||||
scriptinfo
|
||||
cleanuplogfiles
|
||||
$currentlogfile
|
||||
);
|
||||
use NGCP::BulkProcessor::LogError qw (
|
||||
completion
|
||||
done
|
||||
scriptwarn
|
||||
scripterror
|
||||
filewarn
|
||||
fileerror
|
||||
);
|
||||
use NGCP::BulkProcessor::LoadConfig qw(
|
||||
load_config
|
||||
$SIMPLE_CONFIG_TYPE
|
||||
$YAML_CONFIG_TYPE
|
||||
$ANY_CONFIG_TYPE
|
||||
);
|
||||
use NGCP::BulkProcessor::Array qw(removeduplicates);
|
||||
use NGCP::BulkProcessor::Utils qw(getscriptpath prompt cleanupdir);
|
||||
use NGCP::BulkProcessor::Mail qw(
|
||||
cleanupmsgfiles
|
||||
);
|
||||
|
||||
use NGCP::BulkProcessor::Dao::Trunk::accounting::cdr qw();
|
||||
use NGCP::BulkProcessor::Dao::Trunk::kamailio::acc qw();
|
||||
use NGCP::BulkProcessor::Dao::Trunk::kamailio::acc_trash qw();
|
||||
|
||||
use NGCP::BulkProcessor::Projects::Disaster::Acc::Check qw(
|
||||
check_accounting_db_tables
|
||||
check_kamailio_db_tables
|
||||
);
|
||||
#check_rest_get_items
|
||||
|
||||
use NGCP::BulkProcessor::Projects::Disaster::Acc::AccTrash qw(
|
||||
process_acc_trash
|
||||
);
|
||||
|
||||
#use NGCP::BulkProcessor::Projects::Disaster::Balances::Api qw(
|
||||
# set_call_forwards
|
||||
# set_call_forwards_batch
|
||||
#);
|
||||
|
||||
use NGCP::BulkProcessor::ConnectorPool qw(
|
||||
destroy_dbs
|
||||
);
|
||||
|
||||
scripterror(getscriptpath() . ' already running',getlogger(getscriptpath())) unless flock DATA, LOCK_EX | LOCK_NB; # not tested on windows yet
|
||||
|
||||
my @TASK_OPTS = ();
|
||||
|
||||
my $tasks = [];
|
||||
|
||||
my $check_task_opt = 'check';
|
||||
push(@TASK_OPTS,$check_task_opt);
|
||||
|
||||
my $cleanup_task_opt = 'cleanup';
|
||||
push(@TASK_OPTS,$cleanup_task_opt);
|
||||
|
||||
my $process_acc_trash_task_opt = 'process_acc_trash';
|
||||
push(@TASK_OPTS,$process_acc_trash_task_opt);
|
||||
|
||||
if (init()) {
|
||||
main();
|
||||
exit(0);
|
||||
} else {
|
||||
exit(1);
|
||||
}
|
||||
|
||||
sub init {
|
||||
|
||||
my $configfile = $defaultconfig;
|
||||
my $settingsfile = $defaultsettings;
|
||||
|
||||
return 0 unless GetOptions(
|
||||
"config=s" => \$configfile,
|
||||
"settings=s" => \$settingsfile,
|
||||
"task=s" => $tasks,
|
||||
"dry" => \$dry,
|
||||
"skip-errors" => \$skip_errors,
|
||||
"force" => \$force,
|
||||
); # or scripterror('error in command line arguments',getlogger(getscriptpath()));
|
||||
|
||||
$tasks = removeduplicates($tasks,1);
|
||||
|
||||
my $result = load_config($configfile);
|
||||
init_log();
|
||||
$result &= load_config($settingsfile,\&update_settings,$SIMPLE_CONFIG_TYPE);
|
||||
return $result;
|
||||
|
||||
}
|
||||
|
||||
sub main() {
|
||||
|
||||
my @messages = ();
|
||||
my @attachmentfiles = ();
|
||||
my $result = 1;
|
||||
my $completion = 0;
|
||||
|
||||
if (defined $tasks and 'ARRAY' eq ref $tasks and (scalar @$tasks) > 0) {
|
||||
scriptinfo('skip-errors: processing won\'t stop upon errors',getlogger(__PACKAGE__)) if $skip_errors;
|
||||
foreach my $task (@$tasks) {
|
||||
|
||||
if (lc($check_task_opt) eq lc($task)) {
|
||||
$result &= check_task(\@messages) if taskinfo($check_task_opt,$result);
|
||||
} elsif (lc($cleanup_task_opt) eq lc($task)) {
|
||||
$result &= cleanup_task(\@messages) if taskinfo($cleanup_task_opt,$result);
|
||||
|
||||
} elsif (lc($process_acc_trash_task_opt) eq lc($task)) {
|
||||
if (taskinfo($process_acc_trash_task_opt,$result)) {
|
||||
next unless check_dry();
|
||||
$result &= process_acc_trash_task(\@messages);
|
||||
$completion |= 1;
|
||||
}
|
||||
|
||||
} else {
|
||||
$result = 0;
|
||||
scripterror("unknow task option '" . $task . "', must be one of " . join(', ',@TASK_OPTS),getlogger(getscriptpath()));
|
||||
last;
|
||||
}
|
||||
}
|
||||
} else {
|
||||
$result = 0;
|
||||
scripterror('at least one task option is required. supported tasks: ' . join(', ',@TASK_OPTS),getlogger(getscriptpath()));
|
||||
}
|
||||
|
||||
push(@attachmentfiles,$attachmentlogfile);
|
||||
if ($completion) {
|
||||
completion(join("\n\n",@messages),\@attachmentfiles,getlogger(getscriptpath()));
|
||||
} else {
|
||||
done(join("\n\n",@messages),\@attachmentfiles,getlogger(getscriptpath()));
|
||||
}
|
||||
|
||||
return $result;
|
||||
}
|
||||
|
||||
sub taskinfo {
|
||||
my ($task,$result) = @_;
|
||||
scriptinfo($result ? "starting task: '$task'" : "skipping task '$task' due to previous problems",getlogger(getscriptpath()));
|
||||
return $result;
|
||||
}
|
||||
|
||||
sub check_task {
|
||||
my ($messages) = @_;
|
||||
my @check_messages = ();
|
||||
my $result = check_kamailio_db_tables(\@check_messages);
|
||||
#$result &= ..
|
||||
push(@$messages,join("\n",@check_messages));
|
||||
|
||||
@check_messages = ();
|
||||
$result &= check_accounting_db_tables(\@check_messages);
|
||||
##$result &= ..
|
||||
push(@$messages,join("\n",@check_messages));
|
||||
|
||||
|
||||
destroy_dbs();
|
||||
return $result;
|
||||
}
|
||||
|
||||
sub cleanup_task {
|
||||
my ($messages) = @_;
|
||||
my $result = 0;
|
||||
eval {
|
||||
#cleanupcvsdirs() if $clean_generated;
|
||||
#cleanupdbfiles() if $clean_generated;
|
||||
cleanuplogfiles(\&fileerror,\&filewarn,($currentlogfile,$attachmentlogfile));
|
||||
cleanupmsgfiles(\&fileerror,\&filewarn);
|
||||
#cleanupdir($output_path,1,\&filewarn,getlogger(getscriptpath())) if $clean_generated;
|
||||
#cleanupdir($rollback_path,1,\&filewarn,getlogger(getscriptpath())) if $clean_generated;
|
||||
$result = 1;
|
||||
};
|
||||
|
||||
if ($@ or !$result) {
|
||||
push(@$messages,'cleanup INCOMPLETE');
|
||||
return 0;
|
||||
} else {
|
||||
push(@$messages,'cleanup completed');
|
||||
return 1;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
sub process_acc_trash_task {
|
||||
|
||||
my ($messages) = @_;
|
||||
my ($result,$warning_count) = (0,0);
|
||||
eval {
|
||||
($result,$warning_count) = process_acc_trash(1,1);
|
||||
#if ($batch) {
|
||||
# ($result,$warning_count) = set_barring_profiles_batch();
|
||||
#} else {
|
||||
# ($result,$warning_count) = set_barring_profiles();
|
||||
#}
|
||||
};
|
||||
my $err = $@;
|
||||
my $stats = ($skip_errors ? ": $warning_count warnings" : '');
|
||||
eval {
|
||||
$stats .= "\n total acc trash records left: " .
|
||||
NGCP::BulkProcessor::Dao::Trunk::kamailio::acc_trash::count() . ' rows';
|
||||
$stats .= "\n total acc trash callids left: " .
|
||||
NGCP::BulkProcessor::Dao::Trunk::kamailio::acc_trash::count_callids();
|
||||
};
|
||||
if ($err or !$result) {
|
||||
push(@$messages,"process acc trash INCOMPLETE$stats");
|
||||
} else {
|
||||
push(@$messages,"process acc trash completed$stats");
|
||||
}
|
||||
destroy_dbs(); #every task should leave with closed connections.
|
||||
return $result;
|
||||
|
||||
}
|
||||
|
||||
#END {
|
||||
# # this should not be required explicitly, but prevents Log4Perl's
|
||||
# # "rootlogger not initialized error upon exit..
|
||||
# destroy_all_dbs
|
||||
#}
|
||||
|
||||
__DATA__
|
||||
This exists to allow the locking code at the beginning of the file to work.
|
||||
DO NOT REMOVE THESE LINES!
|
||||
@ -0,0 +1,12 @@
|
||||
|
||||
#dry=0
|
||||
#skip_errors=0
|
||||
|
||||
process_acc_trash_blocksize = 300
|
||||
process_acc_trash_multithreading = 1
|
||||
process_acc_trash_numofthreads = 1
|
||||
|
||||
delete_cdr = 1
|
||||
|
||||
sleep_secs = 2.0
|
||||
acc_record_limit = 2000
|
||||
Loading…
Reference in new issue