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