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   # Since we're finding these records by ID range instead of flagging them
  87   # individually, we can't leave gaps.  So rollback and die on any errors
  88   # except for duplicate cdrid.
  89   if ( $row->{flow} eq 'out' ) {
  90     $cdr->src($row->{'extension_number'});
  91     $cdr->dst($row->{'partyid'});
  92   }
  93   elsif ( $row->{flow} eq 'in' ) {
  94     $cdr->dst($row->{'extension_number'});
  95     $cdr->src($row->{'partyid'});
  96   }
  97   else {
  98     $fsdbh->rollback;
  99     die $row->{id} .": invalid flow value: '".$row->{flow}."'\n";
 100   }
 101   my $error = $cdr->insert;
 102   if($error) {
 103     $fsdbh->rollback;
 104     die $row->{id} . ": failed import: $error\n";
 105   }
 106   $first ||= $row->{id};
 107   $last = $row->{id};
 108   $imported++;
 109 }
 110 $fsdbh->commit or die $fsdbh->errstr;
 111 print "Done.\n";
 112 print "Imported $imported CDRs ($first - $last).\n" if $imported;
 113 print "Skipped $skipped duplicates.\n" if $skipped;
 114 $mysql->disconnect;
 115
 116 sub usage {
 117   "Usage: \n  cdr-mysql.import\n\t[ -v ]\n\t[ -H host ]\n\t-D database\n\t-U user\n\t-P password\n\tfreesideuser\n";
 118 }
 119