Open-ILS/src/support-scripts/marc_stream_importer.pl

   1 #!/usr/bin/perl
   2 # Copyright (C) 2008-2010 Equinox Software, Inc.
   3 # Author: Bill Erickson <erickson@esilibrary.com>
   4 #
   5 # This program is free software; you can redistribute it and/or
   6 # modify it under the terms of the GNU General Public License
   7 # as published by the Free Software Foundation; either version 2
   8 # of the License, or (at your option) any later version.
   9 #
  10 # This program is distributed in the hope that it will be useful,
  11 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  12 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13 # GNU General Public License for more details.
  14
  15
  16 use strict; use warnings;
  17 use Net::Server::PreFork;
  18 use base qw/Net::Server::PreFork/;
  19 use MARC::Record;
  20 use MARC::Batch;
  21 use MARC::File::XML;
  22 use MARC::File::USMARC;
  23
  24 use Data::Dumper;
  25 use File::Basename qw/fileparse/;
  26 use File::Temp;
  27 use Getopt::Long qw(:DEFAULT GetOptionsFromArray);
  28 use Pod::Usage;
  29
  30 use OpenSRF::Utils::Logger qw/$logger/;
  31 use OpenSRF::AppSession;
  32 use OpenSRF::EX qw/:try/;
  33 use OpenILS::Utils::Cronscript;
  34 use OpenSRF::Transport::PeerHandle;
  35 require 'oils_header.pl';
  36 use vars qw/$apputils/;
  37
  38 my $vl_ses;
  39
  40 my $debug = 0;
  41
  42 my %defaults = (
  43     'buffsize=i'    => 4096,
  44     'merge-profile=i' => 0,
  45     'source=i'      => 1,
  46 #    'osrf-config=s' => '/openils/conf/opensrf_core.xml',
  47     'user=s'        => 'admin',
  48     'password=s'    => '',
  49     'tempdir=s'     => '',
  50     'nolockfile'    => 1,
  51     'queue=i'       => 1,
  52     'noqueue'       => 0,
  53     'wait=i'        => 5,
  54     'import-by-queue' => 0
  55 );
  56
  57 $OpenILS::Utils::Cronscript::debug=1 if $debug;
  58 $Getopt::Long::debug=1 if $debug > 1;
  59 my $o = OpenILS::Utils::Cronscript->new(\%defaults);
  60
  61 my @script_args = ();
  62
  63 if (grep {$_ eq '--'} @ARGV) {
  64     print "Splitting options into groups\n" if $debug;
  65     while (@ARGV) {
  66         $_ = shift @ARGV;
  67         $_ eq '--' and last;    # stop at the first --
  68         push @script_args, $_;
  69     }
  70 } else {
  71     @script_args = @ARGV;
  72     @ARGV = ();
  73 }
  74
  75 print "Calling MyGetOptions ",
  76     (@script_args ? "with options: " . join(' ', @script_args) : 'without options from command line'),
  77     "\n" if $debug;
  78
  79 my $real_opts = $o->MyGetOptions(\@script_args);
  80 $o->bootstrap;
  81 # GetOptionsFromArray(\@script_args, \%defaults, %defaults); # similar to
  82
  83 $real_opts->{tempdir} ||= tempdir_setting();    # This doesn't go in defaults because it reads config, must come after bootstrap
  84
  85 my $bufsize       = $real_opts->{buffsize};
  86 my $bib_source    = $real_opts->{source};
  87 my $osrf_config   = $real_opts->{'osrf-config'};
  88 my $oils_username = $real_opts->{user};
  89 my $oils_password = $real_opts->{password};
  90 my $help          = $real_opts->{help};
  91 my $merge_profile = $real_opts->{'merge-profile'};
  92 my $queue_id      = $real_opts->{queue};
  93 my $tempdir       = $real_opts->{tempdir};
  94 my $import_by_queue  = $real_opts->{'import-by-queue'};
  95    $debug        += $real_opts->{debug};
  96
  97 foreach (keys %$real_opts) {
  98     print("real_opt->{$_} = ", $real_opts->{$_}, "\n") if $real_opts->{debug} or $debug;
  99 }
 100 my $wait_time     = $real_opts->{wait};
 101 my $authtoken     = '';
 102
 103 # DEFAULTS for Net::Server
 104 my $filename   = fileparse($0, '.pl');
 105 my $conf_file  = (-r "$filename.conf") ? "$filename.conf" : undef;
 106 # $conf_file is the Net::Server config for THIS script (not EG), if it exists and is readable
 107
 108
 109 # FEEDBACK
 110
 111 pod2usage(1) if $help;
 112 unless ($oils_password) {
 113     print STDERR "\nERROR: password option required for session login\n\n";
 114     # pod2usage(1);
 115 }
 116
 117 print Dumper($o) if $debug;
 118
 119 if ($debug) {
 120     foreach my $ref (qw/bufsize bib_source osrf_config oils_username oils_password help conf_file debug/) {
 121         no strict 'refs';
 122         printf "%16s => %s\n", $ref, (eval("\$$ref") || '');
 123     }
 124 }
 125
 126 print warning();
 127 print Dumper($real_opts);
 128
 129 # SUBS
 130
 131 sub tempdir_setting {
 132     my $ret = $apputils->simplereq( qw# opensrf.settings opensrf.settings.xpath.get
 133         /opensrf/default/apps/open-ils.vandelay/app_settings/databases/importer # );
 134     return $ret->[0] || '/tmp';
 135 }
 136
 137 sub warning {
 138     return <<WARNING;
 139
 140 WARNING:  This script provides no security layer.  Any client that has
 141 access to the server+port can inject MARC records into the system.
 142
 143 WARNING
 144 }
 145
 146 sub xml_import {
 147     return $apputils->simplereq(
 148         'open-ils.cat',
 149         'open-ils.cat.biblio.record.xml.import',
 150         @_
 151     );
 152 }
 153
 154 sub old_process_batch_data {
 155     my $data = shift or $logger->error("process_batch_data called without any data");
 156     $data or return;
 157
 158     my $handle;
 159     open $handle, '<', \$data;
 160     my $batch = MARC::Batch->new('USMARC', $handle);
 161     $batch->strict_off;
 162
 163     my $index = 0;
 164     my $imported = 0;
 165     my $failed = 0;
 166
 167     while (1) {
 168         my $rec;
 169         $index++;
 170
 171         eval { $rec = $batch->next; };
 172
 173         if ($@) {
 174             $logger->error("Failed parsing MARC record $index");
 175             $failed++;
 176             next;
 177         }
 178         last unless $rec;   # The only way out
 179
 180         my $resp = xml_import($authtoken, $rec->as_xml_record, $bib_source);
 181
 182         # has the session timed out?
 183         if (oils_event_equals($resp, 'NO_SESSION')) {
 184             new_auth_token();
 185             $resp = xml_import($authtoken, $rec->as_xml_record, $bib_source);   # try again w/ new token
 186         }
 187         oils_event_die($resp);
 188         $imported++;
 189     }
 190
 191     return ($imported, $failed);
 192 }
 193
 194 sub process_spool { # filename
 195
 196     my $marcfile = shift;
 197     my @rec_ids;
 198
 199     if($import_by_queue) {
 200
 201         # don't collect the record IDs, just spool the queue
 202
 203         $apputils->simplereq(
 204             'open-ils.vandelay',
 205             'open-ils.vandelay.bib.process_spool',
 206             $authtoken,
 207             undef,
 208             $queue_id,
 209             'import',
 210             $marcfile,
 211             $bib_source
 212         );
 213
 214     } else {
 215
 216         # collect the newly queued record IDs for processing
 217
 218         my $req = $vl_ses->request(
 219             'open-ils.vandelay.bib.process_spool.stream_results',
 220             $authtoken,
 221             undef, # cache key not needed
 222             $queue_id,
 223             'import',
 224             $marcfile,
 225             $bib_source
 226         );
 227
 228         while(my $resp = $req->recv) {
 229
 230             if($req->failed) {
 231                 $logger->error("Error spooling MARC data: $resp");
 232
 233             } elsif($resp->content) {
 234                 push(@rec_ids, $resp->content);
 235             }
 236         }
 237     }
 238
 239     return \@rec_ids;
 240 }
 241
 242 sub bib_queue_import {
 243     my $rec_ids = shift;
 244     my $extra = {auto_overlay_exact => 1};
 245     $extra->{merge_profile} = $merge_profile if $merge_profile;
 246
 247     my $req;
 248     my @cleanup_recs;
 249
 250     if($import_by_queue) {
 251         # import by queue
 252
 253         $req = $vl_ses->request(
 254             'open-ils.vandelay.bib_queue.import',
 255             $authtoken,
 256             $queue_id,
 257             $extra
 258         );
 259
 260     } else {
 261         # import explicit record IDs
 262
 263         $req = $vl_ses->request(
 264             'open-ils.vandelay.bib_record.list.import',
 265             $authtoken,
 266             $rec_ids,
 267             $extra
 268         );
 269     }
 270
 271     # collect the successfully imported vandelay records
 272     my $failed = 0;
 273     while(my $resp = $req->recv) {
 274          if($req->failed) {
 275             $logger->error("Error importing MARC data: $resp");
 276
 277         } elsif(my $data = $resp->content) {
 278
 279             if($data->{err_event}) {
 280
 281                 $logger->error(Dumper($data->{err_event}));
 282                 $failed++;
 283
 284             } else {
 285                 push(@cleanup_recs, $data->{imported}) if $data->{imported};
 286             }
 287         }
 288     }
 289
 290     # clean up the successfully imported vandelay records to prevent queue bloat
 291     my $pcrud = OpenSRF::AppSession->create('open-ils.pcrud');
 292     $pcrud->connect;
 293     $pcrud->request('open-ils.pcrud.transaction.begin', $authtoken)->recv;
 294     my $err;
 295
 296     foreach (@cleanup_recs) {
 297
 298         try {
 299
 300             $pcrud->request('open-ils.pcrud.delete.vqbr', $authtoken, $_)->recv;
 301
 302         } catch Error with {
 303             $err = shift;
 304             $logger->error("Error deleteing queued bib record $_: $err");
 305         };
 306     }
 307
 308     $pcrud->request('open-ils.pcrud.transaction.commit', $authtoken)->recv unless $err;
 309     $pcrud->disconnect;
 310
 311     $logger->info("imported queued vandelay records: @cleanup_recs");
 312     return (scalar(@cleanup_recs), $failed);
 313 }
 314
 315 sub process_batch_data {
 316     my $data = shift or $logger->error("process_batch_data called without any data");
 317     $data or return;
 318
 319     $vl_ses = OpenSRF::AppSession->create('open-ils.vandelay');
 320
 321     my ($handle, $tempfile) = File::Temp->tempfile("$0_XXXX", DIR => $tempdir) or die "Cannot write tempfile in $tempdir";
 322     print $handle $data;
 323     close $handle;
 324
 325     $logger->info("Calling process_spool on tempfile $tempfile (queue: $queue_id; source: $bib_source)");
 326     my $rec_ids = process_spool($tempfile);
 327
 328     if (oils_event_equals($rec_ids, 'NO_SESSION')) {  # has the session timed out?
 329         new_auth_token();
 330         $rec_ids = process_spool($tempfile);                # try again w/ new token
 331     }
 332
 333     my ($imported, $failed) = bib_queue_import($rec_ids);
 334
 335     if (oils_event_equals($imported, 'NO_SESSION')) {  # has the session timed out?
 336         new_auth_token();
 337         ($imported, $failed) = bib_queue_import();                # try again w/ new token
 338     }
 339
 340     oils_event_die($imported);
 341
 342     return ($imported, $failed);
 343 }
 344
 345 sub process_request {   # The core Net::Server method
 346     my $self = shift;
 347     my $client = $self->{server}->{client};
 348
 349     $logger->info("stream parser received contact from $client");
 350
 351     my $ph = OpenSRF::Transport::PeerHandle->retrieve;
 352     if(!$ph->flush_socket()) {
 353         $logger->error("We received a request, bu we are no longer connected to opensrf.  ".
 354             "Exiting and dropping request from $client");
 355         exit;
 356     }
 357
 358     my $data;
 359     eval {
 360         local $SIG{ALRM} = sub { die "alarm\n" };
 361         alarm $wait_time; # prevent accidental tie ups of backend processes
 362         local $/ = "\x1D"; # MARC record separator
 363         $data = <STDIN>;
 364         alarm 0;
 365     };
 366
 367     if($@) {
 368         $logger->error("reading from STDIN failed or timed out: $@");
 369         return;
 370     }
 371
 372     $logger->info("stream parser read " . length($data) . " bytes");
 373
 374     my ($imported, $failed) = (0, 0);
 375
 376     if ($real_opts->{noqueue}) {
 377         ($imported, $failed) = old_process_batch_data($data);
 378     } else {
 379         ($imported, $failed) = process_batch_data($data);
 380     }
 381
 382     my $profile = (!$merge_profile) ? '' :
 383         $apputils->simplereq(
 384             'open-ils.pcrud',
 385             'open-ils.pcrud.retrieve.vmp',
 386             $authtoken,
 387             $merge_profile)->name;
 388
 389     my $msg = '';
 390     $msg .= "Successfully imported $imported records using merge profile '$profile'\n" if $imported;
 391     $msg .= "Failed to import $failed records\n" if $failed;
 392     $msg .= "\x00";
 393     print $client $msg;
 394 }
 395
 396
 397 # the authtoken will timeout after the configured inactivity period.
 398 # When that happens, get a new one.
 399 sub new_auth_token {
 400     $authtoken = oils_login($oils_username, $oils_password, 'staff')
 401         or die "Unable to login to Evergreen as user $oils_username";
 402     return $authtoken;
 403 }
 404
 405 ##### MAIN ######
 406
 407 osrf_connect($osrf_config);
 408 new_auth_token();
 409 print "Calling Net::Server run ", (@ARGV ? "with command-line options: " . join(' ', @ARGV) : ''), "\n";
 410 __PACKAGE__->run(conf_file => $conf_file);
 411
 412 __END__
 413
 414 =head1 NAME
 415
 416 marc_stream_importer.pl - Import MARC records via bare socket connection.
 417
 418 =head1 SYNOPSIS
 419
 420 ./marc_stream_importer.pl [common opts ...] [script opts ...] -- [Net::Server opts ...] &
 421
 422 This script uses the EG common options from B<Cronscript>.  See --help output for those.
 423
 424 Run C<perldoc marc_stream_importer.pl> for full documentation.
 425
 426 Note the extra C<--> to separate options for the script wrapper from options for the
 427 underlying L<Net::Server> options.
 428
 429 Note: this script has to be run in the same directory as B<oils_header.pl>.
 430
 431 Typical execution will include a trailing C<&> to run in the background.
 432
 433 =head1 DESCRIPTION
 434
 435 This script is a L<Net::Server::PreFork> instance for shoving records into Evergreen from a remote system.
 436
 437 =head1 OPTIONS
 438
 439 The only required option is --password
 440
 441  --password         =<eg_password>
 442  --user             =<eg_username>  default: admin
 443  --source           =<bib_source>   default: 1         Integer
 444  --merge-profile    =<i>            default: 0
 445  --tempdir          =</temp/dir/>   default: from L<opensrf.conf> <open-ils.vandelay/app_settings/databases/importer>
 446  --source           =<i>            default: 1
 447  --import-by-queue  =<i>            default: 0
 448
 449
 450 =head2 Old style: --noqueue and associated options
 451
 452 To bypass vandelay queue processing and push directly into the database (as the old style)
 453
 454  --noqueue         default: OFF
 455  --buffsize =<i>   default: 4096    Buffer size.  Only used by --noqueue
 456  --wait     =<i>   default: 5       Seconds to read socket before processing.  Only used by --noqueue
 457
 458 =head2 Net::Server Options
 459
 460 By default, the script will use the Net::Server configuration file B<marc_stream_importer.conf>.  You can
 461 override this by passing a filepath with the --conf_file option.
 462
 463 Other Net::Server options include: --port=<port> --min_servers=<X> --max_servers=<Y> and --log_file=[path/to/file]
 464
 465 See L<Net::Server> for a complete list.
 466
 467 =head2 Configuration
 468
 469 =head3 OCLC Connexion
 470
 471 To use this script with OCLC Connexion, configure the client as follows:
 472
 473 Under Tools -> Options -> Export (tab)
 474    Create -> Choose Connection -> OK -> Leave translation at "None"
 475        -> Create -> Create -> choose TCP/IP (internet)
 476        -> Enter hostname and Port, leave 'Use Telnet Protocol' checked
 477        -> Create/OK your way out of the dialogs
 478    Record Characteristics (button) -> Choose 'UTF-8 Unicode' for the Character Set
 479
 480
 481 OCLC and Connexion are trademark/service marks of OCLC Online Computer Library Center, Inc.
 482
 483 =head1 CAVEATS
 484
 485 WARNING: This script provides no inherent security layer.  Any client that has
 486 access to the server+port can inject MARC records into the system.
 487 Use the available options (like allow/deny) in the Net::Server config file
 488 or via the command line to restrict access as necessary.
 489
 490 =head1 EXAMPLES
 491
 492 ./marc_stream_importer.pl  \
 493     admin open-ils connexion --port 5555 --min_servers 2 \
 494     --max_servers=20 --log_file=/openils/var/log/marc_net_importer.log &
 495
 496 =head1 SEE ALSO
 497
 498 L<Net::Server::PreFork>, L<marc_stream_importer.conf>
 499
 500 =head1 AUTHORS
 501
 502     Bill Erickson <erickson@esilibrary.com>
 503     Joe Atzberger <jatzberger@esilibrary.com>
 504
 505 =cut