now using the local, uni-transaction versions of bib create/update
[Evergreen.git] / Open-ILS / src / perlmods / OpenILS / Application / Vandelay.pm
1 package OpenILS::Application::Vandelay;
2 use strict; use warnings;
3 use OpenILS::Application;
4 use base qw/OpenILS::Application/;
5 use Unicode::Normalize;
6 use OpenSRF::EX qw/:try/;
7 use OpenSRF::AppSession;
8 use OpenSRF::Utils::SettingsClient;
9 use OpenSRF::Utils::Cache;
10 use OpenILS::Utils::Fieldmapper;
11 use OpenILS::Utils::CStoreEditor qw/:funcs/;
12 use MARC::Batch;
13 use MARC::Record;
14 use MARC::File::XML;
15 use OpenILS::Utils::Fieldmapper;
16 use Time::HiRes qw(time);
17 use OpenSRF::Utils::Logger qw/$logger/;
18 use MIME::Base64;
19 use OpenILS::Application::AppUtils;
20 use OpenILS::Application::Cat::BibCommon;
21 my $U = 'OpenILS::Application::AppUtils';
22
23 sub initialize {}
24 sub child_init {}
25
26 # --------------------------------------------------------------------------------
27 # Biblio ingest
28
29 sub create_bib_queue {
30         my $self = shift;
31         my $client = shift;
32         my $auth = shift;
33         my $name = shift;
34         my $owner = shift;
35         my $type = shift;
36
37         my $e = new_editor(authtoken => $auth, xact => 1);
38
39         return $e->die_event unless $e->checkauth;
40         return $e->die_event unless $e->allowed('CREATE_BIB_IMPORT_QUEUE');
41     $owner ||= $e->requestor->id;
42
43         my $queue = new Fieldmapper::vandelay::bib_queue();
44         $queue->name( $name );
45         $queue->owner( $owner );
46         $queue->queue_type( $type ) if ($type);
47
48         my $new_q = $e->create_vandelay_bib_queue( $queue );
49         return $e->die_event unless ($new_q);
50         $e->commit;
51
52     return $new_q;
53 }
54 __PACKAGE__->register_method(  
55         api_name        => "open-ils.vandelay.bib_queue.create",
56         method          => "create_bib_queue",
57         api_level       => 1,
58         argc            => 3,
59 );                      
60
61
62 sub create_auth_queue {
63         my $self = shift;
64         my $client = shift;
65         my $auth = shift;
66         my $name = shift;
67         my $owner = shift;
68         my $type = shift;
69
70         my $e = new_editor(authtoken => $auth, xact => 1);
71
72         return $e->die_event unless $e->checkauth;
73         return $e->die_event unless $e->allowed('CREATE_AUTHORITY_IMPORT_QUEUE');
74     $owner ||= $e->requestor->id;
75
76         my $queue = new Fieldmapper::vandelay::authority_queue();
77         $queue->name( $name );
78         $queue->owner( $owner );
79         $queue->queue_type( $type ) if ($type);
80
81         my $new_q = $e->create_vandelay_authority_queue( $queue );
82         $e->die_event unless ($new_q);
83         $e->commit;
84
85     return $new_q;
86 }
87 __PACKAGE__->register_method(  
88         api_name        => "open-ils.vandelay.authority_queue.create",
89         method          => "create_auth_queue",
90         api_level       => 1,
91         argc            => 3,
92 );                      
93
94 sub add_record_to_bib_queue {
95         my $self = shift;
96         my $client = shift;
97         my $auth = shift;
98         my $queue = shift;
99         my $marc = shift;
100         my $purpose = shift;
101
102         my $e = new_editor(authtoken => $auth, xact => 1);
103
104         $queue = $e->retrieve_vandelay_bib_queue($queue);
105
106         return $e->die_event unless $e->checkauth;
107         return $e->die_event unless
108                 ($e->allowed('CREATE_BIB_IMPORT_QUEUE', undef, $queue) ||
109                  $e->allowed('CREATE_BIB_IMPORT_QUEUE'));
110
111         my $new_rec = _add_bib_rec($e, $marc, $queue->id, $purpose);
112
113         return $e->die_event unless ($new_rec);
114         $e->commit;
115     return $new_rec;
116 }
117 __PACKAGE__->register_method(  
118         api_name        => "open-ils.vandelay.queued_bib_record.create",
119         method          => "add_record_to_bib_queue",
120         api_level       => 1,
121         argc            => 3,
122 );                      
123
124 sub _add_bib_rec {
125         my $e = shift;
126         my $marc = shift;
127         my $queue = shift;
128         my $purpose = shift;
129
130         my $rec = new Fieldmapper::vandelay::queued_bib_record();
131         $rec->marc( $marc );
132         $rec->queue( $queue );
133         $rec->purpose( $purpose ) if ($purpose);
134
135         return $e->create_vandelay_queued_bib_record( $rec );
136 }
137
138 sub add_record_to_authority_queue {
139         my $self = shift;
140         my $client = shift;
141         my $auth = shift;
142         my $queue = shift;
143         my $marc = shift;
144         my $purpose = shift;
145
146         my $e = new_editor(authtoken => $auth, xact => 1);
147
148         $queue = $e->retrieve_vandelay_authority_queue($queue);
149
150         return $e->die_event unless $e->checkauth;
151         return $e->die_event unless
152                 ($e->allowed('CREATE_AUTHORITY_IMPORT_QUEUE', undef, $queue) ||
153                  $e->allowed('CREATE_AUTHORITY_IMPORT_QUEUE'));
154
155         my $new_rec = _add_auth_rec($e, $marc, $queue->id, $purpose);
156
157         return $e->die_event unless ($new_rec);
158         $e->commit;
159     return $new_rec;
160 }
161 __PACKAGE__->register_method(
162         api_name        => "open-ils.vandelay.queued_authority_record.create",
163         method          => "add_record_to_authority_queue",
164         api_level       => 1,
165         argc            => 3,
166 );
167
168 sub _add_auth_rec {
169         my $e = shift;
170         my $marc = shift;
171         my $queue = shift;
172     my $purpose = shift;
173
174         my $rec = new Fieldmapper::vandelay::queued_authority_record();
175         $rec->marc( $marc );
176         $rec->queue( $queue );
177         $rec->purpose( $purpose ) if ($purpose);
178
179         return $e->create_vandelay_queued_authority_record( $rec );
180 }
181
182 sub process_spool {
183         my $self = shift;
184         my $client = shift;
185         my $auth = shift;
186         my $fingerprint = shift;
187         my $queue_id = shift;
188
189         my $e = new_editor(authtoken => $auth, xact => 1);
190     return $e->die_event unless $e->checkauth;
191
192     my $queue;
193     my $type = $self->{record_type};
194
195     if($type eq 'bib') {
196         $queue = $e->retrieve_vandelay_bib_queue($queue_id) or return $e->die_event;
197     } else {
198         $queue = $e->retrieve_vandelay_authority_queue($queue_id) or return $e->die_event;
199     }
200
201     my $evt = check_queue_perms($e, $type, $queue);
202     return $evt if $evt;
203
204         my $method = "open-ils.vandelay.queued_${type}_record.create";
205         $method = $self->method_lookup( $method );
206
207     my $cache = new OpenSRF::Utils::Cache();
208
209     my $data = $cache->get_cache('vandelay_import_spool_' . $fingerprint);
210         my $purpose = $data->{purpose};
211     $data = decode_base64($data->{marc});
212
213     $logger->info("vandelay loaded $fingerprint purpose=$purpose and ".length($data)." bytes of data");
214
215     my $fh;
216     open $fh, '<', \$data;
217
218     my $marctype = 'USMARC'; # ?
219         my $batch = new MARC::Batch ( $marctype, $fh );
220         $batch->strict_off;
221
222         my $count = 0;
223         while (my $r = $batch->next) {
224         $logger->info("processing record $count");
225                 try {
226                         (my $xml = $r->as_xml_record()) =~ s/\n//sog;
227                         $xml =~ s/^<\?xml.+\?\s*>//go;
228                         $xml =~ s/>\s+</></go;
229                         $xml =~ s/\p{Cc}//go;
230                         $xml = $U->entityize($xml);
231                         $xml =~ s/[\x00-\x1f]//go;
232
233                         if ($type eq 'bib') {
234                                 _add_bib_rec( $e, $xml, $queue_id, $purpose ) or return $e->die_event;
235                         } else {
236                                 _add_auth_rec( $e, $xml, $queue_id, $purpose ) or return $e->die_event;
237                         }
238                         $count++;
239                         
240                         $client->respond( $count );
241                 } catch Error with {
242                         my $error = shift;
243                         $logger->warn("Encountered a bad record at Vandelay ingest: ".$error);
244                 }
245         }
246
247         $e->commit;
248         return undef;
249 }
250 __PACKAGE__->register_method(  
251         api_name        => "open-ils.vandelay.bib.process_spool",
252         method          => "process_spool",
253         api_level       => 1,
254         argc            => 3,
255         record_type     => 'bib'
256 );                      
257 __PACKAGE__->register_method(  
258         api_name        => "open-ils.vandelay.auth.process_spool",
259         method          => "process_spool",
260         api_level       => 1,
261         argc            => 3,
262         record_type     => 'auth'
263 );                      
264
265
266 __PACKAGE__->register_method(  
267         api_name        => "open-ils.vandelay.bib_queue.records.retrieve",
268         method          => 'retrieve_queue',
269         api_level       => 1,
270         argc            => 2,
271     stream      => 1,
272         record_type     => 'bib'
273 );
274 __PACKAGE__->register_method(  
275         api_name        => "open-ils.vandelay.auth_queue.records.retrieve",
276         method          => 'retrieve_queue',
277         api_level       => 1,
278         argc            => 2,
279     stream      => 1,
280         record_type     => 'auth'
281 );
282
283 __PACKAGE__->register_method(  
284         api_name        => "open-ils.vandelay.bib_queue.records.matches.retrieve",
285         method          => 'retrieve_queue',
286         api_level       => 1,
287         argc            => 2,
288     stream      => 1,
289         record_type     => 'bib',
290     signature   => {
291         desc => q/Only retrieve queued bib records that have matches against existing records/
292     }
293 );
294 __PACKAGE__->register_method(  
295         api_name        => "open-ils.vandelay.auth_queue.records.matches.retrieve",
296         method          => 'retrieve_queue',
297         api_level       => 1,
298         argc            => 2,
299     stream      => 1,
300         record_type     => 'auth',
301     signature   => {
302         desc => q/Only retrieve queued authority records that have matches against existing records/
303     }
304
305 );
306
307 sub retrieve_queue {
308     my($self, $conn, $auth, $queue_id, $options) = @_;
309     my $e = new_editor(authtoken => $auth);
310     return $e->event unless $e->checkauth;
311     $options ||= {};
312     my $limit = $$options{limit} || 20;
313     my $offset = $$options{offset} || 0;
314
315     my $type = $self->{record_type};
316     my $queue;
317     if($type eq 'bib') {
318         $queue = $e->retrieve_vandelay_bib_queue($queue_id) or return $e->die_event;
319     } else {
320         $queue = $e->retrieve_vandelay_authority_queue($queue_id) or return $e->die_event;
321     }
322     my $evt = check_queue_perms($e, $type, $queue);
323     return $evt if $evt;
324
325     my $class = ($type eq 'bib') ? 'vqbr' : 'vqar';
326     my $search = ($type eq 'bib') ? 
327         'search_vandelay_queued_bib_record' : 'search_vandelay_queued_authority_record';
328     my $retrieve = ($type eq 'bib') ? 
329         'retrieve_vandelay_queued_bib_record' : 'retrieve_vandelay_queued_authority_record';
330
331     my $record_ids;
332     if($self->api_name =~ /matches/) {
333         # fetch only matched records
334         $record_ids = queued_records_with_matches($e, $type, $queue_id, $limit, $offset);
335     } else {
336         # fetch all queue records
337         $record_ids = $e->$search([
338                 {queue => $queue_id}, 
339                 {order_by => {$class => 'id'}, limit => $limit, offset => $offset}
340             ],
341             {idlist => 1}
342         );
343     }
344
345
346     for my $rec_id (@$record_ids) {
347         my $params = {   
348             flesh => 1,
349             flesh_fields => {$class => ['attributes', 'matches']},
350         };
351         my $rec = $e->$retrieve([$rec_id, $params]);
352         $rec->clear_marc if $$options{clear_marc};
353         $conn->respond($rec);
354     }
355     return undef;
356 }
357
358 sub check_queue_perms {
359     my($e, $type, $queue) = @_;
360         if ($type eq 'bib') {
361                 return $e->die_event unless
362                         ($e->allowed('CREATE_BIB_IMPORT_QUEUE', undef, $queue) ||
363                          $e->allowed('CREATE_BIB_IMPORT_QUEUE'));
364         } else {
365                 return $e->die_event unless
366                         ($e->allowed('CREATE_AUTHORITY_IMPORT_QUEUE', undef, $queue) ||
367                          $e->allowed('CREATE_AUTHORITY_IMPORT_QUEUE'));
368         }
369
370     return undef;
371 }
372
373 __PACKAGE__->register_method(  
374         api_name        => "open-ils.vandelay.bib_record.list.import",
375         method          => 'import_record_list',
376         api_level       => 1,
377         argc            => 2,
378     stream      => 1,
379         record_type     => 'bib'
380 );
381
382 __PACKAGE__->register_method(  
383         api_name        => "open-ils.vandelay.auth_record.list.import",
384         method          => 'import_record_list',
385         api_level       => 1,
386         argc            => 2,
387     stream      => 1,
388         record_type     => 'auth'
389 );
390
391 sub import_record_list {
392     my($self, $conn, $auth, $rec_ids, $args) = @_;
393     my $e = new_editor(xact => 1, authtoken => $auth);
394     return $e->die_event unless $e->checkauth;
395     $args ||= {};
396     my $err = import_record_list_impl($self, $conn, $auth, $e, $rec_ids, $args);
397     return $err if $err;
398     $e->commit;
399     return {complete => 1};
400 }
401
402
403 __PACKAGE__->register_method(  
404         api_name        => "open-ils.vandelay.bib_queue.import",
405         method          => 'import_queue',
406         api_level       => 1,
407         argc            => 2,
408     stream      => 1,
409         record_type     => 'bib'
410 );
411
412 __PACKAGE__->register_method(  
413         api_name        => "open-ils.vandelay.auth_queue.import",
414         method          => 'import_queue',
415         api_level       => 1,
416         argc            => 2,
417     stream      => 1,
418         record_type     => 'auth'
419 );
420 __PACKAGE__->register_method(  
421         api_name        => "open-ils.vandelay.bib_queue.nomatch.import",
422         method          => 'import_queue',
423         api_level       => 1,
424         argc            => 2,
425     stream      => 1,
426     signature   => {
427         desc => q/Only import records that have no collisions/
428     },
429         record_type     => 'bib'
430 );
431
432 __PACKAGE__->register_method(  
433         api_name        => "open-ils.vandelay.auth_queue.nomatch.import",
434         method          => 'import_queue',
435         api_level       => 1,
436         argc            => 2,
437     stream      => 1,
438     signature   => {
439         desc => q/Only import records that have no collisions/
440     },
441         record_type     => 'auth'
442 );
443 sub import_queue {
444     my($self, $conn, $auth, $q_id, $options) = @_;
445     my $e = new_editor(xact => 1, authtoken => $auth);
446     return $e->die_event unless $e->checkauth;
447     $options ||= {};
448     my $type = $self->{record_type};
449     my $class = ($type eq 'bib') ? 'vqbr' : 'vqar';
450
451     my $query = {queue => $q_id, import_time => undef};
452
453     if($self->api_name =~ /nomatch/) {
454         my $matched_recs = queued_records_with_matches($e, $type, $q_id, undef, undef, {import_time => undef});
455         $query->{id} = {'not in' => $matched_recs} if @$matched_recs;
456     }
457
458     my $search = ($type eq 'bib') ? 
459         'search_vandelay_queued_bib_record' : 'search_vandelay_queued_authority_record';
460     my $rec_ids = $e->$search($query, {idlist => 1});
461     my $err = import_record_list_impl($self, $conn, $auth, $e, $rec_ids, $options);
462     return $err if $err;
463     $e->commit;
464     return {complete => 1};
465 }
466
467
468 # returns a list of queued record IDs for a given queue that 
469 # have at least one entry in the match table
470 sub queued_records_with_matches {
471     my($e, $type, $q_id, $limit, $offset, $filter) = @_;
472
473     my $match_class = 'vbm';
474     my $rec_class = 'vqbr';
475     if($type eq 'auth') {
476         $match_class = 'vam';
477          $rec_class = 'vqar';
478     }
479
480     $filter ||= {};
481     $filter->{queue} = $q_id;
482
483     my $query = {
484         distinct => 1, 
485         select => {$match_class => ['queued_record']}, 
486         from => {
487             $match_class => {
488                 $rec_class => {
489                     field => 'id',
490                     fkey => 'queued_record',
491                     filter => $filter,
492                 }
493             }
494         }
495     };        
496
497     if($limit or defined $offset) {
498         $limit ||= 20;
499         $offset ||= 0;
500         $query->{limit} = $limit;
501         $query->{offset} = $offset;
502     }
503
504     my $data = $e->json_query($query);
505     return [ map {$_->{queued_record}} @$data ];
506 }
507
508 sub import_record_list_impl {
509     my($self, $conn, $auth, $e, $rec_ids, $args) = @_;
510
511     my $overlay_map = $args->{overlay_map} || {};
512     my $type = $self->{record_type};
513     my $total = @$rec_ids;
514     my $count = 0;
515     my %queues;
516
517     for my $rec_id (@$rec_ids) {
518         if($type eq 'bib') {
519
520             my $rec = $e->retrieve_vandelay_queued_bib_record($rec_id) 
521                 or return $e->die_event;
522
523             next if $rec->import_time;
524             $queues{$rec->queue} = 1;
525
526             my $record;
527             if(defined $overlay_map->{$rec_id}) {
528                 $logger->info("vl: overlaying record $rec_id");
529                 $record = OpenILS::Application::Cat::BibCommon->biblio_record_replace_marc(
530                     $e, $rec_id, $rec->marc); #$rec->bib_source
531             } else {
532                 $logger->info("vl: importing new record");
533                 $record = OpenILS::Application::Cat::BibCommon->biblio_record_xml_import(
534                     $e, $rec->marc); #$rec->bib_source
535             }
536
537             return $record if $U->event_code($record);
538             $rec->imported_as($record->id);
539             $rec->import_time('now');
540             $e->update_vandelay_queued_bib_record($rec) or return $e->die_event;
541
542         } else { # authority
543
544             my $rec = $e->retrieve_vandelay_queued_authority_record($rec_id) 
545                 or return $e->die_event;
546
547             next if $rec->import_time;
548             $queues{$rec->queue} = 1;
549
550             my $record;
551             if(defined $overlay_map->{$rec_id}) {
552                 $logger->info("vl: overlaying record $rec_id");
553                 $record = $U->simplereq(
554                     'open-ils.cat',
555                     'open-ils.cat.authority.record.overlay',
556                     $auth, $overlay_map->{$rec_id}, $rec->marc); #$rec->bib_source);
557             } else {
558                 $logger->info("vl: importing new record");
559                 $record = $U->simplereq(
560                     'open-ils.cat',
561                     'open-ils.cat.authority.record.import',
562                     $auth, $rec->marc); #$rec->bib_source);
563             }
564
565             if($U->event_code($record)) {
566                 $e->rollback;
567                 return $record;
568             }
569
570             $rec->imported_as($record->id);
571             $rec->import_time('now');
572             $e->update_vandelay_queued_authority_record($rec) or return $e->die_event;
573         }
574
575         $conn->respond({total => $total, progress => ++$count, imported => $rec_id});
576     }
577
578     # see if we need to mark any queues as complete
579     for my $q_id (keys %queues) {
580         if($type eq 'bib') {
581             my $remaining = $e->search_vandelay_queued_bib_record(
582                 {queue => $q_id, import_time => undef}, {idlist => 1});
583             unless(@$remaining) {
584                 my $queue = $e->retrieve_vandelay_bib_queue($q_id);
585                 unless($U->is_true($queue->complete)) {
586                     $queue->complete('t');
587                     $e->update_vandelay_bib_queue($queue) or return $e->die_event;
588                 }
589             }
590         } else {
591             my $remaining = $e->search_vandelay_queued_authority_record(
592                 {queue => $q_id, import_time => undef}, {idlist => 1});
593             unless(@$remaining) {
594                 my $queue = $e->retrieve_vandelay_authority_queue($q_id);
595                 unless($U->is_true($queue->complete)) {
596                     $queue->complete('t');
597                     $e->update_vandelay_authority_queue($queue) or return $e->die_event;
598                 }
599             }
600         }
601     }
602
603     return undef;
604 }
605
606
607 __PACKAGE__->register_method(  
608         api_name        => "open-ils.vandelay.bib_queue.owner.retrieve",
609         method          => 'owner_queue_retrieve',
610         api_level       => 1,
611         argc            => 2,
612     stream      => 1,
613         record_type     => 'bib'
614 );
615 __PACKAGE__->register_method(  
616         api_name        => "open-ils.vandelay.authority_queue.owner.retrieve",
617         method          => 'owner_queue_retrieve',
618         api_level       => 1,
619         argc            => 2,
620     stream      => 1,
621         record_type     => 'auth'
622 );
623
624 sub owner_queue_retrieve {
625     my($self, $conn, $auth, $owner_id) = @_;
626     my $e = new_editor(authtoken => $auth);
627     return $e->die_event unless $e->checkauth;
628     $owner_id = $e->requestor->id; # XXX add support for viewing other's queues?
629     my $queues;
630     if($self->{record_type} eq 'bib') {
631         $queues = $e->search_vandelay_bib_queue(
632             {complete => 'f', owner => $owner_id});
633     } else {
634         $queues = $e->search_vandelay_authority_queue(
635             {complete => 'f', owner => $owner_id});
636     }
637     $conn->respond($_) for @$queues;
638     return undef;
639 }
640
641 __PACKAGE__->register_method(  
642         api_name        => "open-ils.vandelay.bib_queue.delete",
643         method          => "delete_queue",
644         api_level       => 1,
645         argc            => 2,
646         record_type     => 'bib'
647 );            
648 __PACKAGE__->register_method(  
649         api_name        => "open-ils.vandelay.auth_queue.delete",
650         method          => "delete_queue",
651         api_level       => 1,
652         argc            => 2,
653         record_type     => 'auth'
654 );  
655
656 sub delete_queue {
657     my($self, $conn, $auth, $q_id) = @_;
658     my $e = new_editor(xact => 1, authtoken => $auth);
659     return $e->die_event unless $e->checkauth;
660     if($self->{record_type} eq 'bib') {
661             return $e->die_event unless $e->allowed('CREATE_BIB_IMPORT_QUEUE');
662         my $queue = $e->retrieve_vandelay_bib_queue($q_id)
663             or return $e->die_event;
664         $e->delete_vandelay_bib_queue($queue)
665             or return $e->die_event;
666     } else {
667             return $e->die_event unless $e->allowed('CREATE_AUTHORITY_IMPORT_QUEUE');
668         my $queue = $e->retrieve_vandelay_authority_queue($q_id)
669             or return $e->die_event;
670         $e->delete_vandelay_authority_queue($queue)
671             or return $e->die_event;
672     }
673     $e->commit;
674     return 1;
675 }
676
677 1;