koha/koha2riak-search.pl

   1 #!/usr/bin/perl -w
   2
   3 use strict;
   4 use DBI;
   5 use RiakSearch;
   6 use Data::Dump qw/dump/;
   7
   8
   9 my ( $riak_url, $table, $dbi ) = @ARGV;
  10
  11 $riak_url ||= 'http://10.60.0.92:8098';
  12 $table    ||= 'biblioitems';
  13 $dbi      ||= 'DBI:mysql:dbname=koha;host=10.60.0.10;port=3306';
  14
  15 my $batch_size = 1000;
  16
  17 my $dbh = DBI->connect($dbi,"","") || die $DBI::errstr;
  18 my $riak = RiakSearch->new( $riak_url );
  19 #$riak->{args} = 'w=2';
  20
  21
  22 sub riak_search_kv_hook {
  23         my $bucket = shift;
  24         $riak->request( 'PUT' => $bucket, { props => {
  25                 precommit => [ { mod => 'riak_search_kv_hook', fun => 'precommit' } ],
  26 #               precommit => [],
  27 #               last_write_wins => 'true',
  28                 }
  29         });
  30         warn "riak_search_kv_hook $bucket ", $riak->request( 'GET' => '/koha.marcxml' );
  31 }
  32
  33 riak_search_kv_hook 'koha.marcxml';
  34 riak_search_kv_hook "koha.$table";
  35
  36 my $offset = 0;
  37 while(1) {
  38
  39         my $limit = "LIMIT $batch_size OFFSET $offset";
  40         warn "SELECT * FROM $table $limit\n";
  41
  42     my $sth = $dbh->prepare(qq{ select * from $table $limit}) || die $dbh->errstr();
  43     $sth->execute || die $sth->errstr();
  44     my @pk = $dbh->primary_key( undef, undef, $table );
  45
  46     print "import ", $sth->rows, " rows from $table pk:",dump( @pk ),"...\n";
  47
  48     while (my $row = $sth->fetchrow_hashref() ) {
  49
  50         my $key = join('_', map { $row->{$_} } @pk);
  51                 my $biblionumber = $row->{biblionumber};
  52
  53         if ( my $marcxml = delete $row->{marcxml} ) {
  54                         $riak->request( 'PUT' => "/koha.marcxml/$key", $marcxml, {
  55                                 'Content-Type' => 'text/xml',
  56                                 'Link' => qq|</riak/koha.$table/$biblionumber>; riaktag="biblio"|,
  57                         } );
  58         }
  59
  60 #               warn "## $key ",dump($row);
  61
  62                 my $headers;
  63                 foreach my $reference ( qw(biblio biblioitem) ) {
  64                         my $col = $reference . 'number';
  65                         next if $key =~ m/$col/;
  66                         my $number = $row->{$col} || next;
  67                         push @{ $headers->{Link} }, qq|</riak/koha.$reference/$number>; riaktag="$reference"|;
  68                 }
  69
  70                 $riak->request( 'PUT' => "/koha.$table/$key", $row, $headers );
  71     }
  72
  73         $offset += $sth->rows;
  74         last if $sth->rows < $batch_size;
  75
  76 }
  77
  78 warn "END total_rows $offset\n";