dbi2mongo.pl

   1 #!/usr/bin/perl -w
   2
   3 sub BEGIN {
   4 #$ENV{DBI_AUTOPROXY}='dbi:Gofer:transport=stream;url=ssh:dpavlin@koha.ffzg.hr';
   5 }
   6
   7 use strict;
   8 use DBI;
   9 use MongoDB;
  10 use Data::Dump qw/dump/;
  11 use Getopt::Long;
  12
  13 my $debug = 0;
  14 my $drop = 0;
  15 my $limit = 10000;
  16
  17 GetOptions(
  18         'debug!'  => \$debug,
  19         'drop!'   => \$drop,
  20         'limit=i' => \$limit,
  21 ) || die $!;
  22
  23 our ( $dbi, $user, $password ) = ( "DBI:mysql:database=test" );
  24 our ( $database, $collection ) = ( '', '' );
  25 our ( $table,  $pk ) = ( 'biblio'      => 'biblionumber' );
  26 our ( $table2, $fk ) = ( 'biblioitems' => 'biblionumber' );
  27
  28 my $config = shift @ARGV || die "usage: $0 config.pl\n";
  29 require $config;
  30
  31 warn "# $dbi $user -> $database $collection $table.$pk<->$table2.$fk\n";
  32
  33 my $conn = MongoDB::Connection->new;
  34 my $db   = $conn->get_database( $database );
  35 my $coll = $db->get_collection( $collection );
  36 my $dbh  = DBI->connect($dbi,$user,$password, {
  37         RaiseError => 1,
  38 #       mysql_enable_utf8 => 1,
  39 });
  40
  41 $db->drop if $drop;
  42
  43 # db.items.find().sort({_id:-1}).limit(1);
  44 my $last = $coll->query()->sort({ '_id' => -1 })->limit(1)->next;
  45 warn dump( $last );
  46 my $last_id = $last->{_id} || 0;
  47
  48 print "import $table.$pk > $last_id from $dbi\n";
  49
  50 our $offset = 0;
  51 our $sth;
  52
  53 sub select_table {
  54
  55         $sth = $dbh->prepare(qq{
  56                 select
  57                         $pk as _id,
  58                         $table.*
  59                 from $table
  60                 where $pk > ?
  61                 order by $pk asc
  62                 limit $limit
  63                 offset $offset
  64         });
  65
  66         print STDERR " $table:$offset ";
  67         $sth->execute( $last_id );
  68 #       warn "# $table columns ",dump( $sth->{NAME} ) if $offset == 0;
  69         print STDERR " join ", $sth->rows, " rows ";
  70 }
  71
  72 our $row;
  73 sub fetch_row {
  74         $row = $sth->fetchrow_hashref();
  75         if ( ! $row && $sth->rows == $limit ) {
  76                 $offset += $limit;
  77                 select_table;
  78                 $row = $sth->fetchrow_hashref();
  79         }
  80         return $row;
  81 }
  82
  83 our $join_offset = 0;
  84 our $sth_join;
  85
  86 sub join_table {
  87         $sth_join = $dbh->prepare(qq{
  88                 select
  89                         $table2.*
  90                 from $table2
  91                 where $fk > ?
  92                 order by $fk asc
  93                 limit $limit
  94                 offset $join_offset
  95         });
  96         print STDERR " $table2:$join_offset ";
  97         $sth_join->execute( $last_id );
  98 #       warn "# $table2 columns ",dump( $sth_join->{NAME} ) if $join_offset = 0;
  99         print STDERR " join ",$sth_join->rows, " rows ";
 100 }
 101
 102 our $row_join;
 103 sub fetch_row_join {
 104         $row_join = $sth_join->fetchrow_hashref();
 105         if ( ! $row_join && $sth_join->rows == $limit ) {
 106                 $join_offset += $limit;
 107                 join_table;
 108                 $row_join = $sth_join->fetchrow_hashref();
 109         }
 110         return $row_join;
 111 }
 112
 113 sub guess_types {
 114         my $row = shift;
 115         map { $row->{$_} * 1 } grep { defined $row->{$_} && $row->{$_} =~ /^\d+$/ } keys %$row;
 116         return $row;
 117 }
 118
 119 select_table;
 120 join_table;
 121 fetch_row_join;
 122
 123 while (my $row = fetch_row() ) {
 124
 125         while ( $row_join && $row_join->{$fk} < $row->{$pk} ) {
 126                 fetch_row_join;
 127         }
 128
 129         while ( $row_join && $row_join->{$fk} == $row->{$pk} ) {
 130                 push @{ $row->{ $table2 } }, guess_types($row_join);
 131                 print STDERR "j";
 132                 fetch_row_join;
 133         }
 134
 135         $coll->insert( guess_types($row) );
 136         print STDERR ".";
 137
 138 }
 139
 140 print STDERR "\n";