bin/cdr-voipnow.import

   1 #!/usr/bin/perl
   2
   3 use strict;
   4 use vars qw( $DEBUG );
   5 use Date::Parse 'str2time';
   6 use Date::Format 'time2str';
   7 use FS::UID qw(adminsuidsetup dbh);
   8 use FS::Record qw(qsearchs);
   9 use FS::cdr;
  10 use DBI;
  11 use Getopt::Std;
  12
  13 my %opt;
  14 getopts('H:U:P:D:T:vs:e:', \%opt);
  15 my $user = shift or die &usage;
  16
  17 my $dsn = 'dbi:mysql';
  18 $dsn .= ":database=$opt{D}" if $opt{D};
  19 $dsn .= ":host=$opt{H}" if $opt{H};
  20
  21 my $mysql = DBI->connect($dsn, $opt{U}, $opt{P})
  22   or die $DBI::errstr;
  23
  24 adminsuidsetup $user;
  25 $FS::UID::AutoCommit = 0;
  26 my $fsdbh = FS::UID::dbh;
  27
  28 # don't use freesidestatus
  29
  30 my $start_id;
  31 if ( $opt{s} ) {
  32   $start_id = $opt{s};
  33 }
  34 else {
  35   my $last_cdr = qsearchs({
  36       'table'   => 'cdr',
  37       'hashref' => {},
  38       'extra_sql' => 'ORDER BY cdrid DESC LIMIT 1',
  39     });
  40   $start_id = $last_cdr ? $last_cdr->cdrid + 1: 1;
  41 }
  42 my $end_id = $opt{e};
  43 print "Selecting CDRs from $start_id to ".($end_id || 'end')."...\n";
  44
  45 my $table = $opt{T} || 'call_history';
  46 # spelled "disposion" in the table
  47 my @cols = ( qw(
  48   id extension_number flow channel partyid start answer duration disposion) );
  49 my $sql = 'SELECT '.join(',', @cols). " FROM $table WHERE id >= $start_id";
  50 $sql .= " AND id <= $end_id" if $end_id;
  51 $sql .= " ORDER BY id";
  52 my $sth = $mysql->prepare($sql);
  53 $sth->execute;
  54 print "Importing ".$sth->rows." records...\n";
  55
  56 my $cdr_batch = new FS::cdr_batch({
  57     'cdrbatch' => 'mysql-import-'. time2str('%Y/%m/%d-%T',time),
  58   });
  59 my $error = $cdr_batch->insert;
  60 die $error if $error;
  61 my $cdrbatchnum = $cdr_batch->cdrbatchnum;
  62 my $imported = 0;
  63 my $skipped = 0;
  64
  65 my $row;
  66 my ($first, $last);
  67 while ( $row = $sth->fetchrow_hashref ) {
  68   if ( $opt{s} # skip this check if the range doesn't overlap
  69       and qsearchs('cdr', { cdrid => $row->{id} } ) ) {
  70     $skipped++;
  71     print $row->{id} ." (skipped)\n" if $opt{v};
  72     next;
  73   }
  74   my $cdr = FS::cdr->new({
  75       cdrid         => $row->{id},
  76       channel       => $row->{channel},
  77       duration      => $row->{duration},
  78       billsec       => $row->{duration},
  79       disposition   => $row->{disposion},
  80       startdate     => str2time($row->{start}),
  81       answerdate    => str2time($row->{answer}),
  82       cdrbatchnum   => $cdrbatchnum,
  83     }
  84   );
  85   print $row->{id},"\n" if $opt{v};
  86   if ( $row->{flow} eq 'out' ) {
  87     $cdr->src($row->{'extension_number'});
  88     $cdr->dst($row->{'partyid'});
  89   }
  90   elsif ( $row->{flow} eq 'in' ) {
  91     $cdr->dst($row->{'extension_number'});
  92     $cdr->src($row->{'partyid'});
  93   }
  94   else {
  95     $fsdbh->rollback;
  96     die $row->{id} .": invalid flow value: '".$row->{flow}."'\n";
  97   }
  98   my $error = $cdr->insert;
  99   if($error) {
 100     $fsdbh->rollback;
 101     die $row->{id} . ": failed import: $error\n";
 102   }
 103   $first ||= $row->{id};
 104   $last = $row->{id};
 105   $imported++;
 106 }
 107 $fsdbh->commit or die $fsdbh->errstr;
 108 print "Done.\n";
 109 print "Imported $imported CDRs ($first - $last).\n" if $imported;
 110 print "Skipped $skipped duplicates.\n" if $skipped;
 111 $mysql->disconnect;
 112
 113 sub usage {
 114   "Usage: \n  cdr-voipnow.import\n\t[ -H host ]\n\t-D database\n\t-U user\n\t-P password\n\t[ -v ] [ -s start ] [ -e end ]\n\tfreesideuser\n";
 115 }
 116
 117 =head1 NAME
 118
 119 cdr-voipnow.import - Import call data records (CDRs) from a 4psa VoipNow system
 120
 121 =head1 SYNOPSIS
 122
 123   cdr-voipnow.import [ -H host ] -D database -U user -P password
 124     [ -v ] [ -s start ] [ -e end ] freesideuser
 125
 126 =head1 DESCRIPTION
 127
 128 Connects to a MySQL database and downloads CDRs from the "call_history" table.
 129 The "id" field maps to "cdrid" in Freeside.  Other than that, the following
 130 fields are imported: channel, duration, billsec, startdate, answerdate,
 131 disposition, src, dst.  src and dst are inferred from the "extension_number"
 132 and "partyid" fields, with the value of the "flow" field (in or out) deciding
 133 which is the source number and which is the destination.
 134
 135 Any import errors (except duplicates) will abort and roll back the
 136 transaction.
 137
 138 =head1 OPTIONS
 139
 140 -H, -D, -U, -P: parameters to connect to the database: host, database name
 141 user, password.  Required, except -H, which defaults to localhost.
 142
 143 -s: set the lowest CDR id to import.  By default, the script will find
 144 the highest existing cdrid and import all CDRs with ids greater than that.
 145 -s overrides this and turns on duplicate checking.
 146
 147 -e: set the highest CDR id to import.  By default, this is unlimited.
 148
 149 -v: report all CDR ids as they are imported.
 150
 151 =cut