MT#18663 MT#20893 row bulk processing framework WIP #7
+allow #* symbols in setoptionitems in Features_Define.cfg +prevent double execution+log properly +make sure to close all db connections before forking threads, not only project-specific ones +per-db dao folder for ipgallery migration project +start with ngcp db dao's: +contract_balances +contracts +refactor dao/sql processing class +params hash for method with long arg list +take out obselete methods +dedicated unique field option for "insert_record" +split SqlRecord class into SqlRecord&SqlProcessor +dependencies in Build.pl and debian/control +took out comments from debian/control +calcualte deltas of Subscriber_Define.cfg imports +"truncate subscriber import" task Change-Id: I8eab88319e6a4ba3d5b6ba935e50a56cc1763ad9changes/43/6943/8
Rene Krenn
10 years ago
parent
1c669a3a73
commit
b4f244a9f1
@ -0,0 +1,74 @@
|
||||
use Module::Build qw();
|
||||
my $builder = Module::Build->new(
|
||||
dist_name => 'NGCP::BulkProcessor',
|
||||
dist_abstract => 'Framework for parallel/distributed processing of record blocks',
|
||||
license => 'GPL_3',
|
||||
dist_author => 'Rene Krenn <rkrenn@sipwise.com>',
|
||||
dist_version_from => 'lib/NGCP/BulkProcessor/Globals.pm',
|
||||
perl_version_from => 'lib/NGCP/BulkProcessor/Globals.pm',
|
||||
requires => {
|
||||
'Archive::Zip' => 0,
|
||||
'OLE::Storage_Lite' => 0,
|
||||
'XML::LibXML::Reader' => 0,
|
||||
|
||||
'Email::MIME' => 0,
|
||||
'Email::MIME::Attachment::Stripper' => 0,
|
||||
|
||||
'URI::Find' => 0,
|
||||
'LWP::UserAgent' => 0,
|
||||
'HTTP::Request' => 0,
|
||||
|
||||
'DateTime' => 0,
|
||||
'Time::HiRes' => 0,
|
||||
'Time::Warp' => 0,
|
||||
'DateTime::TimeZone' => 0,
|
||||
'DateTime::Format::Strptime' => 0,
|
||||
'DateTime::Format::ISO8601' => 0,
|
||||
|
||||
'Tie::IxHash' => 0,
|
||||
|
||||
'YAML::Tiny' => 0,
|
||||
'Log::Log4Perl' => 0,
|
||||
'MIME::Base64' => 0,
|
||||
'MIME::Lite' => 0,
|
||||
'Net::SMTP' => 0,
|
||||
'JSON::XS' => 0,
|
||||
'Data::Dump' => 0,
|
||||
'YAML::XS' => 0,
|
||||
'XML::Dumper' => '0.81',
|
||||
'PHP::Serialization' => 0,
|
||||
#'Gearman::Worker' => 0,
|
||||
'Gearman::Client' => 0,
|
||||
#'Gearman::Task' => 0,
|
||||
'Digest::MD5' => 0,
|
||||
'Data::UUID' => 0,
|
||||
'Net::Address::IP::Local' => 0,
|
||||
'Date::Manip' => 0,
|
||||
'Date::Calc' => 0,
|
||||
#Sys::CpuAffinity
|
||||
'Marpa::R2' => 0,
|
||||
'Data::Dumper::Concise' => 0,
|
||||
'IO::Socket::SSL' => 0,
|
||||
'Mail::IMAPClient' => 0,
|
||||
'DBI' => '1.608',
|
||||
'DBD::CSV' => '0.26',
|
||||
'Locale::Recode' => 0,
|
||||
'Spreadsheet::ParseExcel' => 0,
|
||||
'Spreadsheet::ParseExcel::FmtUnicode' => 0,
|
||||
'Text::CSV_XS' => 0,
|
||||
'MIME::Parser' => 0,
|
||||
'HTML::Entities' => 0,
|
||||
'IO::Uncompress::Unzip' => 0,
|
||||
'DBD::mysql' => '4.014',
|
||||
#'DBD::Oracle' => '1.21',
|
||||
#'DBD::Pg' => '2.17.2',
|
||||
#'DBD::ODBC' => '1.50',
|
||||
'DBD::SQLite' => '1.29',
|
||||
},
|
||||
test_requires => {
|
||||
'Module::Runtime' => 0,
|
||||
'Test::Unit::Procedural' => 0,
|
||||
},
|
||||
add_to_cleanup => ['NGCP-BulkProcessor-*', 'Excel-Reader-*'],
|
||||
);
|
||||
$builder->create_build_script;
|
||||
@ -0,0 +1,40 @@
|
||||
=encoding UTF-8
|
||||
|
||||
Bulk-Processor
|
||||
|
||||
=head1 NAME
|
||||
|
||||
README - basic information for users prior to downloading
|
||||
|
||||
=head1 INSTALLATION
|
||||
|
||||
See L<http://www.cpan.org/modules/INSTALL.html>.
|
||||
|
||||
=head1 DEPENDENCIES
|
||||
|
||||
See distribution meta file.
|
||||
|
||||
=head1 LICENCE
|
||||
|
||||
This software is Copyright © 2016 by Sipwise GmbH, Austria.
|
||||
|
||||
This program is free software; you can redistribute it
|
||||
and/or modify it under the terms of the GNU General Public
|
||||
License as published by the Free Software Foundation; either
|
||||
version 3 of the License, or (at your option) any later
|
||||
version.
|
||||
|
||||
This program is distributed in the hope that it will be
|
||||
useful, but WITHOUT ANY WARRANTY; without even the implied
|
||||
warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
|
||||
PURPOSE. See the GNU General Public License for more
|
||||
details.
|
||||
|
||||
You should have received a copy of the GNU General Public
|
||||
License along with this package; if not, write to the Free
|
||||
Software Foundation, Inc., 51 Franklin St, Fifth Floor,
|
||||
Boston, MA 02110-1301 USA
|
||||
|
||||
On Debian systems, the full text of the GNU General Public
|
||||
License version 3 can be found in the file
|
||||
F</usr/share/common-licenses/GPL-3>.
|
||||
@ -0,0 +1,110 @@
|
||||
package NGCP::BulkProcessor::Dao::Trunk::billing::billing_mappings;
|
||||
use strict;
|
||||
|
||||
## no critic
|
||||
|
||||
#use File::Basename;
|
||||
#use Cwd;
|
||||
#use lib Cwd::abs_path(File::Basename::dirname(__FILE__) . '/../../../');
|
||||
|
||||
use NGCP::BulkProcessor::ConnectorPool qw(
|
||||
get_billing_db
|
||||
|
||||
);
|
||||
|
||||
use NGCP::BulkProcessor::SqlProcessor qw(
|
||||
checktableinfo
|
||||
insert_record
|
||||
);
|
||||
use NGCP::BulkProcessor::SqlRecord qw();
|
||||
|
||||
require Exporter;
|
||||
our @ISA = qw(Exporter NGCP::BulkProcessor::SqlRecord);
|
||||
our @EXPORT_OK = qw(
|
||||
gettablename
|
||||
check_table
|
||||
insert_row
|
||||
);
|
||||
|
||||
#my $logger = getlogger(__PACKAGE__);
|
||||
|
||||
my $tablename = 'billing_mappings';
|
||||
my $get_db = \&get_billing_db;
|
||||
|
||||
my $expected_fieldnames = [
|
||||
'id',
|
||||
'start_date',
|
||||
'end_date',
|
||||
'billing_profile_id',
|
||||
'contract_id',
|
||||
'product_id',
|
||||
'network_id',
|
||||
];
|
||||
|
||||
my $indexes = {};
|
||||
# 'balance_interval' => [ 'contract_id','start','end' ],
|
||||
# 'invoice_idx' => [ 'invoice_id' ],
|
||||
#};
|
||||
|
||||
my $insert_unique_fields = []; #[ 'contract_id','start','end' ];
|
||||
|
||||
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 insert_row {
|
||||
|
||||
my ($data,$insert_ignore) = @_;
|
||||
check_table();
|
||||
return insert_record($get_db,$tablename,$data,$insert_ignore,$unique_fields) = @_;
|
||||
|
||||
}
|
||||
|
||||
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,
|
||||
$tablename,
|
||||
$expected_fieldnames,
|
||||
$indexes);
|
||||
|
||||
}
|
||||
|
||||
1;
|
||||
@ -0,0 +1,122 @@
|
||||
package NGCP::BulkProcessor::Dao::Trunk::billing::contracts;
|
||||
use strict;
|
||||
|
||||
## no critic
|
||||
|
||||
#use File::Basename;
|
||||
#use Cwd;
|
||||
#use lib Cwd::abs_path(File::Basename::dirname(__FILE__) . '/../../../');
|
||||
|
||||
use NGCP::BulkProcessor::ConnectorPool qw(
|
||||
get_billing_db
|
||||
|
||||
);
|
||||
|
||||
use NGCP::BulkProcessor::SqlProcessor qw(
|
||||
checktableinfo
|
||||
insert_record
|
||||
);
|
||||
use NGCP::BulkProcessor::SqlRecord qw();
|
||||
|
||||
require Exporter;
|
||||
our @ISA = qw(Exporter NGCP::BulkProcessor::SqlRecord);
|
||||
our @EXPORT_OK = qw(
|
||||
gettablename
|
||||
check_table
|
||||
insert_row
|
||||
);
|
||||
|
||||
#my $logger = getlogger(__PACKAGE__);
|
||||
|
||||
my $tablename = 'contracts';
|
||||
my $get_db = \&get_billing_db;
|
||||
|
||||
my $expected_fieldnames = [
|
||||
'id',
|
||||
'customer_id',
|
||||
'contact_id',
|
||||
'order_id',
|
||||
'profile_package_id',
|
||||
'status',
|
||||
'external_id',
|
||||
'modify_timestamp',
|
||||
'create_timestamp',
|
||||
'activate_timestamp',
|
||||
'terminate_timestamp',
|
||||
'max_subscribers',
|
||||
'send_invoice',
|
||||
'subscriber_email_template_id',
|
||||
'passreset_email_template_id',
|
||||
'invoice_email_template_id',
|
||||
'invoice_template_id',
|
||||
'vat_rate',
|
||||
'add_vat',
|
||||
];
|
||||
|
||||
my $indexes = {};
|
||||
# 'balance_interval' => [ 'contract_id','start','end' ],
|
||||
# 'invoice_idx' => [ 'invoice_id' ],
|
||||
#};
|
||||
|
||||
my $insert_unique_fields = []; #[ 'contract_id','start','end' ];
|
||||
|
||||
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 insert_row {
|
||||
|
||||
my ($data,$insert_ignore) = @_;
|
||||
check_table();
|
||||
return insert_record($get_db,$tablename,$data,$insert_ignore,$unique_fields) = @_;
|
||||
|
||||
}
|
||||
|
||||
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,
|
||||
$tablename,
|
||||
$expected_fieldnames,
|
||||
$indexes);
|
||||
|
||||
}
|
||||
|
||||
1;
|
||||
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
Loading…
Reference in new issue