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

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