+ control command to reset fsn
+ control command to reset a export stream's export_status
+ produces intact .ama files now
Change-Id: I3b9456416851d2caa3273b54d9e835097663c98b
(cherry picked from commit 72c17907e4)
changes/96/26796/1
Rene Krenn
7 years ago
parent
8e50c98a1d
commit
d28dc6badf
@ -0,0 +1,256 @@
|
||||
package NGCP::BulkProcessor::Dao::Trunk::accounting::mark;
|
||||
use strict;
|
||||
|
||||
## no critic
|
||||
|
||||
use NGCP::BulkProcessor::Logging qw(
|
||||
getlogger
|
||||
rowsdeleted
|
||||
);
|
||||
|
||||
use NGCP::BulkProcessor::ConnectorPool qw(
|
||||
get_accounting_db
|
||||
destroy_dbs
|
||||
);
|
||||
|
||||
use NGCP::BulkProcessor::SqlProcessor qw(
|
||||
checktableinfo
|
||||
copy_row
|
||||
|
||||
insert_record
|
||||
update_record
|
||||
);
|
||||
use NGCP::BulkProcessor::SqlRecord qw();
|
||||
|
||||
require Exporter;
|
||||
our @ISA = qw(Exporter NGCP::BulkProcessor::SqlRecord);
|
||||
our @EXPORT_OK = qw(
|
||||
gettablename
|
||||
check_table
|
||||
|
||||
get_system_mark
|
||||
get_reseller_mark
|
||||
|
||||
set_system_mark
|
||||
set_reseller_mark
|
||||
|
||||
insert_system_mark
|
||||
insert_reseller_mark
|
||||
|
||||
cleanup_system_marks
|
||||
cleanup_reseller_marks
|
||||
);
|
||||
|
||||
my $tablename = 'mark';
|
||||
my $get_db = \&get_accounting_db;
|
||||
|
||||
my $expected_fieldnames = [
|
||||
"id",
|
||||
"collector",
|
||||
"acc_id",
|
||||
];
|
||||
|
||||
my $indexes = {};
|
||||
|
||||
my $insert_unique_fields = [];
|
||||
|
||||
my $system_collector_format = '%s-lastseq';
|
||||
my $reseller_collector_format = '%s-lastseq-%d';
|
||||
|
||||
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 get_system_mark {
|
||||
|
||||
my ($xa_db,$stream) = @_;
|
||||
|
||||
return _get_mark($xa_db,sprintf($system_collector_format,$stream));
|
||||
|
||||
}
|
||||
|
||||
sub get_reseller_mark {
|
||||
|
||||
my ($xa_db,$stream,$reseller_id) = @_;
|
||||
|
||||
return _get_mark($xa_db,sprintf($reseller_collector_format,$stream,$reseller_id // ''));
|
||||
|
||||
}
|
||||
|
||||
sub _get_mark {
|
||||
|
||||
my ($xa_db,$collector) = @_;
|
||||
|
||||
check_table();
|
||||
my $db = &$get_db();
|
||||
$xa_db //= $db;
|
||||
my $table = $db->tableidentifier($tablename);
|
||||
|
||||
my $stmt = 'SELECT MAX(' . $db->columnidentifier('acc_id') . ') FROM ' . $table . ' WHERE ' .
|
||||
$db->columnidentifier('collector') . ' = ?';
|
||||
my @params = ($collector);
|
||||
|
||||
my $mark = $xa_db->db_get_value($stmt,@params);
|
||||
|
||||
return (defined $mark ? $mark : '0');
|
||||
|
||||
}
|
||||
|
||||
sub set_system_mark {
|
||||
|
||||
my ($xa_db,$stream,$mark) = @_;
|
||||
|
||||
return _set_mark($xa_db,sprintf($system_collector_format,$stream),$mark,0);
|
||||
|
||||
}
|
||||
|
||||
sub set_reseller_mark {
|
||||
|
||||
my ($xa_db,$stream,$reseller_id,$mark) = @_;
|
||||
|
||||
return _set_mark($xa_db,sprintf($reseller_collector_format,$stream,$reseller_id // ''),$mark,0);
|
||||
|
||||
}
|
||||
|
||||
sub insert_system_mark {
|
||||
|
||||
my ($xa_db,$stream,$mark) = @_;
|
||||
|
||||
return _set_mark($xa_db,sprintf($system_collector_format,$stream),$mark,1);
|
||||
|
||||
}
|
||||
|
||||
sub insert_reseller_mark {
|
||||
|
||||
my ($xa_db,$stream,$reseller_id,$mark) = @_;
|
||||
|
||||
return _set_mark($xa_db,sprintf($reseller_collector_format,$stream,$reseller_id // ''),$mark,1);
|
||||
|
||||
}
|
||||
|
||||
sub _set_mark {
|
||||
|
||||
my ($xa_db,$collector,$mark,$force_insert) = @_;
|
||||
|
||||
check_table();
|
||||
my $db = &$get_db();
|
||||
$xa_db //= $db;
|
||||
my $table = $db->tableidentifier($tablename);
|
||||
|
||||
my $id;
|
||||
unless ($force_insert) {
|
||||
my $stmt = 'SELECT MAX(t1.id) FROM ' . $table . ' t1 LEFT JOIN ' . $table . ' t2' .
|
||||
' ON t1.collector = t2.collector and t2.acc_id > t1.acc_id'.
|
||||
' WHERE t2.collector IS NULL AND t1.collector = ?';
|
||||
my @params = ($collector);
|
||||
$id = $xa_db->db_get_value($stmt,@params);
|
||||
}
|
||||
|
||||
if (defined $id) {
|
||||
return update_record($get_db,$xa_db,__PACKAGE__,{
|
||||
id => $id,
|
||||
acc_id => $mark,
|
||||
});
|
||||
} else {
|
||||
if (insert_record($db,$xa_db,__PACKAGE__,{
|
||||
collector => $collector,
|
||||
acc_id => $mark
|
||||
},0,$insert_unique_fields)) {
|
||||
return $xa_db->db_last_insert_id();
|
||||
}
|
||||
return undef;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
sub cleanup_system_marks {
|
||||
|
||||
my ($xa_db,$stream) = @_;
|
||||
|
||||
return _cleanup_marks($xa_db,sprintf($system_collector_format,$stream));
|
||||
|
||||
}
|
||||
|
||||
sub cleanup_reseller_marks {
|
||||
|
||||
my ($xa_db,$stream,$reseller_id) = @_;
|
||||
|
||||
return _cleanup_marks($xa_db,sprintf($reseller_collector_format,$stream,$reseller_id // ''));
|
||||
|
||||
}
|
||||
|
||||
sub _cleanup_marks {
|
||||
|
||||
my ($xa_db,$collector) = @_;
|
||||
|
||||
check_table();
|
||||
my $db = &$get_db();
|
||||
$xa_db //= $db;
|
||||
my $table = $db->tableidentifier($tablename);
|
||||
|
||||
my $stmt = 'SELECT MAX(t1.id) FROM ' . $table . ' t1 LEFT JOIN ' . $table . ' t2' .
|
||||
' ON t1.collector = t2.collector and t2.acc_id > t1.acc_id'.
|
||||
' WHERE t2.collector IS NULL AND t1.collector = ?';
|
||||
my @params = ($collector);
|
||||
my $id = $xa_db->db_get_value($stmt,@params);
|
||||
|
||||
if (defined $id) {
|
||||
$stmt = 'DELETE FROM ' . $table . ' WHERE collector = ? AND id != ?';
|
||||
push(@params,$id);
|
||||
if ($xa_db->db_do($stmt,@params)) {
|
||||
rowsdeleted($db,$tablename,1,1,getlogger(__PACKAGE__));
|
||||
return 1;
|
||||
} else {
|
||||
rowsdeleted($db,$tablename,0,0,getlogger(__PACKAGE__));
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
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,4 @@
|
||||
|
||||
|
||||
TRUNK_FACILITY_ID 129990000C
|
||||
TRUNK_FACILITY_ID 220140000C
|
||||
@ -1,10 +1,11 @@
|
||||
|
||||
#skip_errors=0
|
||||
|
||||
providers_yml = providers.yml
|
||||
|
||||
export_cdr_multithreading = 1
|
||||
export_cdr_numofthreads = 2
|
||||
export_cdr_blocksize = 1000
|
||||
export_cdr_joins = { 'billing.voip_subscribers source_voip_subscribers' => { 'source_voip_subscribers.uuid' => 'accounting.cdr.source_user_id' } }, { 'billing.voip_subscribers destination_voip_subscribers' => { 'destination_voip_subscribers.uuid' => 'accounting.cdr.destination_user_id' } }, { 'billing.billing_zones_history source_carrier_bbz' => { 'source_carrier_bbz.id' => 'accounting.cdr.source_carrier_billing_zone_id' } }, { 'billing.billing_zones_history source_reseller_bbz' => { 'source_reseller_bbz.id' => 'accounting.cdr.source_reseller_billing_zone_id' } }, { 'billing.billing_zones_history source_customer_bbz' => { 'source_customer_bbz.id' => 'accounting.cdr.source_customer_billing_zone_id' } }, { 'billing.billing_zones_history destination_carrier_bbz' => { 'destination_carrier_bbz.id' => 'accounting.cdr.destination_carrier_billing_zone_id' } }, { 'billing.billing_zones_history destination_reseller_bbz' => { 'destination_reseller_bbz.id' => 'accounting.cdr.destination_reseller_billing_zone_id' } }, { 'billing.billing_zones_history destination_customer_bbz' => { 'destination_customer_bbz.id' => 'accounting.cdr.destination_customer_billing_zone_id' } }
|
||||
export_cdr_conditions = { 'accounting.cdr.export_status' => { '=' => '"unexported"' } }, { 'accounting.cdr.call_status' => { '=' => '"ok"' } }, { 'accounting.cdr.rating_status' => { '=' => '"ok"' } }
|
||||
export_cdr_limit = 300000
|
||||
export_cdr_stream = ama_simple
|
||||
|
||||
ama_filename_format = %1$s%2$03d.debug.ama
|
||||
|
||||
@ -1,14 +1,15 @@
|
||||
|
||||
#skip_errors=0
|
||||
|
||||
providers_yml = providers.yml
|
||||
|
||||
export_cdr_multithreading = 1
|
||||
export_cdr_numofthreads = 2
|
||||
export_cdr_blocksize = 1000
|
||||
#export_cdr_joins = { 'accounting.cdr_export_status_data esd' => { 'esd.cdr_id' => 'accounting.cdr.id' } }, { 'accounting.cdr_export_status es' => { 'es.id' => 'esd.status_id' } }
|
||||
#export_cdr_conditions = { 'es.type' => { '=' => '"default"' } }, { 'esd.export_status' => { '=' => '"unexported"' } }, { 'accounting.cdr.call_status' => { '=' => '"ok"' } }
|
||||
#, { 'accounting.cdr.rating_status' => { '=' => '"ok"' } }
|
||||
export_cdr_conditions = { 'accounting.cdr.call_status' => { '=' => '"ok"' } }
|
||||
export_cdr_stream = default
|
||||
export_cdr_stream = ama_simple
|
||||
#default
|
||||
#export_cdr_limit = 3500
|
||||
#300000
|
||||
|
||||
ama_filename_format = %1$s%2$03d.debug.ama
|
||||
|
||||
Loading…
Reference in new issue