Open-ILS/src/extras/import/parallel_pg_loader.pl

   1 #!/usr/bin/perl
   2 use strict;
   3
   4 use lib '/openils/lib/perl5/';
   5
   6 use OpenSRF::System;
   7 use OpenSRF::EX qw/:try/;
   8 use OpenSRF::Utils::SettingsClient;
   9 use OpenILS::Utils::Fieldmapper;
  10 use OpenSRF::Utils::JSON;
  11 use FileHandle;
  12
  13 use Time::HiRes qw/time/;
  14 use Getopt::Long;
  15
  16 my @files;
  17 my ($config, $output, @auto, @order, @wipe) =
  18         ('/openils/conf/opensrf_core.xml', 'pg_loader-output');
  19
  20 GetOptions(
  21         'config=s'      => \$config,
  22         'output=s'      => \$output,
  23         'wipe=s'        => \@wipe,
  24         'autoprimary=s' => \@auto,
  25         'order=s'       => \@order,
  26 );
  27
  28 my $pwd = `pwd`;
  29 chop($pwd);
  30
  31 my %lineset;
  32 my %fieldcache;
  33
  34 OpenSRF::System->bootstrap_client( config_file => $config );
  35 Fieldmapper->import(IDL => OpenSRF::Utils::SettingsClient->new->config_value("IDL"));
  36
  37 my $main_out = FileHandle->new(">$output.sql") if ($output);
  38
  39 binmode($main_out,'utf8');
  40
  41 $main_out->print("SET CLIENT_ENCODING TO 'UNICODE';\n\n");
  42 $main_out->print("BEGIN;\n\n");
  43
  44 my %out_files;
  45 my %out_headers;
  46 for my $h (@order) {
  47         $out_files{$h} = FileHandle->new(">$output.$h.sql");
  48         binmode($out_files{$h},'utf8');
  49 }
  50
  51 my $count = 0;
  52 my $starttime = time;
  53 while ( my $rec = <> ) {
  54         next unless ($rec);
  55
  56         my $row;
  57         try {
  58                 $row = OpenSRF::Utils::JSON->JSON2perl($rec);
  59         } catch Error with {
  60                 my $e = shift;
  61                 warn "\n\n !!! Error : $e \n\n at or around line $count\n";
  62         };
  63         next unless ($row);
  64
  65         my $class = $row->class_name;
  66         my $hint = $row->json_hint;
  67
  68         next unless ( grep /$hint/, @order );
  69
  70         if (!$fieldcache{$hint}) {
  71                 my @cols = $row->real_fields;
  72                 if (grep { $_ eq $hint} @auto) {
  73                         @cols = grep { $_ ne $class->Identity } @cols;
  74                 }
  75
  76                 $fieldcache{$hint} =
  77                         { table => $class->Table,
  78                           sequence => $class->Sequence,
  79                           pkey => $class->Identity,
  80                           fields => \@cols,
  81                         };
  82
  83                 my $fields = join(',', @{ $fieldcache{$hint}{fields} });
  84                 $main_out->print( "DELETE FROM $fieldcache{$hint}{table};\n" ) if (grep {$_ eq $hint } @wipe);
  85                 $main_out->print( "COPY $fieldcache{$hint}{table} ($fields) FROM '$pwd/$output.$hint.sql';\n" );
  86
  87         }
  88
  89         my $line = [map { $row->$_ } @{ $fieldcache{$hint}{fields} }];
  90         my @data;
  91         my $x = 0;
  92         for my $d (@$line) {
  93                 if (!defined($d)) {
  94                         $d = '\N';
  95                 } else {
  96                         $d =~ s/\f/\\f/gos;
  97                         $d =~ s/\n/\\n/gos;
  98                         $d =~ s/\r/\\r/gos;
  99                         $d =~ s/\t/\\t/gos;
 100                         $d =~ s/\\/\\\\/gos;
 101                 }
 102                 if ($hint eq 'bre' and $fieldcache{$hint}{fields}[$x] eq 'quality') {
 103                         $d = int($d);
 104                 }
 105                 push @data, $d;
 106                 $x++;
 107         }
 108         $out_files{$hint}->print( join("\t", @data)."\n" );
 109
 110         if (!($count % 500)) {
 111                 print STDERR "\r$count\t". $count / (time - $starttime);
 112         }
 113
 114         $count++;
 115 }
 116
 117 $main_out->print("-- COMMIT;\n\n");
 118 $main_out->close;