FS/FS/cdr/Import.pm

   1 package FS::cdr::Import;
   2
   3 use strict;
   4 use Date::Format 'time2str';
   5 use FS::UID qw(adminsuidsetup dbh);
   6 use FS::cdr;
   7 use DBI;
   8 use Getopt::Std;
   9
  10 use vars qw( $DEBUG );
  11 $DEBUG = 0;
  12
  13 =head1 NAME
  14
  15 FS::cdr::Import - CDR importing
  16
  17 =head1 SYNOPSIS
  18
  19   use FS::cdr::Import;
  20
  21   FS::cdr::Import->dbi_import(
  22     'dbd'         => 'Pg', #mysql, Sybase, etc.
  23     'table'       => 'TABLE_NAME',
  24     'primary_key' => 'BILLING_ID',
  25     'status_table' = > 'STATUS_TABLE_NAME', # if using a table rather than field in main table
  26     'column_map'  => { #freeside => remote_db
  27       'freeside_column' => 'remote_db_column',
  28       'freeside_column' => sub { my $row = shift; $row->{remote_db_column}; },
  29     },
  30     'batch_name' => 'batch_name', # cdr_batch name -import-date gets appended.
  31   );
  32
  33 =head1 DESCRIPTION
  34
  35 CDR importing
  36
  37 =head1 CLASS METHODS
  38
  39 =item dbi_import
  40
  41 =cut
  42
  43 sub dbi_import {
  44   my $class = shift;
  45   my %args = @_; #args are specifed by the script using this sub
  46
  47   my %opt; #opt is specified for each install / run of the script
  48   getopts('H:U:P:D:T:c:L:S:', \%opt);
  49
  50   my $user = shift(@ARGV) or die $class->cli_usage;
  51   my $database = $opt{D} || $args{database};
  52   my $table = $opt{T} || $args{table};
  53   my $pkey = $args{primary_key};
  54   my $pkey_info = $args{primary_key_info} ? $args{primary_key_info} : 'BIGINT';
  55   my $status_table = $opt{S} || $args{status_table};
  56   my $dbd_type = $args{'dbd'} ? $args{'dbd'} : 'Pg';
  57   my $status_column = $args{status_column} ? $args{status_column} : 'freesidestatus';
  58   my $status_column_info = $args{status_column_info} ? $args{status_column} : 'VARCHAR(32)';
  59
  60   my $queries = get_queries({
  61     'dbd'                 => $dbd_type,
  62     'table'               => $table,
  63     'status_column'       => $status_column,
  64     'status_column_info'  => $status_column_info,
  65     'status_table'        => $status_table,
  66     'primary_key'         => $pkey,
  67     'primary_key_info'    => $pkey_info,
  68   });
  69
  70   my $dsn = 'dbi:'. $dbd_type;
  71   $dsn .= $queries->{connect_type} . "=$opt{H}";
  72   $dsn .= ";database=$database" if $database;
  73
  74   my $dbi = DBI->connect($dsn, $opt{U}, $opt{P})
  75     or die $DBI::errstr;
  76
  77   adminsuidsetup $user;
  78
  79   ## check for status table if using. if not there create it.
  80   if ($status_table) {
  81     my $status = $dbi->selectall_arrayref( $queries->{check_statustable} );
  82     if( ! @$status ) {
  83       print "Adding status table $status_table ...\n";
  84       $dbi->do( $queries->{create_statustable} )
  85         or die $dbi->errstr;
  86     }
  87   }
  88   ## check for column freeside status if not using status table and create it if not there.
  89   else {
  90     my $status = $dbi->selectall_arrayref( $queries->{check_statuscolumn} );
  91     if( ! @$status ) {
  92       print "Adding $status_column column...\n";
  93       $dbi->do( $queries->{create_statuscolumn} )
  94         or die $dbi->errstr;
  95     }
  96   }
  97
  98   #my @cols = values %{ $args{column_map} };
  99   my $sql = "SELECT $table.* FROM $table "; # join(',', @cols). " FROM $table ".
 100   $sql .=  "LEFT JOIN $status_table ON ( $table.$pkey = $status_table.$pkey ) "
 101     if $status_table;
 102   $sql .= "WHERE  $status_column IS NULL ";
 103
 104   #$sql .= ' LIMIT '. $opt{L} if $opt{L};
 105   my $sth = $dbi->prepare($sql);
 106   $sth->execute or die $sth->errstr. " executing $sql";
 107
 108   my $cdr_batch = new FS::cdr_batch({
 109       'cdrbatch' => $args{batch_name} . '-import-'. time2str('%Y/%m/%d-%T',time),
 110     });
 111   my $error = $cdr_batch->insert;
 112   die $error if $error;
 113   my $cdrbatchnum = $cdr_batch->cdrbatchnum;
 114   my $imported = 0;
 115
 116   my $row;
 117   while ( $row = $sth->fetchrow_hashref ) {
 118
 119     my %hash = ( 'cdrbatchnum' => $cdrbatchnum );
 120     foreach my $field ( keys %{ $args{column_map} } ) {
 121       my $col_or_coderef = $args{column_map}->{$field};
 122       if ( ref($col_or_coderef) eq 'CODE' ) {
 123         $hash{$field} = &{ $col_or_coderef }( $row );
 124       } else {
 125         $hash{$field} = $row->{ $col_or_coderef };
 126       }
 127       $hash{$field} = '' if $hash{$field} =~ /^\s+$/; #IVR (MSSQL?) bs
 128     }
 129
 130     my $cdr = FS::cdr->new(\%hash);
 131
 132     $cdr->cdrtypenum($opt{c}) if $opt{c};
 133
 134     my $pkey_value = $row->{$pkey};
 135
 136     #print "$pkey_value\n" if $opt{v};
 137     my $error = $cdr->insert;
 138
 139     if ($error) {
 140
 141       #die "$pkey_value: failed import: $error\n";
 142       print "$pkey_value: failed import: $error\n";
 143
 144     } else {
 145
 146       $imported++;
 147
 148       my $st_sql;
 149       if ( $status_table ) {
 150
 151         $st_sql =
 152           'INSERT INTO '. $status_table. " ( $pkey, $status_column ) ".
 153             " VALUES ( ?, 'done' )";
 154
 155       } else {
 156
 157         $st_sql = "UPDATE $table SET $status_column = 'done' WHERE $pkey = ?";
 158
 159       }
 160
 161       my $updated = $dbi->do($st_sql, undef, $pkey_value );
 162       #$updates += $updated;
 163       die "failed to set status: ".$dbi->errstr."\n" unless $updated;
 164
 165     }
 166
 167     if ( $opt{L} && $imported >= $opt{L} ) {
 168       $sth->finish;
 169       last;
 170     }
 171
 172   }
 173   print "Done.\n";
 174   print "Imported $imported CDRs.\n" if $imported;
 175
 176   $dbi->disconnect;
 177
 178 }
 179
 180 sub cli_usage {
 181   #"Usage: \n  $0\n\t[ -H hostname ]\n\t-D database\n\t-U user\n\t-P password\n\tfreesideuser\n";
 182   #"Usage: \n  $0\n\t-H hostname\n\t-D database\n\t-U user\n\t-P password\n\t[ -c cdrtypenum ]\n\tfreesideuser\n";
 183   "Usage: \n  $0\n\t-H hostname\n\t[ -D database ]\n\t-U user\n\t-P password\n\t[ -c cdrtypenum ]\n\t[ -L num_cdrs_limit ]\n\t[ -T table ]\n\t[ -S status table ]\n\tfreesideuser\n";
 184 }
 185
 186 sub get_queries {
 187   #my ($dbd, $table, $column, $column_create_info, $status_table, $primary_key, $primary_key_info) = @_;
 188   my $info = shift;
 189
 190   #do we want to add more types? or add as we go?
 191   my %dbi_connect_types = (
 192     'Sybase'  => ':server',
 193     'Pg'      => ':host',
 194   );
 195
 196   #Check for freeside status table Sybase has not been tested
 197   my %dbi_check_statustable = (
 198     'Sybase'  => "SELECT * FROM sysobjects WHERE name = $info->{status_table}",
 199     'Pg'      => "SELECT * FROM information_schema.columns WHERE table_schema = 'public' AND table_name = '$info->{status_table}' AND column_name = '$info->{status_column}'",
 200   );
 201
 202   #Check for freeside status table Sybase has not been tested
 203   my %dbi_create_statustable = (
 204     'Sybase'  => "CREATE TABLE $info->{status_table} ( $info->{primary_key} $info->{primary_key_info}, $info->{status_column} $info->{status_column_info} )",
 205     'Pg'      => "CREATE TABLE $info->{status_table} ( $info->{primary_key} $info->{primary_key_info}, $info->{status_column} $info->{status_column_info} )",
 206   );
 207
 208   #Check for freeside status column Sybase has not been tested
 209   my %dbi_check_statuscolumn = (
 210     'Sybase'  => "SELECT syscolumns.name FROM sysobjects
 211                   JOIN syscolumns ON sysobjects.id = syscolumns.id
 212                   WHERE sysobjects.name LIKE '$info->{table}' AND syscolumns.name = $info->{status_column}",
 213     'Pg'      => "SELECT * FROM information_schema.columns WHERE table_schema = 'public' AND table_name = '$info->{table}' AND column_name = '$info->{status_column}' ",
 214   );
 215
 216     #Check for freeside status column Sybase has not been tested
 217   my %dbi_create_statuscolumn = (
 218     'Sybase'  => "ALTER TABLE $info->{table} ADD $info->{status_column} $info->{status_column_info}",
 219     'Pg'      => "ALTER TABLE $info->{table} ADD COLUMN $info->{status_column} $info->{status_column_info}",
 220   );
 221
 222   my $queries = {
 223     'connect_type'         =>  $dbi_connect_types{$info->{dbd}},
 224     'check_statustable'    =>  $dbi_check_statustable{$info->{dbd}},
 225     'create_statustable'   =>  $dbi_create_statustable{$info->{dbd}},
 226     'check_statuscolumn'   =>  $dbi_check_statuscolumn{$info->{dbd}},
 227     'create_statuscolumn'  =>  $dbi_create_statuscolumn{$info->{dbd}},
 228   };
 229
 230   return $queries;
 231 }
 232
 233 =head1 BUGS
 234
 235 This has only been test with Pg -> postgresql databases
 236
 237 Sparse documentation.
 238
 239 =head1 SEE ALSO
 240
 241 L<FS::cdr>
 242
 243 =cut
 244
 245 1;