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