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::Const qw/:const/;
  20 use OpenILS::Application::AppUtils;
  21 use OpenILS::Application::Cat::BibCommon;
  22 use OpenILS::Application::Cat::AuthCommon;
  23 use OpenILS::Application::Cat::AssetCommon;
  24 my $U = 'OpenILS::Application::AppUtils';
  25
  26 # A list of LDR/06 values from http://loc.gov/marc
  27 my %record_types = (
  28         a => 'bib',
  29         c => 'bib',
  30         d => 'bib',
  31         e => 'bib',
  32         f => 'bib',
  33         g => 'bib',
  34         i => 'bib',
  35         j => 'bib',
  36         k => 'bib',
  37         m => 'bib',
  38         o => 'bib',
  39         p => 'bib',
  40         r => 'bib',
  41         t => 'bib',
  42         u => 'holdings',
  43         v => 'holdings',
  44         x => 'holdings',
  45         y => 'holdings',
  46         z => 'auth',
  47       ' ' => 'bib',
  48 );
  49
  50 sub initialize {}
  51 sub child_init {}
  52
  53 # --------------------------------------------------------------------------------
  54 # Biblio ingest
  55
  56 sub create_bib_queue {
  57     my $self = shift;
  58     my $client = shift;
  59     my $auth = shift;
  60     my $name = shift;
  61     my $owner = shift;
  62     my $type = shift;
  63     my $import_def = shift;
  64
  65     my $e = new_editor(authtoken => $auth, xact => 1);
  66
  67     return $e->die_event unless $e->checkauth;
  68     return $e->die_event unless $e->allowed('CREATE_BIB_IMPORT_QUEUE');
  69     $owner ||= $e->requestor->id;
  70
  71     if ($e->search_vandelay_bib_queue( {name => $name, owner => $owner, queue_type => $type})->[0]) {
  72         $e->rollback;
  73         return OpenILS::Event->new('BIB_QUEUE_EXISTS')
  74     }
  75
  76     my $queue = new Fieldmapper::vandelay::bib_queue();
  77     $queue->name( $name );
  78     $queue->owner( $owner );
  79     $queue->queue_type( $type ) if ($type);
  80     $queue->item_attr_def( $import_def ) if ($import_def);
  81
  82     my $new_q = $e->create_vandelay_bib_queue( $queue );
  83     return $e->die_event unless ($new_q);
  84     $e->commit;
  85
  86     return $new_q;
  87 }
  88 __PACKAGE__->register_method(
  89     api_name   => "open-ils.vandelay.bib_queue.create",
  90     method     => "create_bib_queue",
  91     api_level  => 1,
  92     argc       => 4,
  93 );
  94
  95
  96 sub create_auth_queue {
  97     my $self = shift;
  98     my $client = shift;
  99     my $auth = shift;
 100     my $name = shift;
 101     my $owner = shift;
 102     my $type = shift;
 103
 104     my $e = new_editor(authtoken => $auth, xact => 1);
 105
 106     return $e->die_event unless $e->checkauth;
 107     return $e->die_event unless $e->allowed('CREATE_AUTHORITY_IMPORT_QUEUE');
 108     $owner ||= $e->requestor->id;
 109
 110     if ($e->search_vandelay_bib_queue({name => $name, owner => $owner, queue_type => $type})->[0]) {
 111         $e->rollback;
 112         return OpenILS::Event->new('AUTH_QUEUE_EXISTS')
 113     }
 114
 115     my $queue = new Fieldmapper::vandelay::authority_queue();
 116     $queue->name( $name );
 117     $queue->owner( $owner );
 118     $queue->queue_type( $type ) if ($type);
 119
 120     my $new_q = $e->create_vandelay_authority_queue( $queue );
 121     $e->die_event unless ($new_q);
 122     $e->commit;
 123
 124     return $new_q;
 125 }
 126 __PACKAGE__->register_method(
 127     api_name   => "open-ils.vandelay.authority_queue.create",
 128     method     => "create_auth_queue",
 129     api_level  => 1,
 130     argc       => 3,
 131 );
 132
 133 sub add_record_to_bib_queue {
 134     my $self = shift;
 135     my $client = shift;
 136     my $auth = shift;
 137     my $queue = shift;
 138     my $marc = shift;
 139     my $purpose = shift;
 140     my $bib_source = shift;
 141
 142     my $e = new_editor(authtoken => $auth, xact => 1);
 143
 144     $queue = $e->retrieve_vandelay_bib_queue($queue);
 145
 146     return $e->die_event unless $e->checkauth;
 147     return $e->die_event unless
 148         ($e->allowed('CREATE_BIB_IMPORT_QUEUE', undef, $queue) ||
 149          $e->allowed('CREATE_BIB_IMPORT_QUEUE'));
 150
 151     my $new_rec = _add_bib_rec($e, $marc, $queue->id, $purpose, $bib_source);
 152
 153     return $e->die_event unless ($new_rec);
 154     $e->commit;
 155     return $new_rec;
 156 }
 157 __PACKAGE__->register_method(
 158     api_name   => "open-ils.vandelay.queued_bib_record.create",
 159     method     => "add_record_to_bib_queue",
 160     api_level  => 1,
 161     argc       => 3,
 162 );
 163
 164 sub _add_bib_rec {
 165     my $e = shift;
 166     my $marc = shift;
 167     my $queue = shift;
 168     my $purpose = shift;
 169     my $bib_source = shift;
 170
 171     my $rec = new Fieldmapper::vandelay::queued_bib_record();
 172     $rec->marc( $marc );
 173     $rec->queue( $queue );
 174     $rec->purpose( $purpose ) if ($purpose);
 175     $rec->bib_source($bib_source);
 176
 177     return $e->create_vandelay_queued_bib_record( $rec );
 178 }
 179
 180 sub add_record_to_authority_queue {
 181     my $self = shift;
 182     my $client = shift;
 183     my $auth = shift;
 184     my $queue = shift;
 185     my $marc = shift;
 186     my $purpose = shift;
 187
 188     my $e = new_editor(authtoken => $auth, xact => 1);
 189
 190     $queue = $e->retrieve_vandelay_authority_queue($queue);
 191
 192     return $e->die_event unless $e->checkauth;
 193     return $e->die_event unless
 194         ($e->allowed('CREATE_AUTHORITY_IMPORT_QUEUE', undef, $queue) ||
 195          $e->allowed('CREATE_AUTHORITY_IMPORT_QUEUE'));
 196
 197     my $new_rec = _add_auth_rec($e, $marc, $queue->id, $purpose);
 198
 199     return $e->die_event unless ($new_rec);
 200     $e->commit;
 201     return $new_rec;
 202 }
 203 __PACKAGE__->register_method(
 204     api_name   => "open-ils.vandelay.queued_authority_record.create",
 205     method     => "add_record_to_authority_queue",
 206     api_level  => 1,
 207     argc       => 3,
 208 );
 209
 210 sub _add_auth_rec {
 211     my $e = shift;
 212     my $marc = shift;
 213     my $queue = shift;
 214     my $purpose = shift;
 215
 216     my $rec = new Fieldmapper::vandelay::queued_authority_record();
 217     $rec->marc( $marc );
 218     $rec->queue( $queue );
 219     $rec->purpose( $purpose ) if ($purpose);
 220
 221     return $e->create_vandelay_queued_authority_record( $rec );
 222 }
 223
 224 sub process_spool {
 225     my $self = shift;
 226     my $client = shift;
 227     my $auth = shift;
 228     my $fingerprint = shift || '';
 229     my $queue_id = shift;
 230     my $purpose = shift;
 231     my $filename = shift;
 232     my $bib_source = shift;
 233
 234     my $e = new_editor(authtoken => $auth, xact => 1);
 235     return $e->die_event unless $e->checkauth;
 236
 237     my $queue;
 238     my $type = $self->{record_type};
 239
 240     if($type eq 'bib') {
 241         $queue = $e->retrieve_vandelay_bib_queue($queue_id) or return $e->die_event;
 242     } else {
 243         $queue = $e->retrieve_vandelay_authority_queue($queue_id) or return $e->die_event;
 244     }
 245
 246     my $evt = check_queue_perms($e, $type, $queue);
 247     return $evt if ($evt);
 248
 249     my $cache = new OpenSRF::Utils::Cache();
 250
 251     if($fingerprint) {
 252         my $data = $cache->get_cache('vandelay_import_spool_' . $fingerprint);
 253         $purpose = $data->{purpose};
 254         $filename = $data->{path};
 255         $bib_source = $data->{bib_source};
 256     }
 257
 258     unless(-r $filename) {
 259         $logger->error("unable to read MARC file $filename");
 260         return -1; # make this an event XXX
 261     }
 262
 263     $logger->info("vandelay spooling $fingerprint purpose=$purpose file=$filename");
 264
 265     my $marctype = 'USMARC';
 266
 267     open F, $filename;
 268     $marctype = 'XML' if (getc(F) =~ /^\D/o);
 269     close F;
 270
 271     my $batch = new MARC::Batch ($marctype, $filename);
 272     $batch->strict_off;
 273
 274     my $response_scale = 10;
 275     my $count = 0;
 276     my $r = -1;
 277     while (try { $r = $batch->next } otherwise { $r = -1 }) {
 278         if ($r == -1) {
 279             $logger->warn("Processing of record $count in set $filename failed.  Skipping this record");
 280             $count++;
 281         }
 282
 283         $logger->info("processing record $count");
 284
 285         try {
 286             (my $xml = $r->as_xml_record()) =~ s/\n//sog;
 287             $xml =~ s/^<\?xml.+\?\s*>//go;
 288             $xml =~ s/>\s+</></go;
 289             $xml =~ s/\p{Cc}//go;
 290             $xml = $U->entityize($xml);
 291             $xml =~ s/[\x00-\x1f]//go;
 292
 293             my $qrec;
 294             # Check the leader to ensure we've got something resembling the expected
 295             # Allow spaces to give records the benefit of the doubt
 296             my $ldr_type = substr($r->leader(), 6, 1);
 297             if ($type eq 'bib' && ($record_types{$ldr_type}) eq 'bib' || $ldr_type eq ' ') {
 298                 $qrec = _add_bib_rec( $e, $xml, $queue_id, $purpose, $bib_source ) or return $e->die_event;
 299             } elsif ($type eq 'auth' && ($record_types{$ldr_type}) eq 'auth' || $ldr_type eq ' ') {
 300                 $qrec = _add_auth_rec( $e, $xml, $queue_id, $purpose ) or return $e->die_event;
 301             } else {
 302                 # I don't know how to handle this type; rock on
 303                 $logger->error("In process_spool(), type was $type and leader type was $ldr_type ; not currently supported");
 304                 next;
 305             }
 306
 307             if($self->api_name =~ /stream_results/ and $qrec) {
 308                 $client->respond($qrec->id)
 309             } else {
 310                 $client->respond($count) if (++$count % $response_scale) == 0;
 311                 $response_scale *= 10 if ($count == ($response_scale * 10));
 312             }
 313         } catch Error with {
 314             my $error = shift;
 315             $logger->warn("Encountered a bad record at Vandelay ingest: ".$error);
 316         }
 317     }
 318
 319     $e->commit;
 320     unlink($filename);
 321     $cache->delete_cache('vandelay_import_spool_' . $fingerprint) if $fingerprint;
 322     return $count;
 323 }
 324
 325 __PACKAGE__->register_method(
 326     api_name    => "open-ils.vandelay.bib.process_spool",
 327     method      => "process_spool",
 328     api_level   => 1,
 329     argc        => 3,
 330     record_type => 'bib'
 331 );
 332 __PACKAGE__->register_method(
 333     api_name    => "open-ils.vandelay.auth.process_spool",
 334     method      => "process_spool",
 335     api_level   => 1,
 336     argc        => 3,
 337     record_type => 'auth'
 338 );
 339
 340 __PACKAGE__->register_method(
 341     api_name    => "open-ils.vandelay.bib.process_spool.stream_results",
 342     method      => "process_spool",
 343     api_level   => 1,
 344     argc        => 3,
 345     stream      => 1,
 346     record_type => 'bib'
 347 );
 348 __PACKAGE__->register_method(
 349     api_name    => "open-ils.vandelay.auth.process_spool.stream_results",
 350     method      => "process_spool",
 351     api_level   => 1,
 352     argc        => 3,
 353     stream      => 1,
 354     record_type => 'auth'
 355 );
 356
 357 __PACKAGE__->register_method(
 358     api_name    => "open-ils.vandelay.bib_queue.records.retrieve",
 359     method      => 'retrieve_queued_records',
 360     api_level   => 1,
 361     argc        => 2,
 362     stream      => 1,
 363     record_type => 'bib'
 364 );
 365 __PACKAGE__->register_method(
 366     api_name    => "open-ils.vandelay.auth_queue.records.retrieve",
 367     method      => 'retrieve_queued_records',
 368     api_level   => 1,
 369     argc        => 2,
 370     stream      => 1,
 371     record_type => 'auth'
 372 );
 373
 374 __PACKAGE__->register_method(
 375     api_name    => "open-ils.vandelay.bib_queue.records.matches.retrieve",
 376     method      => 'retrieve_queued_records',
 377     api_level   => 1,
 378     argc        => 2,
 379     stream      => 1,
 380     record_type => 'bib',
 381     signature   => {
 382         desc => q/Only retrieve queued bib records that have matches against existing records/
 383     }
 384 );
 385 __PACKAGE__->register_method(
 386     api_name    => "open-ils.vandelay.auth_queue.records.matches.retrieve",
 387     method      => 'retrieve_queued_records',
 388     api_level   => 1,
 389     argc        => 2,
 390     stream      => 1,
 391     record_type => 'auth',
 392     signature   => {
 393         desc => q/Only retrieve queued authority records that have matches against existing records/
 394     }
 395
 396 );
 397
 398 sub retrieve_queued_records {
 399     my($self, $conn, $auth, $queue_id, $options) = @_;
 400     my $e = new_editor(authtoken => $auth, xact => 1);
 401     return $e->die_event unless $e->checkauth;
 402     $options ||= {};
 403     my $limit = $$options{limit} || 20;
 404     my $offset = $$options{offset} || 0;
 405
 406     my $type = $self->{record_type};
 407     my $queue;
 408     if($type eq 'bib') {
 409         $queue = $e->retrieve_vandelay_bib_queue($queue_id) or return $e->die_event;
 410     } else {
 411         $queue = $e->retrieve_vandelay_authority_queue($queue_id) or return $e->die_event;
 412     }
 413     my $evt = check_queue_perms($e, $type, $queue);
 414     return $evt if ($evt);
 415
 416     my $class = ($type eq 'bib') ? 'vqbr' : 'vqar';
 417     my $search = ($type eq 'bib') ?
 418         'search_vandelay_queued_bib_record' : 'search_vandelay_queued_authority_record';
 419     my $retrieve = ($type eq 'bib') ?
 420         'retrieve_vandelay_queued_bib_record' : 'retrieve_vandelay_queued_authority_record';
 421
 422     my $filter = ($$options{non_imported}) ? {import_time => undef} : {};
 423
 424     my $record_ids;
 425     if($self->api_name =~ /matches/) {
 426         # fetch only matched records
 427         $record_ids = queued_records_with_matches($e, $type, $queue_id, $limit, $offset, $filter);
 428     } else {
 429         # fetch all queue records
 430         $record_ids = $e->$search([
 431                 {queue => $queue_id, %$filter},
 432                 {order_by => {$class => 'id'}, limit => $limit, offset => $offset}
 433             ],
 434             {idlist => 1}
 435         );
 436     }
 437
 438
 439     for my $rec_id (@$record_ids) {
 440         my $params = {
 441             flesh => 1,
 442             flesh_fields => {$class => ['attributes', 'matches']},
 443         };
 444         my $rec = $e->$retrieve([$rec_id, $params]);
 445         $rec->clear_marc if $$options{clear_marc};
 446         $conn->respond($rec);
 447     }
 448     $e->rollback;
 449     return undef;
 450 }
 451
 452 sub check_queue_perms {
 453     my($e, $type, $queue) = @_;
 454     if ($type eq 'bib') {
 455         return $e->die_event unless
 456             ($e->allowed('CREATE_BIB_IMPORT_QUEUE', undef, $queue) ||
 457              $e->allowed('CREATE_BIB_IMPORT_QUEUE'));
 458     } else {
 459         return $e->die_event unless
 460             ($e->allowed('CREATE_AUTHORITY_IMPORT_QUEUE', undef, $queue) ||
 461              $e->allowed('CREATE_AUTHORITY_IMPORT_QUEUE'));
 462     }
 463
 464     return undef;
 465 }
 466
 467 __PACKAGE__->register_method(
 468     api_name    => "open-ils.vandelay.bib_record.list.import",
 469     method      => 'import_record_list',
 470     api_level   => 1,
 471     argc        => 2,
 472     stream      => 1,
 473     record_type => 'bib'
 474 );
 475
 476 __PACKAGE__->register_method(
 477     api_name    => "open-ils.vandelay.auth_record.list.import",
 478     method      => 'import_record_list',
 479     api_level   => 1,
 480     argc        => 2,
 481     stream      => 1,
 482     record_type => 'auth'
 483 );
 484
 485 sub import_record_list {
 486     my($self, $conn, $auth, $rec_ids, $args) = @_;
 487     my $e = new_editor(authtoken => $auth, xact => 1);
 488     return $e->die_event unless $e->checkauth;
 489     $args ||= {};
 490     my $err = import_record_list_impl($self, $conn, $rec_ids, $e->requestor, $args);
 491     $e->rollback;
 492     return $err if $err;
 493     return {complete => 1};
 494 }
 495
 496
 497 __PACKAGE__->register_method(
 498     api_name    => "open-ils.vandelay.bib_queue.import",
 499     method      => 'import_queue',
 500     api_level   => 1,
 501     argc        => 2,
 502     stream      => 1,
 503     record_type => 'bib'
 504 );
 505
 506 __PACKAGE__->register_method(
 507     api_name    => "open-ils.vandelay.auth_queue.import",
 508     method      => 'import_queue',
 509     api_level   => 1,
 510     argc        => 2,
 511     stream      => 1,
 512     record_type => 'auth'
 513 );
 514 __PACKAGE__->register_method(
 515     api_name    => "open-ils.vandelay.bib_queue.nomatch.import",
 516     method      => 'import_queue',
 517     api_level   => 1,
 518     argc        => 2,
 519     stream      => 1,
 520     signature   => {
 521         desc => q/Only import records that have no collisions/
 522     },
 523     record_type => 'bib'
 524 );
 525
 526 __PACKAGE__->register_method(
 527     api_name    => "open-ils.vandelay.auth_queue.nomatch.import",
 528     method      => 'import_queue',
 529     api_level   => 1,
 530     argc        => 2,
 531     stream      => 1,
 532     signature   => {
 533         desc => q/Only import records that have no collisions/
 534     },
 535     record_type => 'auth'
 536 );
 537 sub import_queue {
 538     my($self, $conn, $auth, $q_id, $options) = @_;
 539     my $e = new_editor(authtoken => $auth, xact => 1);
 540     return $e->die_event unless $e->checkauth;
 541     $options ||= {};
 542     my $type = $self->{record_type};
 543     my $class = ($type eq 'bib') ? 'vqbr' : 'vqar';
 544
 545     my $query = {queue => $q_id, import_time => undef};
 546
 547     if($self->api_name =~ /nomatch/) {
 548         my $matched_recs = queued_records_with_matches($e, $type, $q_id, undef, undef, {import_time => undef});
 549         $query->{id} = {'not in' => $matched_recs} if @$matched_recs;
 550     }
 551
 552     my $search = ($type eq 'bib') ?
 553         'search_vandelay_queued_bib_record' : 'search_vandelay_queued_authority_record';
 554     my $rec_ids = $e->$search($query, {idlist => 1});
 555     my $err = import_record_list_impl($self, $conn, $rec_ids, $e->requestor, $options);
 556     try {$e->rollback} otherwise {}; # only using this to make the read authoritative -- don't die from it
 557     return $err if $err;
 558     return {complete => 1};
 559 }
 560
 561 # returns a list of queued record IDs for a given queue that
 562 # have at least one entry in the match table
 563 sub queued_records_with_matches {
 564     my($e, $type, $q_id, $limit, $offset, $filter) = @_;
 565
 566     my $match_class = 'vbm';
 567     my $rec_class = 'vqbr';
 568     if($type eq 'auth') {
 569         $match_class = 'vam';
 570          $rec_class = 'vqar';
 571     }
 572
 573     $filter ||= {};
 574     $filter->{queue} = $q_id;
 575
 576     my $query = {
 577         distinct => 1,
 578         select => {$match_class => ['queued_record']},
 579         from => {
 580             $match_class => {
 581                 $rec_class => {
 582                     field => 'id',
 583                     fkey => 'queued_record',
 584                     filter => $filter,
 585                 }
 586             }
 587         }
 588     };
 589
 590     if($limit or defined $offset) {
 591         $limit ||= 20;
 592         $offset ||= 0;
 593         $query->{limit} = $limit;
 594         $query->{offset} = $offset;
 595     }
 596
 597     my $data = $e->json_query($query);
 598     return [ map {$_->{queued_record}} @$data ];
 599 }
 600
 601 sub import_record_list_impl {
 602     my($self, $conn, $rec_ids, $requestor, $args) = @_;
 603
 604     my $overlay_map = $args->{overlay_map} || {};
 605     my $type = $self->{record_type};
 606     my $total = @$rec_ids;
 607     my $count = 0;
 608     my $step = '1' . '0' x ((length($total) * 3) - 2);
 609     my %queues;
 610     my $auto_overlay_exact = $$args{auto_overlay_exact};
 611     my $auto_overlay_1match = $$args{auto_overlay_1match};
 612     my $merge_profile = $$args{merge_profile};
 613     my $bib_source = $$args{bib_source};
 614     my $report_all = $$args{report_all};
 615
 616     my $overlay_func = 'vandelay.overlay_bib_record';
 617     my $auto_overlay_func = 'vandelay.auto_overlay_bib_record';
 618     my $retrieve_func = 'retrieve_vandelay_queued_bib_record';
 619     my $update_func = 'update_vandelay_queued_bib_record';
 620     my $search_func = 'search_vandelay_queued_bib_record';
 621     my $retrieve_queue_func = 'retrieve_vandelay_bib_queue';
 622     my $update_queue_func = 'update_vandelay_bib_queue';
 623     my $rec_class = 'vqbr';
 624
 625     my %bib_sources;
 626     my $editor = new_editor();
 627     my $sources = $editor->search_config_bib_source({id => {'!=' => undef}});
 628
 629     foreach my $src (@$sources) {
 630         $bib_sources{$src->id} = $src->source;
 631     }
 632
 633     if($type eq 'auth') {
 634         $overlay_func =~ s/bib/auth/o;
 635         $auto_overlay_func = s/bib/auth/o;
 636         $retrieve_func =~ s/bib/authority/o;
 637         $retrieve_queue_func =~ s/bib/authority/o;
 638         $update_queue_func =~ s/bib/authority/o;
 639         $update_func =~ s/bib/authority/o;
 640         $search_func =~ s/bib/authority/o;
 641         $rec_class = 'vqar';
 642     }
 643
 644     my @success_rec_ids;
 645     for my $rec_id (@$rec_ids) {
 646
 647         my $overlay_target = $overlay_map->{$rec_id};
 648
 649         my $e = new_editor(xact => 1);
 650         $e->requestor($requestor);
 651
 652         my $rec = $e->$retrieve_func([
 653             $rec_id,
 654             {   flesh => 1,
 655                 flesh_fields => { $rec_class => ['matches']},
 656             }
 657         ]);
 658
 659         unless($rec) {
 660             $conn->respond({total => $total, progress => ++$count, imported => $rec_id, err_event => $e->event});
 661             $e->rollback;
 662             next;
 663         }
 664
 665         if($rec->import_time) {
 666             $e->rollback;
 667             next;
 668         }
 669
 670         $queues{$rec->queue} = 1;
 671
 672         my $record;
 673         my $imported = 0;
 674
 675         if(defined $overlay_target) {
 676             # Caller chose an explicit overlay target
 677
 678             my $res = $e->json_query(
 679                 {
 680                     from => [
 681                         $overlay_func,
 682                         $rec->id,
 683                         $overlay_target,
 684                         $merge_profile
 685                     ]
 686                 }
 687             )->[0];
 688
 689             if($res->{$overlay_func} eq 't') {
 690                 $logger->info("vl: $type direct overlay succeeded for queued rec " .
 691                     $rec->id . " and overlay target $overlay_target");
 692                 $imported = 1;
 693             }
 694
 695         } else {
 696
 697             if($auto_overlay_1match) {
 698                 # caller says to overlay if there is exactly 1 match
 699
 700                 my %match_recs = map { $_->eg_record => 1 } @{$rec->matches};
 701
 702                 if( scalar(keys %match_recs) == 1) { # all matches point to the same record
 703
 704                     my $res = $e->json_query(
 705                         {
 706                             from => [
 707                                 $overlay_func,
 708                                 $rec->id,
 709                                 $rec->matches->[0]->eg_record,
 710                                 $merge_profile
 711                             ]
 712                         }
 713                     )->[0];
 714
 715                     if($res->{$overlay_func} eq 't') {
 716                         $logger->info("vl: $type overlay-1match succeeded for queued rec " . $rec->id);
 717                         $imported = 1;
 718                     }
 719                 }
 720             }
 721
 722             if(!$imported and $auto_overlay_exact and scalar(@{$rec->matches}) == 1 ) {
 723
 724                 # caller says to overlay if there is an /exact/ match
 725
 726                 my $res = $e->json_query(
 727                     {
 728                         from => [
 729                             $auto_overlay_func,
 730                             $rec->id,
 731                             $merge_profile
 732                         ]
 733                     }
 734                 )->[0];
 735
 736                 if($res->{$auto_overlay_func} eq 't') {
 737                     $logger->info("vl: $type auto-overlay succeeded for queued rec " . $rec->id);
 738                     $imported = 1;
 739                 }
 740             }
 741
 742             if(!$imported) {
 743
 744                 # No overlay / merge occurred.  Do a traditional record import by creating a new record
 745
 746                 if($type eq 'bib') {
 747                     $record = OpenILS::Application::Cat::BibCommon->biblio_record_xml_import($e, $rec->marc, $bib_sources{$rec->bib_source});
 748                 } else {
 749
 750                     $record = OpenILS::Application::Cat::AuthCommon->import_authority_record($e, $rec->marc); #$source);
 751                 }
 752
 753                 if($U->event_code($record)) {
 754
 755                     $e->event($record);
 756                     $e->rollback;
 757
 758                 } else {
 759
 760                     $logger->info("vl: successfully imported new $type record");
 761                     $rec->imported_as($record->id);
 762                     $rec->import_time('now');
 763
 764                     $imported = 1 if $e->$update_func($rec);
 765                 }
 766             }
 767         }
 768
 769         if($imported) {
 770             push @success_rec_ids, $rec_id;
 771             $e->commit;
 772         } else {
 773             # Send an update whenever there's an error
 774             $conn->respond({total => $total, progress => ++$count, imported => $rec_id, err_event => $e->event});
 775         }
 776
 777         $conn->respond({total => $total, progress => $count, imported => $rec_id}) if (!$report_all and ++$count % $step) == 0;
 778     }
 779
 780     # see if we need to mark any queues as complete
 781     for my $q_id (keys %queues) {
 782
 783         my $e = new_editor(xact => 1);
 784         my $remaining = $e->$search_func(
 785             [{queue => $q_id, import_time => undef}, {limit =>1}], {idlist => 1});
 786
 787         unless(@$remaining) {
 788             my $queue = $e->$retrieve_queue_func($q_id);
 789
 790             unless($U->is_true($queue->complete)) {
 791                 $queue->complete('t');
 792                 $e->$update_queue_func($queue) or return $e->die_event;
 793                 $e->commit;
 794                 next;
 795             }
 796         }
 797         $e->rollback;
 798     }
 799
 800     import_record_asset_list_impl($conn, \@success_rec_ids, $requestor);
 801
 802     $conn->respond({total => $total, progress => $count});
 803     return undef;
 804 }
 805
 806
 807 __PACKAGE__->register_method(
 808     api_name    => "open-ils.vandelay.bib_queue.owner.retrieve",
 809     method      => 'owner_queue_retrieve',
 810     api_level   => 1,
 811     argc        => 2,
 812     stream      => 1,
 813     record_type => 'bib'
 814 );
 815 __PACKAGE__->register_method(
 816     api_name    => "open-ils.vandelay.authority_queue.owner.retrieve",
 817     method      => 'owner_queue_retrieve',
 818     api_level   => 1,
 819     argc        => 2,
 820     stream      => 1,
 821     record_type => 'auth'
 822 );
 823
 824 sub owner_queue_retrieve {
 825     my($self, $conn, $auth, $owner_id, $filters) = @_;
 826     my $e = new_editor(authtoken => $auth, xact => 1);
 827     return $e->die_event unless $e->checkauth;
 828     $owner_id = $e->requestor->id; # XXX add support for viewing other's queues?
 829     my $queues;
 830     $filters ||= {};
 831     my $search = {owner => $owner_id};
 832     $search->{$_} = $filters->{$_} for keys %$filters;
 833
 834     if($self->{record_type} eq 'bib') {
 835         $queues = $e->search_vandelay_bib_queue(
 836             [$search, {order_by => {vbq => 'lower(name)'}}]);
 837     } else {
 838         $queues = $e->search_vandelay_authority_queue(
 839             [$search, {order_by => {vaq => 'lower(name)'}}]);
 840     }
 841     $conn->respond($_) for @$queues;
 842     $e->rollback;
 843     return undef;
 844 }
 845
 846 __PACKAGE__->register_method(
 847     api_name    => "open-ils.vandelay.bib_queue.delete",
 848     method      => "delete_queue",
 849     api_level   => 1,
 850     argc        => 2,
 851     record_type => 'bib'
 852 );
 853 __PACKAGE__->register_method(
 854     api_name    => "open-ils.vandelay.auth_queue.delete",
 855     method      => "delete_queue",
 856     api_level   => 1,
 857     argc        => 2,
 858     record_type => 'auth'
 859 );
 860
 861 sub delete_queue {
 862     my($self, $conn, $auth, $q_id) = @_;
 863     my $e = new_editor(xact => 1, authtoken => $auth);
 864     return $e->die_event unless $e->checkauth;
 865     if($self->{record_type} eq 'bib') {
 866         return $e->die_event unless $e->allowed('CREATE_BIB_IMPORT_QUEUE');
 867         my $queue = $e->retrieve_vandelay_bib_queue($q_id)
 868             or return $e->die_event;
 869         $e->delete_vandelay_bib_queue($queue)
 870             or return $e->die_event;
 871     } else {
 872            return $e->die_event unless $e->allowed('CREATE_AUTHORITY_IMPORT_QUEUE');
 873         my $queue = $e->retrieve_vandelay_authority_queue($q_id)
 874             or return $e->die_event;
 875         $e->delete_vandelay_authority_queue($queue)
 876             or return $e->die_event;
 877     }
 878     $e->commit;
 879     return 1;
 880 }
 881
 882
 883 __PACKAGE__->register_method(
 884     api_name    => "open-ils.vandelay.queued_bib_record.html",
 885     method      => 'queued_record_html',
 886     api_level   => 1,
 887     argc        => 2,
 888     stream      => 1,
 889     record_type => 'bib'
 890 );
 891 __PACKAGE__->register_method(
 892     api_name    => "open-ils.vandelay.queued_authority_record.html",
 893     method      => 'queued_record_html',
 894     api_level   => 1,
 895     argc        => 2,
 896     stream      => 1,
 897     record_type => 'auth'
 898 );
 899
 900 sub queued_record_html {
 901     my($self, $conn, $auth, $rec_id) = @_;
 902     my $e = new_editor(xact=>1,authtoken => $auth);
 903     return $e->die_event unless $e->checkauth;
 904     my $rec;
 905     if($self->{record_type} eq 'bib') {
 906         $rec = $e->retrieve_vandelay_queued_bib_record($rec_id)
 907             or return $e->die_event;
 908     } else {
 909         $rec = $e->retrieve_vandelay_queued_authority_record($rec_id)
 910             or return $e->die_event;
 911     }
 912
 913     $e->rollback;
 914     return $U->simplereq(
 915         'open-ils.search',
 916         'open-ils.search.biblio.record.html', undef, 1, $rec->marc);
 917 }
 918
 919
 920 __PACKAGE__->register_method(
 921     api_name    => "open-ils.vandelay.bib_queue.summary.retrieve",
 922     method      => 'retrieve_queue_summary',
 923     api_level   => 1,
 924     argc        => 2,
 925     stream      => 1,
 926     record_type => 'bib'
 927 );
 928 __PACKAGE__->register_method(
 929     api_name    => "open-ils.vandelay.auth_queue.summary.retrieve",
 930     method      => 'retrieve_queue_summary',
 931     api_level   => 1,
 932     argc        => 2,
 933     stream      => 1,
 934     record_type => 'auth'
 935 );
 936
 937 sub retrieve_queue_summary {
 938     my($self, $conn, $auth, $queue_id) = @_;
 939     my $e = new_editor(xact=>1, authtoken => $auth);
 940     return $e->die_event unless $e->checkauth;
 941
 942     my $queue;
 943     my $type = $self->{record_type};
 944     if($type eq 'bib') {
 945         $queue = $e->retrieve_vandelay_bib_queue($queue_id)
 946             or return $e->die_event;
 947     } else {
 948         $queue = $e->retrieve_vandelay_authority_queue($queue_id)
 949             or return $e->die_event;
 950     }
 951
 952     my $evt = check_queue_perms($e, $type, $queue);
 953     return $evt if $evt;
 954
 955     my $search = 'search_vandelay_queued_bib_record';
 956     $search =~ s/bib/authority/ if $type ne 'bib';
 957
 958     return {
 959         queue => $queue,
 960         total => scalar(@{$e->$search({queue => $queue_id}, {idlist=>1})}),
 961         imported => scalar(@{$e->$search({queue => $queue_id, import_time => {'!=' => undef}}, {idlist=>1})}),
 962     };
 963 }
 964
 965
 966 __PACKAGE__->register_method(
 967     api_name    => "open-ils.vandelay.bib_record.list.asset.import",
 968     method      => 'noop_import_items',
 969     api_level   => 1,
 970     argc        => 2,
 971     stream      => 1,
 972     record_type => 'bib'
 973 );
 974 __PACKAGE__->register_method(
 975     api_name    => "open-ils.vandelay.bib_record.queue.asset.import",
 976     method      => 'noop_import_items',
 977     api_level   => 1,
 978     argc        => 2,
 979     stream      => 1,
 980     record_type => 'bib'
 981 );
 982
 983 sub noop_import_items { return {complete => 1} }
 984
 985 #sub import_record_list_assets {
 986 #    my($self, $conn, $auth, $import_def, $rec_ids) = @_;
 987 #    my $e = new_editor(xact=>1, authtoken => $auth);
 988 #    return $e->die_event unless $e->checkauth;
 989 #    my $err = import_record_asset_list_impl($conn, $import_def, $rec_ids, $e->requestor);
 990 #    $e->rollback;
 991 #    return $err if $err;
 992 #    return {complete => 1};
 993 #}
 994 #
 995 #sub import_record_queue_assets {
 996 #    my($self, $conn, $auth, $import_def, $q_id) = @_;
 997 #    my $e = new_editor(xact=>1, authtoken => $auth);
 998 #    return $e->die_event unless $e->checkauth;
 999 #    my $rec_ids = $e->search_vandelay_queued_bib_record(
1000 #        {queue => $q_id, import_time => {'!=' => undef}}, {idlist => 1});
1001 #    my $err = import_record_asset_list_impl($conn, $import_def, $rec_ids, $e->requestor);
1002 #    $e->rollback;
1003 #    return $err if $err;
1004 #    return {complete => 1};
1005 #}
1006
1007 # --------------------------------------------------------------------------------
1008 # Given a list of queued record IDs, imports all items attached to those records
1009 # --------------------------------------------------------------------------------
1010 sub import_record_asset_list_impl {
1011     my($conn, $rec_ids, $requestor) = @_;
1012
1013     my $total = @$rec_ids;
1014     my $try_count = 0;
1015     my $in_count = 0;
1016     my $roe = new_editor(xact=> 1, requestor => $requestor);
1017
1018     for my $rec_id (@$rec_ids) {
1019         my $rec = $roe->retrieve_vandelay_queued_bib_record($rec_id);
1020         next unless $rec and $rec->import_time;
1021         my $item_ids = $roe->search_vandelay_import_item({record => $rec->id}, {idlist=>1});
1022
1023         for my $item_id (@$item_ids) {
1024             my $e = new_editor(requestor => $requestor, xact => 1);
1025             my $item = $e->retrieve_vandelay_import_item($item_id);
1026             $try_count++;
1027
1028             # --------------------------------------------------------------------------------
1029             # Find or create the volume
1030             # --------------------------------------------------------------------------------
1031             my ($vol, $evt) =
1032                 OpenILS::Application::Cat::AssetCommon->find_or_create_volume(
1033                     $e, $item->call_number, $rec->imported_as, $item->owning_lib);
1034
1035             if($evt) {
1036                 respond_with_status($conn, $total, $try_count, $in_count, $evt);
1037                 next;
1038             }
1039
1040             # --------------------------------------------------------------------------------
1041             # Create the new copy
1042             # --------------------------------------------------------------------------------
1043             my $copy = Fieldmapper::asset::copy->new;
1044             $copy->loan_duration(2);
1045             $copy->fine_level(2);
1046             $copy->barcode($item->barcode);
1047             $copy->location($item->location);
1048             $copy->circ_lib($item->circ_lib || $item->owning_lib);
1049             $copy->status($item->status || OILS_COPY_STATUS_IN_PROCESS);
1050             $copy->circulate($item->circulate);
1051             $copy->deposit($item->deposit);
1052             $copy->deposit_amount($item->deposit_amount);
1053             $copy->ref($item->ref);
1054             $copy->holdable($item->holdable);
1055             $copy->price($item->price);
1056             $copy->circ_as_type($item->circ_as_type);
1057             $copy->alert_message($item->alert_message);
1058             $copy->opac_visible($item->opac_visible);
1059             $copy->circ_modifier($item->circ_modifier);
1060
1061             # --------------------------------------------------------------------------------
1062             # see if a valid circ_modifier was provided
1063             # --------------------------------------------------------------------------------
1064             #if($copy->circ_modifier and not $e->retrieve_config_circ_modifier($item->circ_modifier)) {
1065             if($copy->circ_modifier and not $e->search_config_circ_modifier({code=>$item->circ_modifier})->[0]) {
1066                 respond_with_status($conn, $total, $try_count, $in_count, $e->die_event);
1067                 next;
1068             }
1069
1070             if($evt = OpenILS::Application::Cat::AssetCommon->create_copy($e, $vol, $copy)) {
1071                 try { $e->rollback } otherwise {}; # sometimes calls die_event, sometimes not
1072                 respond_with_status($conn, $total, $try_count, $in_count, $evt);
1073                 next;
1074             }
1075
1076             # --------------------------------------------------------------------------------
1077             # create copy notes
1078             # --------------------------------------------------------------------------------
1079             $evt = OpenILS::Application::Cat::AssetCommon->create_copy_note(
1080                 $e, $copy, '', $item->pub_note, 1) if $item->pub_note;
1081
1082             if($evt) {
1083                 respond_with_status($conn, $total, $try_count, $in_count, $evt);
1084                 next;
1085             }
1086
1087             $evt = OpenILS::Application::Cat::AssetCommon->create_copy_note(
1088                 $e, $copy, '', $item->priv_note, 1) if $item->priv_note;
1089
1090             if($evt) {
1091                 respond_with_status($conn, $total, $try_count, $in_count, $evt);
1092                 next;
1093             }
1094
1095             # --------------------------------------------------------------------------------
1096             # Item import succeeded
1097             # --------------------------------------------------------------------------------
1098             $e->commit;
1099             respond_with_status($conn, $total, $try_count, ++$in_count, undef, imported_as => $copy->id);
1100         }
1101     }
1102     $roe->rollback;
1103     return undef;
1104 }
1105
1106
1107 sub respond_with_status {
1108     my($conn, $total, $try_count, $success_count, $err, %args) = @_;
1109     $conn->respond({
1110         total => $total,
1111         progress => $try_count,
1112         err_event => $err,
1113         success_count => $success_count, %args }) if $err or ($try_count % 5 == 0);
1114 }
1115
1116
1117 1;