bin/cdr-opensips.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::cdr;
   9 use DBI;
  10 use Getopt::Std;
  11
  12 my %opt;
  13 getopts('H:U:P:D:T:s:e:', \%opt);
  14 my $user = shift or die &usage;
  15
  16 my $dsn = 'dbi:mysql';
  17 $dsn .= ":database=$opt{D}" if $opt{D};
  18 $dsn .= ":host=$opt{H}" if $opt{H};
  19
  20 my $mysql = DBI->connect($dsn, $opt{U}, $opt{P})
  21   or die $DBI::errstr;
  22
  23 my ($start, $end) = ('', '');
  24 if ( $opt{s} ) {
  25   $start = str2time($opt{s}) or die "can't parse start date $opt{s}\n";
  26   $start = time2str('%Y-%m-%d', $start);
  27 }
  28 if ( $opt{e} ) {
  29   $end = str2time($opt{e}) or die "can't parse end date $opt{e}\n";
  30   $end = time2str('%Y-%m-%d', $end);
  31 }
  32
  33 adminsuidsetup $user;
  34
  35 my $fsdbh = FS::UID::dbh;
  36
  37 # check for existence of freesidestatus
  38 my $table = $opt{T} || 'acc';
  39 my $status = $mysql->selectall_arrayref("SHOW COLUMNS FROM $table WHERE Field = 'freesidestatus'");
  40 if( ! @$status ) {
  41   print "Adding freesidestatus column...\n";
  42   $mysql->do("ALTER TABLE $table ADD COLUMN freesidestatus varchar(32)")
  43     or die $mysql->errstr;
  44 }
  45 else {
  46   print "freesidestatus column present\n";
  47 }
  48
  49 my @cols = ( qw(
  50   id caller_id callee_id method from_tag to_tag callid sip_code sip_reason
  51   time )
  52 );
  53
  54 my $sql = 'SELECT '.join(',', @cols). " FROM $table".
  55   ' WHERE freesidestatus IS NULL' .
  56   ' AND sip_code = 200 ' . # only want successful calls
  57   ($start && " AND time >= '$start'") .
  58   ($end   && " AND time <  '$end'") .
  59   ' ORDER BY time'; # should ensure INVITE/ACK/BYE order
  60 my $sth = $mysql->prepare($sql);
  61 $sth->execute;
  62 print "Importing ".$sth->rows." records...\n";
  63
  64 my $cdr_batch = new FS::cdr_batch({
  65     'cdrbatch' => 'mysql-import-'. time2str('%Y/%m/%d-%T',time),
  66   });
  67 my $error = $cdr_batch->insert;
  68 die $error if $error;
  69 my $cdrbatchnum = $cdr_batch->cdrbatchnum;
  70 my $imports = 0;
  71 my $updates = 0;
  72
  73 my %cdrs;
  74 my $row;
  75 while ( $row = $sth->fetchrow_hashref ) {
  76   my ($callid) = $row->{'callid'};
  77   $callid =~ s/@.*//;
  78   if ( !$callid ) {
  79     warn $row->{'time'} . ": no callid, skipped.\n";
  80     next;
  81   }
  82   my ($src) = $row->{'caller_id'} =~ /^sip:(\d+)@/;
  83   my ($dst) = $row->{'callee_id'} =~ /^sip:(\d+)@/;
  84
  85   my $cdr = $cdrs{$callid};
  86   if ( !$cdr ) {
  87     $cdr = $cdrs{$callid} = FS::cdr->new ({
  88       uniqueid    => $callid,
  89       cdrbatchnum => $cdrbatchnum,
  90     });
  91   }
  92   my $date = str2time($row->{'time'});
  93   if ( $row->{'method'} eq 'INVITE' ) {
  94     $cdr->startdate($date);
  95     $cdr->src($src);
  96     $cdr->dst($dst);
  97   }
  98   elsif ( $row->{'method'} eq 'ACK' ) {
  99     $cdr->answerdate($date);
 100     next if !check_cdr($cdr, $src, $dst);
 101   }
 102   elsif ( $row->{'method'} eq 'BYE' ) {
 103     $cdr->enddate($date);
 104     next if !check_cdr($cdr, $src, $dst);
 105   }
 106   if ( $cdr->startdate and $cdr->answerdate and $cdr->enddate ) {
 107     $cdr->duration($cdr->enddate - $cdr->startdate);
 108     $cdr->billsec($cdr->enddate - $cdr->answerdate);
 109     my $error = $cdr->insert;
 110     if($error) {
 111       print "failed import: $error\n";
 112     }
 113     else {
 114       $imports++;
 115       if( $updates += $mysql->do("UPDATE $table SET freesidestatus = 'done'
 116           WHERE sip_code = 200 AND callid = ?",
 117           undef,
 118           $row->{'callid'}
 119         ) ) { #nothing
 120       }
 121       else {
 122         print "failed to set status: ".$mysql->errstr."\n";
 123       }
 124       delete $cdrs{$callid};
 125     }
 126   }
 127 }
 128 print "Done.\nImported $imports CDRs, marked $updates accounting events as done.\n";
 129 if ( keys(%cdrs) ) {
 130   print "Skipped ".scalar(keys(%cdrs))." incomplete calls.\n";
 131 }
 132 $mysql->disconnect;
 133
 134 sub usage {
 135   "Usage: \n  cdr-opensips.import\n\t[ -H host ]\n\t-D database\n\t-U user\n\t-P password\n\t[ -s start ] [ -e end ]\n\tfreesideuser\n";
 136 }
 137
 138 sub check_cdr {
 139   # Verify that these records belong to the same call.
 140   # BYE records sometimes have the caller/callee fields swapped.
 141   # We allow empty src/dst so as not to make noise about incomplete calls. If
 142   # this check fails, something is wrong with the source data.
 143   my ($cdr, $a, $b) = @_;
 144   if ( ( $cdr->src and $cdr->src ne $a and $cdr->src ne $b )
 145     or ( $cdr->dst and $cdr->dst ne $a and $cdr->dst ne $b ) ) {
 146     warn $cdr->uniqueid . ": src/dst mismatch, skipped.\n";
 147     return 0;
 148   }
 149   return 1;
 150 }