Open-ILS/src/perlmods/lib/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 OpenILS::Utils::Normalize qw/clean_marc/;
  13 use MARC::Batch;
  14 use MARC::Record;
  15 use MARC::File::XML ( BinaryEncoding => 'UTF-8' );
  16 use Time::HiRes qw(time);
  17 use OpenSRF::Utils::Logger qw/$logger/;
  18 use MIME::Base64;
  19 use XML::LibXML;
  20 use OpenILS::Const qw/:const/;
  21 use OpenILS::Application::AppUtils;
  22 use OpenILS::Application::Cat::BibCommon;
  23 use OpenILS::Application::Cat::AuthCommon;
  24 use OpenILS::Application::Cat::AssetCommon;
  25 use OpenILS::Application::Cat::Merge;
  26 my $U = 'OpenILS::Application::AppUtils';
  27
  28 # A list of LDR/06 values from http://loc.gov/marc
  29 my %record_types = (
  30         a => 'bib',
  31         c => 'bib',
  32         d => 'bib',
  33         e => 'bib',
  34         f => 'bib',
  35         g => 'bib',
  36         i => 'bib',
  37         j => 'bib',
  38         k => 'bib',
  39         m => 'bib',
  40         o => 'bib',
  41         p => 'bib',
  42         r => 'bib',
  43         t => 'bib',
  44         u => 'holdings',
  45         v => 'holdings',
  46         x => 'holdings',
  47         y => 'holdings',
  48         z => 'auth',
  49       ' ' => 'bib',
  50 );
  51
  52 sub initialize {}
  53 sub child_init {}
  54
  55 # --------------------------------------------------------------------------------
  56 # Biblio ingest
  57
  58 sub create_bib_queue {
  59     my $self = shift;
  60     my $client = shift;
  61     my $auth = shift;
  62     my $name = shift;
  63     my $owner = shift;
  64     my $type = shift;
  65     my $match_set = shift;
  66     my $import_def = shift;
  67
  68     my $e = new_editor(authtoken => $auth, xact => 1);
  69
  70     return $e->die_event unless $e->checkauth;
  71     return $e->die_event unless $e->allowed('CREATE_BIB_IMPORT_QUEUE');
  72     $owner ||= $e->requestor->id;
  73
  74     if ($e->search_vandelay_bib_queue( {name => $name, owner => $owner, queue_type => $type})->[0]) {
  75         $e->rollback;
  76         return OpenILS::Event->new('BIB_QUEUE_EXISTS')
  77     }
  78
  79     my $queue = new Fieldmapper::vandelay::bib_queue();
  80     $queue->name( $name );
  81     $queue->owner( $owner );
  82     $queue->queue_type( $type ) if ($type);
  83     $queue->item_attr_def( $import_def ) if ($import_def);
  84     $queue->match_set($match_set) if $match_set;
  85
  86     my $new_q = $e->create_vandelay_bib_queue( $queue );
  87     return $e->die_event unless ($new_q);
  88     $e->commit;
  89
  90     return $new_q;
  91 }
  92 __PACKAGE__->register_method(
  93     api_name   => "open-ils.vandelay.bib_queue.create",
  94     method     => "create_bib_queue",
  95     api_level  => 1,
  96     argc       => 4,
  97 );
  98
  99
 100 sub create_auth_queue {
 101     my $self = shift;
 102     my $client = shift;
 103     my $auth = shift;
 104     my $name = shift;
 105     my $owner = shift;
 106     my $type = shift;
 107     my $match_set = shift;
 108
 109     my $e = new_editor(authtoken => $auth, xact => 1);
 110
 111     return $e->die_event unless $e->checkauth;
 112     return $e->die_event unless $e->allowed('CREATE_AUTHORITY_IMPORT_QUEUE');
 113     $owner ||= $e->requestor->id;
 114
 115     if ($e->search_vandelay_authority_queue({name => $name, owner => $owner, queue_type => $type})->[0]) {
 116         $e->rollback;
 117         return OpenILS::Event->new('AUTH_QUEUE_EXISTS')
 118     }
 119
 120     my $queue = new Fieldmapper::vandelay::authority_queue();
 121     $queue->name( $name );
 122     $queue->owner( $owner );
 123     $queue->queue_type( $type ) if ($type);
 124
 125     my $new_q = $e->create_vandelay_authority_queue( $queue );
 126     $e->die_event unless ($new_q);
 127     $e->commit;
 128
 129     return $new_q;
 130 }
 131 __PACKAGE__->register_method(
 132     api_name   => "open-ils.vandelay.authority_queue.create",
 133     method     => "create_auth_queue",
 134     api_level  => 1,
 135     argc       => 3,
 136 );
 137
 138 sub add_record_to_bib_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     my $bib_source = shift;
 146
 147     my $e = new_editor(authtoken => $auth, xact => 1);
 148
 149     $queue = $e->retrieve_vandelay_bib_queue($queue);
 150
 151     return $e->die_event unless $e->checkauth;
 152     return $e->die_event unless
 153         ($e->allowed('CREATE_BIB_IMPORT_QUEUE', undef, $queue) ||
 154          $e->allowed('CREATE_BIB_IMPORT_QUEUE'));
 155
 156     my $new_rec = _add_bib_rec($e, $marc, $queue->id, $purpose, $bib_source);
 157
 158     return $e->die_event unless ($new_rec);
 159     $e->commit;
 160     return $new_rec;
 161 }
 162 __PACKAGE__->register_method(
 163     api_name   => "open-ils.vandelay.queued_bib_record.create",
 164     method     => "add_record_to_bib_queue",
 165     api_level  => 1,
 166     argc       => 3,
 167 );
 168
 169 sub _add_bib_rec {
 170     my $e = shift;
 171     my $marc = shift;
 172     my $queue = shift;
 173     my $purpose = shift;
 174     my $bib_source = shift;
 175
 176     my $rec = new Fieldmapper::vandelay::queued_bib_record();
 177     $rec->marc( $marc );
 178     $rec->queue( $queue );
 179     $rec->purpose( $purpose ) if ($purpose);
 180     $rec->bib_source($bib_source);
 181
 182     return $e->create_vandelay_queued_bib_record( $rec, {timeout => 600} );
 183 }
 184
 185 sub add_record_to_authority_queue {
 186     my $self = shift;
 187     my $client = shift;
 188     my $auth = shift;
 189     my $queue = shift;
 190     my $marc = shift;
 191     my $purpose = shift;
 192
 193     my $e = new_editor(authtoken => $auth, xact => 1);
 194
 195     $queue = $e->retrieve_vandelay_authority_queue($queue);
 196
 197     return $e->die_event unless $e->checkauth;
 198     return $e->die_event unless
 199         ($e->allowed('CREATE_AUTHORITY_IMPORT_QUEUE', undef, $queue) ||
 200          $e->allowed('CREATE_AUTHORITY_IMPORT_QUEUE'));
 201
 202     my $new_rec = _add_auth_rec($e, $marc, $queue->id, $purpose);
 203
 204     return $e->die_event unless ($new_rec);
 205     $e->commit;
 206     return $new_rec;
 207 }
 208 __PACKAGE__->register_method(
 209     api_name   => "open-ils.vandelay.queued_authority_record.create",
 210     method     => "add_record_to_authority_queue",
 211     api_level  => 1,
 212     argc       => 3,
 213 );
 214
 215 sub _add_auth_rec {
 216     my $e = shift;
 217     my $marc = shift;
 218     my $queue = shift;
 219     my $purpose = shift;
 220
 221     my $rec = new Fieldmapper::vandelay::queued_authority_record();
 222     $rec->marc( $marc );
 223     $rec->queue( $queue );
 224     $rec->purpose( $purpose ) if ($purpose);
 225
 226     return $e->create_vandelay_queued_authority_record( $rec, {timeout => 600} );
 227 }
 228
 229 sub process_spool {
 230     my $self = shift;
 231     my $client = shift;
 232     my $auth = shift;
 233     my $fingerprint = shift || '';
 234     my $queue_id = shift;
 235     my $purpose = shift;
 236     my $filename = shift;
 237     my $bib_source = shift;
 238
 239     my $e = new_editor(authtoken => $auth, xact => 1);
 240     return $e->die_event unless $e->checkauth;
 241
 242     my $queue;
 243     my $type = $self->{record_type};
 244
 245     if($type eq 'bib') {
 246         $queue = $e->retrieve_vandelay_bib_queue($queue_id) or return $e->die_event;
 247     } else {
 248         $queue = $e->retrieve_vandelay_authority_queue($queue_id) or return $e->die_event;
 249     }
 250
 251     my $evt = check_queue_perms($e, $type, $queue);
 252     return $evt if ($evt);
 253
 254     my $cache = new OpenSRF::Utils::Cache();
 255
 256     if($fingerprint) {
 257         my $data = $cache->get_cache('vandelay_import_spool_' . $fingerprint);
 258         $purpose = $data->{purpose};
 259         $filename = $data->{path};
 260         $bib_source = $data->{bib_source};
 261     }
 262
 263     unless(-r $filename) {
 264         $logger->error("unable to read MARC file $filename");
 265         return -1; # make this an event XXX
 266     }
 267
 268     $logger->info("vandelay spooling $fingerprint purpose=$purpose file=$filename");
 269
 270     my $marctype = 'USMARC';
 271
 272     open F, $filename;
 273     $marctype = 'XML' if (getc(F) =~ /^\D/o);
 274     close F;
 275
 276     my $batch = new MARC::Batch ($marctype, $filename);
 277     $batch->strict_off;
 278
 279     my $response_scale = 10;
 280     my $count = 0;
 281     my $r = -1;
 282     while (try { $r = $batch->next } otherwise { $r = -1 }) {
 283         if ($r == -1) {
 284             $logger->warn("Processing of record $count in set $filename failed.  Skipping this record");
 285             $count++;
 286         }
 287
 288         $logger->info("processing record $count");
 289
 290         try {
 291             my $xml = clean_marc($r);
 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     max_chunk_size => 0,
 331     record_type => 'bib'
 332 );
 333 __PACKAGE__->register_method(
 334     api_name    => "open-ils.vandelay.auth.process_spool",
 335     method      => "process_spool",
 336     api_level   => 1,
 337     argc        => 3,
 338     max_chunk_size => 0,
 339     record_type => 'auth'
 340 );
 341
 342 __PACKAGE__->register_method(
 343     api_name    => "open-ils.vandelay.bib.process_spool.stream_results",
 344     method      => "process_spool",
 345     api_level   => 1,
 346     argc        => 3,
 347     stream      => 1,
 348     max_chunk_size => 0,
 349     record_type => 'bib'
 350 );
 351 __PACKAGE__->register_method(
 352     api_name    => "open-ils.vandelay.auth.process_spool.stream_results",
 353     method      => "process_spool",
 354     api_level   => 1,
 355     argc        => 3,
 356     stream      => 1,
 357     max_chunk_size => 0,
 358     record_type => 'auth'
 359 );
 360
 361 __PACKAGE__->register_method(
 362     api_name    => "open-ils.vandelay.bib_queue.records.retrieve",
 363     method      => 'retrieve_queued_records',
 364     api_level   => 1,
 365     argc        => 2,
 366     stream      => 1,
 367     record_type => 'bib'
 368 );
 369 __PACKAGE__->register_method(
 370     api_name    => "open-ils.vandelay.bib_queue.records.retrieve.export.print",
 371     method      => 'retrieve_queued_records',
 372     api_level   => 1,
 373     argc        => 2,
 374     stream      => 1,
 375     record_type => 'bib'
 376 );
 377 __PACKAGE__->register_method(
 378     api_name    => "open-ils.vandelay.bib_queue.records.retrieve.export.csv",
 379     method      => 'retrieve_queued_records',
 380     api_level   => 1,
 381     argc        => 2,
 382     stream      => 1,
 383     record_type => 'bib'
 384 );
 385 __PACKAGE__->register_method(
 386     api_name    => "open-ils.vandelay.bib_queue.records.retrieve.export.email",
 387     method      => 'retrieve_queued_records',
 388     api_level   => 1,
 389     argc        => 2,
 390     stream      => 1,
 391     record_type => 'bib'
 392 );
 393
 394 __PACKAGE__->register_method(
 395     api_name    => "open-ils.vandelay.auth_queue.records.retrieve",
 396     method      => 'retrieve_queued_records',
 397     api_level   => 1,
 398     argc        => 2,
 399     stream      => 1,
 400     record_type => 'auth'
 401 );
 402 __PACKAGE__->register_method(
 403     api_name    => "open-ils.vandelay.auth_queue.records.retrieve.export.print",
 404     method      => 'retrieve_queued_records',
 405     api_level   => 1,
 406     argc        => 2,
 407     stream      => 1,
 408     record_type => 'auth'
 409 );
 410 __PACKAGE__->register_method(
 411     api_name    => "open-ils.vandelay.auth_queue.records.retrieve.export.csv",
 412     method      => 'retrieve_queued_records',
 413     api_level   => 1,
 414     argc        => 2,
 415     stream      => 1,
 416     record_type => 'auth'
 417 );
 418 __PACKAGE__->register_method(
 419     api_name    => "open-ils.vandelay.auth_queue.records.retrieve.export.email",
 420     method      => 'retrieve_queued_records',
 421     api_level   => 1,
 422     argc        => 2,
 423     stream      => 1,
 424     record_type => 'auth'
 425 );
 426
 427 __PACKAGE__->register_method(
 428     api_name    => "open-ils.vandelay.bib_queue.records.matches.retrieve",
 429     method      => 'retrieve_queued_records',
 430     api_level   => 1,
 431     argc        => 2,
 432     stream      => 1,
 433     record_type => 'bib',
 434     signature   => {
 435         desc => q/Only retrieve queued bib records that have matches against existing records/
 436     }
 437 );
 438 __PACKAGE__->register_method(
 439     api_name    => "open-ils.vandelay.auth_queue.records.matches.retrieve",
 440     method      => 'retrieve_queued_records',
 441     api_level   => 1,
 442     argc        => 2,
 443     stream      => 1,
 444     record_type => 'auth',
 445     signature   => {
 446         desc => q/Only retrieve queued authority records that have matches against existing records/
 447     }
 448 );
 449
 450 sub retrieve_queued_records {
 451     my($self, $conn, $auth, $queue_id, $options) = @_;
 452
 453     $options ||= {};
 454     my $limit = $$options{limit} || 20;
 455     my $offset = $$options{offset} || 0;
 456     my $type = $self->{record_type};
 457
 458     my $e = new_editor(authtoken => $auth, xact => 1);
 459     return $e->die_event unless $e->checkauth;
 460
 461     my $queue;
 462     if($type eq 'bib') {
 463         $queue = $e->retrieve_vandelay_bib_queue($queue_id) or return $e->die_event;
 464     } else {
 465         $queue = $e->retrieve_vandelay_authority_queue($queue_id) or return $e->die_event;
 466     }
 467     my $evt = check_queue_perms($e, $type, $queue);
 468     return $evt if ($evt);
 469
 470     my $class = ($type eq 'bib') ? 'vqbr' : 'vqar';
 471     my $query = {
 472         select => {$class => ['id']},
 473         from => $class,
 474         where => {queue => $queue_id},
 475         distinct => 1,
 476         order_by => {$class => ['id']},
 477         limit => $limit,
 478         offset => $offset,
 479     };
 480     if($self->api_name =~ /export/) {
 481         delete $query->{limit};
 482         delete $query->{offset};
 483     }
 484
 485     $query->{where}->{import_time} = undef if $$options{non_imported};
 486
 487     if($$options{with_import_error}) {
 488
 489         $query->{from} = {$class => {vii => {type => 'left'}}};
 490         $query->{where}->{'-or'} = [
 491             {'+vqbr' => {import_error => {'!=' => undef}}},
 492             {'+vii' => {import_error => {'!=' => undef}}}
 493         ];
 494
 495     } else {
 496
 497         if($$options{with_rec_import_error}) {
 498             $query->{where}->{import_error} = {'!=' => undef};
 499
 500         } elsif( $$options{with_item_import_error} and $type eq 'bib') {
 501
 502             $query->{from} = {$class => 'vii'};
 503             $query->{where}->{'+vii'} = {import_error => {'!=' => undef}};
 504         }
 505     }
 506
 507     if($self->api_name =~ /matches/) {
 508         # find only records that have matches
 509         my $mclass = $type eq 'bib' ? 'vbm' : 'vam';
 510         $query->{from} = {$class => {$mclass => {type => 'right'}}};
 511     }
 512
 513     my $record_ids = $e->json_query($query);
 514
 515     my $retrieve = ($type eq 'bib') ?
 516         'retrieve_vandelay_queued_bib_record' : 'retrieve_vandelay_queued_authority_record';
 517     my $search = ($type eq 'bib') ?
 518         'search_vandelay_queued_bib_record' : 'search_vandelay_queued_authority_record';
 519
 520     if ($self->api_name =~ /export/) {
 521         my $rec_list = $e->$search({id => [map { $_->{id} } @$record_ids]}, {substream => 1});
 522         if ($self->api_name =~ /print/) {
 523
 524             $e->rollback;
 525             return $U->fire_object_event(
 526                 undef,
 527                 'vandelay.queued_'.$type.'_record.print',
 528                 $rec_list,
 529                 $e->requestor->ws_ou
 530             );
 531
 532         } elsif ($self->api_name =~ /csv/) {
 533
 534             $e->rollback;
 535             return $U->fire_object_event(
 536                 undef,
 537                 'vandelay.queued_'.$type.'_record.csv',
 538                 $rec_list,
 539                 $e->requestor->ws_ou
 540             );
 541
 542         } elsif ($self->api_name =~ /email/) {
 543
 544             $conn->respond_complete(1);
 545
 546             for my $rec (@$rec_list) {
 547                 $U->create_events_for_hook(
 548                     'vandelay.queued_'.$type.'_record.email',
 549                     $rec,
 550                     $e->requestor->home_ou,
 551                     undef,
 552                     undef,
 553                     1
 554                 );
 555             }
 556
 557         }
 558     } else {
 559         for my $rec_id (@$record_ids) {
 560             my $flesh = ['attributes', 'matches'];
 561             push(@$flesh, 'import_items') if $$options{flesh_import_items};
 562             my $params = {flesh => 1, flesh_fields => {$class => $flesh}};
 563             my $rec = $e->$retrieve([$rec_id->{id}, $params]);
 564             $rec->clear_marc if $$options{clear_marc};
 565             $conn->respond($rec);
 566         }
 567     }
 568
 569     $e->rollback;
 570     return undef;
 571 }
 572
 573 __PACKAGE__->register_method(
 574     api_name    => 'open-ils.vandelay.import_item.queue.retrieve',
 575     method      => 'retrieve_queue_import_items',
 576     api_level   => 1,
 577     argc        => 2,
 578     stream      => 1,
 579     authoritative => 1,
 580     signature => q/
 581         Returns Import Item (vii) objects for the selected queue.
 582         Filter options:
 583             with_import_error : only return items that failed to import
 584     /
 585 );
 586 __PACKAGE__->register_method(
 587     api_name    => 'open-ils.vandelay.import_item.queue.export.print',
 588     method      => 'retrieve_queue_import_items',
 589     api_level   => 1,
 590     argc        => 2,
 591     stream      => 1,
 592     authoritative => 1,
 593     signature => q/
 594         Returns template-generated printable output of Import Item (vii) objects for the selected queue.
 595         Filter options:
 596             with_import_error : only return items that failed to import
 597     /
 598 );
 599 __PACKAGE__->register_method(
 600     api_name    => 'open-ils.vandelay.import_item.queue.export.csv',
 601     method      => 'retrieve_queue_import_items',
 602     api_level   => 1,
 603     argc        => 2,
 604     stream      => 1,
 605     authoritative => 1,
 606     signature => q/
 607         Returns template-generated CSV output of Import Item (vii) objects for the selected queue.
 608         Filter options:
 609             with_import_error : only return items that failed to import
 610     /
 611 );
 612 __PACKAGE__->register_method(
 613     api_name    => 'open-ils.vandelay.import_item.queue.export.email',
 614     method      => 'retrieve_queue_import_items',
 615     api_level   => 1,
 616     argc        => 2,
 617     stream      => 1,
 618     authoritative => 1,
 619     signature => q/
 620         Emails template-generated output of Import Item (vii) objects for the selected queue.
 621         Filter options:
 622             with_import_error : only return items that failed to import
 623     /
 624 );
 625
 626 sub retrieve_queue_import_items {
 627     my($self, $conn, $auth, $q_id, $options) = @_;
 628
 629     $options ||= {};
 630     my $limit = $$options{limit} || 20;
 631     my $offset = $$options{offset} || 0;
 632
 633     my $e = new_editor(authtoken => $auth);
 634     return $e->event unless $e->checkauth;
 635
 636     my $queue = $e->retrieve_vandelay_bib_queue($q_id) or return $e->event;
 637     my $evt = check_queue_perms($e, 'bib', $queue);
 638     return $evt if $evt;
 639
 640     my $query = {
 641         select => {vii => ['id']},
 642         from => {
 643             vii => {
 644                 vqbr => {
 645                     join => {
 646                         'vbq' => {
 647                             field => 'id',
 648                             fkey => 'queue',
 649                             filter => {id => $q_id}
 650                         }
 651                     }
 652                 }
 653             }
 654         },
 655         order_by => {'vii' => ['record','id']},
 656         limit => $limit,
 657         offset => $offset
 658     };
 659     if($self->api_name =~ /export/) {
 660         delete $query->{limit};
 661         delete $query->{offset};
 662     }
 663
 664     $query->{where} = {'+vii' => {import_error => {'!=' => undef}}}
 665         if $$options{with_import_error};
 666
 667     my $items = $e->json_query($query);
 668     my $item_list = $e->search_vandelay_import_item({id => [map { $_->{id} } @$items]});
 669     if ($self->api_name =~ /export/) {
 670         if ($self->api_name =~ /print/) {
 671
 672             return $U->fire_object_event(
 673                 undef,
 674                 'vandelay.import_items.print',
 675                 $item_list,
 676                 $e->requestor->ws_ou
 677             );
 678
 679         } elsif ($self->api_name =~ /csv/) {
 680
 681             return $U->fire_object_event(
 682                 undef,
 683                 'vandelay.import_items.csv',
 684                 $item_list,
 685                 $e->requestor->ws_ou
 686             );
 687
 688         } elsif ($self->api_name =~ /email/) {
 689
 690             $conn->respond_complete(1);
 691
 692             for my $item (@$item_list) {
 693                 $U->create_events_for_hook(
 694                     'vandelay.import_items.email',
 695                     $item,
 696                     $e->requestor->home_ou,
 697                     undef,
 698                     undef,
 699                     1
 700                 );
 701             }
 702
 703         }
 704     } else {
 705         for my $item (@$item_list) {
 706             $conn->respond($item);
 707         }
 708     }
 709
 710     return undef;
 711 }
 712
 713 sub check_queue_perms {
 714     my($e, $type, $queue) = @_;
 715     if ($type eq 'bib') {
 716         return $e->die_event unless
 717             ($e->allowed('CREATE_BIB_IMPORT_QUEUE', undef, $queue) ||
 718              $e->allowed('CREATE_BIB_IMPORT_QUEUE'));
 719     } else {
 720         return $e->die_event unless
 721             ($e->allowed('CREATE_AUTHORITY_IMPORT_QUEUE', undef, $queue) ||
 722              $e->allowed('CREATE_AUTHORITY_IMPORT_QUEUE'));
 723     }
 724
 725     return undef;
 726 }
 727
 728 __PACKAGE__->register_method(
 729     api_name    => "open-ils.vandelay.bib_record.list.import",
 730     method      => 'import_record_list',
 731     api_level   => 1,
 732     argc        => 2,
 733     stream      => 1,
 734     record_type => 'bib'
 735 );
 736
 737 __PACKAGE__->register_method(
 738     api_name    => "open-ils.vandelay.auth_record.list.import",
 739     method      => 'import_record_list',
 740     api_level   => 1,
 741     argc        => 2,
 742     stream      => 1,
 743     record_type => 'auth'
 744 );
 745
 746 sub import_record_list {
 747     my($self, $conn, $auth, $rec_ids, $args) = @_;
 748     my $e = new_editor(authtoken => $auth, xact => 1);
 749     return $e->die_event unless $e->checkauth;
 750     $args ||= {};
 751     my $err = import_record_list_impl($self, $conn, $rec_ids, $e->requestor, $args);
 752     try {$e->rollback} otherwise {};
 753     return $err if $err;
 754     return {complete => 1};
 755 }
 756
 757
 758 __PACKAGE__->register_method(
 759     api_name    => "open-ils.vandelay.bib_queue.import",
 760     method      => 'import_queue',
 761     api_level   => 1,
 762     argc        => 2,
 763     stream      => 1,
 764     max_chunk_size => 0,
 765     record_type => 'bib',
 766     signature => {
 767         desc => q/
 768             Attempts to import all non-imported records for the selected queue.
 769             Will also attempt import of all non-imported items.
 770         /
 771     }
 772 );
 773
 774 __PACKAGE__->register_method(
 775     api_name    => "open-ils.vandelay.auth_queue.import",
 776     method      => 'import_queue',
 777     api_level   => 1,
 778     argc        => 2,
 779     stream      => 1,
 780     max_chunk_size => 0,
 781     record_type => 'auth'
 782 );
 783
 784 sub import_queue {
 785     my($self, $conn, $auth, $q_id, $options) = @_;
 786     my $e = new_editor(authtoken => $auth, xact => 1);
 787     return $e->die_event unless $e->checkauth;
 788     $options ||= {};
 789     my $type = $self->{record_type};
 790     my $class = ($type eq 'bib') ? 'vqbr' : 'vqar';
 791
 792     # First, collect the not-yet-imported records
 793     my $query = {queue => $q_id, import_time => undef};
 794     my $search = ($type eq 'bib') ?
 795         'search_vandelay_queued_bib_record' :
 796         'search_vandelay_queued_authority_record';
 797     my $rec_ids = $e->$search($query, {idlist => 1});
 798
 799     # Now add any imported records that have un-imported items
 800
 801     if($type eq 'bib') {
 802         my $item_recs = $e->json_query({
 803             select => {vqbr => ['id']},
 804             from => {vqbr => 'vii'},
 805             where => {
 806                 '+vqbr' => {
 807                     queue => $q_id,
 808                     import_time => {'!=' => undef}
 809                 },
 810                 '+vii' => {import_time => undef}
 811             },
 812             distinct => 1
 813         });
 814         push(@$rec_ids, map {$_->{id}} @$item_recs);
 815     }
 816
 817     my $err = import_record_list_impl($self, $conn, $rec_ids, $e->requestor, $options);
 818     try {$e->rollback} otherwise {}; # only using this to make the read authoritative -- don't die from it
 819     return $err if $err;
 820     return {complete => 1};
 821 }
 822
 823 # returns a list of queued record IDs for a given queue that
 824 # have at least one entry in the match table
 825 # XXX DEPRECATED?
 826 sub queued_records_with_matches {
 827     my($e, $type, $q_id, $limit, $offset, $filter) = @_;
 828
 829     my $match_class = 'vbm';
 830     my $rec_class = 'vqbr';
 831     if($type eq 'auth') {
 832         $match_class = 'vam';
 833          $rec_class = 'vqar';
 834     }
 835
 836     $filter ||= {};
 837     $filter->{queue} = $q_id;
 838
 839     my $query = {
 840         distinct => 1,
 841         select => {$match_class => ['queued_record']},
 842         from => {
 843             $match_class => {
 844                 $rec_class => {
 845                     field => 'id',
 846                     fkey => 'queued_record',
 847                     filter => $filter,
 848                 }
 849             }
 850         }
 851     };
 852
 853     if($limit or defined $offset) {
 854         $limit ||= 20;
 855         $offset ||= 0;
 856         $query->{limit} = $limit;
 857         $query->{offset} = $offset;
 858     }
 859
 860     my $data = $e->json_query($query);
 861     return [ map {$_->{queued_record}} @$data ];
 862 }
 863
 864
 865 sub import_record_list_impl {
 866     my($self, $conn, $rec_ids, $requestor, $args) = @_;
 867
 868     my $overlay_map = $args->{overlay_map} || {};
 869     my $type = $self->{record_type};
 870     my %queues;
 871
 872     my $report_args = {
 873         progress => 1,
 874         step => 1,
 875         conn => $conn,
 876         total => scalar(@$rec_ids),
 877         report_all => $$args{report_all}
 878     };
 879
 880     $conn->max_chunk_count(1) if $$args{report_all};
 881
 882     my $auto_overlay_exact = $$args{auto_overlay_exact};
 883     my $auto_overlay_1match = $$args{auto_overlay_1match};
 884     my $auto_overlay_best = $$args{auto_overlay_best_match};
 885     my $match_quality_ratio = $$args{match_quality_ratio};
 886     my $merge_profile = $$args{merge_profile};
 887     my $ft_merge_profile = $$args{fall_through_merge_profile};
 888     my $bib_source = $$args{bib_source};
 889     my $import_no_match = $$args{import_no_match};
 890     my $strip_grps = $$args{strip_field_groups}; # bib-only
 891
 892     my $overlay_func = 'vandelay.overlay_bib_record';
 893     my $auto_overlay_func = 'vandelay.auto_overlay_bib_record';
 894     my $auto_overlay_best_func = 'vandelay.auto_overlay_bib_record_with_best'; # XXX bib-only
 895     my $retrieve_func = 'retrieve_vandelay_queued_bib_record';
 896     my $update_func = 'update_vandelay_queued_bib_record';
 897     my $search_func = 'search_vandelay_queued_bib_record';
 898     my $retrieve_queue_func = 'retrieve_vandelay_bib_queue';
 899     my $update_queue_func = 'update_vandelay_bib_queue';
 900     my $delete_queue_func = 'delete_vandelay_bib_queue';
 901     my $rec_class = 'vqbr';
 902
 903     my $editor = new_editor();
 904
 905     my %bib_sources;
 906     my $sources = $editor->search_config_bib_source({id => {'!=' => undef}});
 907     $bib_sources{$_->id} = $_->source for @$sources;
 908
 909     if($type eq 'auth') {
 910         $overlay_func =~ s/bib/authority/o;
 911         $auto_overlay_func =~ s/bib/authority/o;
 912         $retrieve_func =~ s/bib/authority/o;
 913         $retrieve_queue_func =~ s/bib/authority/o;
 914         $update_queue_func =~ s/bib/authority/o;
 915         $update_func =~ s/bib/authority/o;
 916         $search_func =~ s/bib/authority/o;
 917         $delete_queue_func =~ s/bib/authority/o;
 918         $rec_class = 'vqar';
 919     }
 920
 921     my $new_rec_perm_cache;
 922     my @success_rec_ids;
 923     for my $rec_id (@$rec_ids) {
 924
 925         my $error = 0;
 926         my $overlay_target = $overlay_map->{$rec_id};
 927
 928         my $e = new_editor(xact => 1);
 929         $e->requestor($requestor);
 930
 931         $$report_args{e} = $e;
 932         $$report_args{evt} = undef;
 933         $$report_args{import_error} = undef;
 934         $$report_args{no_import} = 0;
 935
 936         my $rec = $e->$retrieve_func([
 937             $rec_id,
 938             {   flesh => 1,
 939                 flesh_fields => { $rec_class => ['matches']},
 940             }
 941         ]);
 942
 943         unless($rec) {
 944             $$report_args{evt} = $e->event;
 945             finish_rec_import_attempt($report_args);
 946             next;
 947         }
 948
 949         if($rec->import_time) {
 950             # if the record is already imported, that means it may have
 951             # un-imported copies.  Add to success list for later processing.
 952             push(@success_rec_ids, $rec_id);
 953             $e->rollback;
 954             next;
 955         }
 956
 957         $$report_args{rec} = $rec;
 958         $queues{$rec->queue} = 1;
 959
 960         my $record;
 961         my $imported = 0;
 962
 963         if ($type eq 'bib') {
 964             # strip configured / selected MARC tags from inbound records
 965
 966             my $marcdoc = XML::LibXML->new->parse_string($rec->marc);
 967             $rec->marc($U->strip_marc_fields($e, $marcdoc, $strip_grps));
 968
 969             unless ($e->$update_func($rec)) {
 970                 $$report_args{evt} = $e->die_event;
 971                 finish_rec_import_attempt($report_args);
 972                 next;
 973             }
 974         }
 975
 976         if(defined $overlay_target) {
 977             # Caller chose an explicit overlay target
 978
 979             my $res = $e->json_query(
 980                 {
 981                     from => [
 982                         $overlay_func,
 983                         $rec_id,
 984                         $overlay_target,
 985                         $merge_profile
 986                     ]
 987                 }
 988             );
 989
 990             if($res and ($res = $res->[0])) {
 991
 992                 if($res->{$overlay_func} eq 't') {
 993                     $logger->info("vl: $type direct overlay succeeded for queued rec ".
 994                         "$rec_id and overlay target $overlay_target");
 995                     $imported = 1;
 996                 }
 997
 998             } else {
 999                 $error = 1;
1000                 $logger->error("vl: Error attempting overlay with func=$overlay_func, profile=$merge_profile, record=$rec_id");
1001             }
1002
1003         } else {
1004
1005             if($auto_overlay_1match) { # overlay if there is exactly 1 match
1006
1007                 my %match_recs = map { $_->eg_record => 1 } @{$rec->matches};
1008
1009                 if( scalar(keys %match_recs) == 1) { # all matches point to the same record
1010
1011                     ($imported, $error, $rec) = try_auto_overlay(
1012                         $e, $type,
1013                         $report_args,
1014                         $auto_overlay_best_func,
1015                         $retrieve_func,
1016                         $rec_class,
1017                         $rec_id,
1018                         $match_quality_ratio,
1019                         $merge_profile,
1020                         $ft_merge_profile
1021                     );
1022                 }
1023             }
1024
1025             if(!$imported and !$error and $auto_overlay_exact and scalar(@{$rec->matches}) == 1 ) {
1026
1027                 # caller says to overlay if there is an /exact/ match
1028                 # $auto_overlay_func only proceeds and returns true on exact matches
1029
1030                 my $res = $e->json_query(
1031                     {
1032                         from => [
1033                             $auto_overlay_func,
1034                             $rec_id,
1035                             $merge_profile
1036                         ]
1037                     }
1038                 );
1039
1040                 if($res and ($res = $res->[0])) {
1041
1042                     if($res->{$auto_overlay_func} eq 't') {
1043                         $logger->info("vl: $type auto-overlay succeeded for queued rec $rec_id");
1044                         $imported = 1;
1045
1046                         # re-fetch the record to pick up the imported_as value from the DB
1047                         $$report_args{rec} = $rec = $e->$retrieve_func([
1048                             $rec_id, {flesh => 1, flesh_fields => {$rec_class => ['matches']}}]);
1049
1050                     } else {
1051                         $logger->info("vl: $type auto-overlay failed for queued rec $rec_id");
1052                     }
1053
1054                 } else {
1055                     $error = 1;
1056                     $logger->error("vl: Error attempting overlay with func=$auto_overlay_func, profile=$merge_profile, record=$rec_id");
1057                 }
1058             }
1059
1060             if(!$imported and !$error and $auto_overlay_best and scalar(@{$rec->matches}) > 0 ) {
1061                 # caller says to overlay the best match
1062
1063                 ($imported, $error, $rec) = try_auto_overlay(
1064                     $e, $type,
1065                     $report_args,
1066                     $auto_overlay_best_func,
1067                     $retrieve_func,
1068                     $rec_class,
1069                     $rec_id,
1070                     $match_quality_ratio,
1071                     $merge_profile,
1072                     $ft_merge_profile
1073                 );
1074             }
1075
1076             if(!$imported and !$error and $import_no_match and scalar(@{$rec->matches}) == 0) {
1077
1078                 # No overlay / merge occurred.  Do a traditional record import by creating a new record
1079
1080                 if (!$new_rec_perm_cache) {
1081                     $new_rec_perm_cache = {};
1082
1083                     # all users creating new records are required to have the basic permission.
1084                     # if the client requests, we can enforce extra permissions for creating new records.
1085                     # for speed, check the permissions the first time then cache the result.
1086
1087                     my $perm = ($type eq 'bib') ? 'IMPORT_MARC' : 'IMPORT_AUTHORITY_MARC';
1088                     my $xperm = $$args{new_rec_perm};
1089                     my $rec_ou = $e->requestor->ws_ou;
1090
1091                     $new_rec_perm_cache->{evt} = $e->die_event
1092                         if !$e->allowed($perm, $rec_ou) || ($xperm and !$e->allowed($xperm, $rec_ou));
1093                 }
1094
1095                 if ($new_rec_perm_cache->{evt}) {
1096
1097                     # a cached event won't roll back the transaction (a la die_event), but
1098                     # the transaction will get rolled back in finish_rec_import_attempt() below
1099                     $$report_args{evt} = $new_rec_perm_cache->{evt};
1100                     $$report_args{import_error} = 'import.record.perm_failure';
1101
1102                 } else { # perm checks succeeded
1103
1104                     $logger->info("vl: creating new $type record for queued record $rec_id");
1105
1106                     if ($type eq 'bib') {
1107
1108                         $record = OpenILS::Application::Cat::BibCommon->biblio_record_xml_import(
1109                             $e, $rec->marc, $bib_sources{$rec->bib_source}, undef, 1);
1110
1111                     } else { # authority record
1112
1113                         $record = OpenILS::Application::Cat::AuthCommon->import_authority_record($e, $rec->marc); #$source);
1114                     }
1115
1116                     if($U->event_code($record)) {
1117                         $$report_args{import_error} = 'import.duplicate.tcn'
1118                             if $record->{textcode} eq 'OPEN_TCN_NOT_FOUND';
1119                         $$report_args{evt} = $record;
1120
1121                     } else {
1122
1123                         $logger->info("vl: successfully imported new $type record");
1124                         $rec->imported_as($record->id);
1125                         $imported = 1;
1126                     }
1127                 }
1128             }
1129         }
1130
1131         if($imported) {
1132
1133             $rec->import_time('now');
1134             $rec->clear_import_error;
1135             $rec->clear_error_detail;
1136
1137             if($e->$update_func($rec)) {
1138
1139                 if($type eq 'bib') {
1140
1141                     # see if this record is linked from an acq record.
1142                     my $li = $e->search_acq_lineitem(
1143                         {queued_record => $rec->id, state => {'!=' => 'canceled'}})->[0];
1144
1145                     if ($li) {
1146                         # if so, update the acq lineitem to point to the imported record
1147                         $li->eg_bib_id($rec->imported_as);
1148                         $$report_args{evt} = $e->die_event unless $e->update_acq_lineitem($li);
1149                     }
1150                 }
1151
1152                 push @success_rec_ids, $rec_id;
1153                 finish_rec_import_attempt($report_args);
1154
1155             } else {
1156                 $imported = 0;
1157             }
1158         }
1159
1160         if(!$imported) {
1161             $logger->info("vl: record $rec_id was not imported");
1162             $$report_args{evt} = $e->event unless $$report_args{evt};
1163             $$report_args{no_import} = 1;
1164             finish_rec_import_attempt($report_args);
1165         }
1166     }
1167
1168     # see if we need to mark any queues as complete
1169     for my $q_id (keys %queues) {
1170
1171         my $e = new_editor(xact => 1);
1172         my $remaining = $e->$search_func(
1173             [{queue => $q_id, import_time => undef}, {limit =>1}], {idlist => 1});
1174
1175         unless(@$remaining) {
1176             my $queue = $e->$retrieve_queue_func($q_id);
1177             unless($U->is_true($queue->complete)) {
1178                 $queue->complete('t');
1179                 $e->$update_queue_func($queue) or return $e->die_event;
1180                 $e->commit;
1181                 next;
1182             }
1183         }
1184         $e->rollback;
1185     }
1186
1187     # import the copies
1188     import_record_asset_list_impl($conn, \@success_rec_ids, $requestor) if @success_rec_ids;
1189
1190     $conn->respond({total => $$report_args{total}, progress => $$report_args{progress}});
1191     return undef;
1192 }
1193
1194
1195 sub try_auto_overlay {
1196     my $e = shift;
1197     my $type = shift;
1198     my $report_args = shift;
1199     my $overlay_func  = shift;
1200     my $retrieve_func = shift;
1201     my $rec_class = shift;
1202     my $rec_id  = shift;
1203     my $match_quality_ratio = shift;
1204     my $merge_profile  = shift;
1205     my $ft_merge_profile = shift;
1206
1207     my $imported = 0;
1208     my $error = 0;
1209
1210     # Find the best match and overlay if the quality ratio allows it.
1211     my $res = $e->json_query(
1212         {
1213             from => [
1214                 $overlay_func,
1215                 $rec_id,
1216                 $merge_profile,
1217                 $match_quality_ratio
1218             ]
1219         }
1220     );
1221
1222     if($res and ($res = $res->[0])) {
1223
1224         if($res->{$overlay_func} eq 't') {
1225
1226             # first attempt succeeded
1227             $imported = 1;
1228
1229         } else {
1230
1231             # quality-limited merge failed with insufficient quality.  If there is a
1232             # fall-through merge profile, re-do the merge with the alternate profile
1233             # and no quality restriction.
1234
1235             if($ft_merge_profile and $match_quality_ratio > 0) {
1236
1237                 $logger->info("vl: $type auto-merge failed with profile $merge_profile; ".
1238                     "re-merging with fall-through profile $ft_merge_profile");
1239
1240                 my $res = $e->json_query(
1241                     {
1242                         from => [
1243                             $overlay_func,
1244                             $rec_id,
1245                             $ft_merge_profile,
1246                             0 # minimum quality not required
1247                         ]
1248                     }
1249                 );
1250
1251                 if($res and ($res = $res->[0])) {
1252
1253                     if($res->{$overlay_func} eq 't') {
1254
1255                         # second attempt succeeded
1256                         $imported = 1;
1257
1258                     } else {
1259
1260                         # failed to merge on second attempt
1261                         $logger->info("vl: $type auto-merge with fall-through failed for queued rec $rec_id");
1262                     }
1263                 } else {
1264
1265                     # second attempt died
1266                     $error = 1;
1267                     $logger->error("vl: Error attempting overlay with func=$overlay_func, profile=$merge_profile, record=$rec_id");
1268                 }
1269
1270             } else {
1271
1272                 # failed to merge on first attempt, no fall-through was provided
1273                 $$report_args{import_error} = 'overlay.record.quality' if $match_quality_ratio > 0;
1274                 $logger->info("vl: $type auto-merge failed for queued rec $rec_id");
1275             }
1276         }
1277
1278     } else {
1279
1280         # first attempt died
1281         $error = 1;
1282         $logger->error("vl: Error attempting overlay with func=$overlay_func, profile=$merge_profile, record=$rec_id");
1283     }
1284
1285     if($imported) {
1286
1287         # at least 1 of the attempts succeeded
1288         $logger->info("vl: $type auto-merge succeeded for queued rec $rec_id");
1289
1290         # re-fetch the record to pick up the imported_as value from the DB
1291         $$report_args{rec} = $e->$retrieve_func([
1292             $rec_id, {flesh => 1, flesh_fields => {$rec_class => ['matches']}}]);
1293     }
1294
1295     return ($imported, $error, $$report_args{rec});
1296 }
1297
1298
1299 # tracks any import errors, commits the current xact, responds to the client
1300 sub finish_rec_import_attempt {
1301     my $args = shift;
1302     my $evt = $$args{evt};
1303     my $rec = $$args{rec};
1304     my $e = $$args{e};
1305
1306     my $error = $$args{import_error};
1307     $error = 'general.unknown' if $evt and not $error;
1308
1309     # error tracking
1310     if($rec) {
1311
1312         if($error or $evt) {
1313             # failed import
1314             # since an error occurred, there's no guarantee the transaction wasn't
1315             # rolled back.  force a rollback and create a new editor.
1316             $e->rollback;
1317             $e = new_editor(xact => 1);
1318             $rec->import_error($error);
1319
1320             if($evt) {
1321                 my $detail = sprintf("%s : %s", $evt->{textcode}, substr($evt->{desc}, 0, 140));
1322                 $rec->error_detail($detail);
1323             }
1324
1325             my $method = 'update_vandelay_queued_bib_record';
1326             $method =~ s/bib/authority/ if $$args{type} eq 'auth';
1327             $e->$method($rec) and $e->commit or $e->rollback;
1328
1329         } else {
1330             # commit the successful import
1331             $e->commit;
1332         }
1333
1334     } else {
1335         # requested queued record was not found
1336         $e->rollback;
1337     }
1338
1339     # respond to client
1340     if($$args{report_all} or ($$args{progress} % $$args{step}) == 0) {
1341         $$args{conn}->respond({
1342             total => $$args{total},
1343             progress => $$args{progress},
1344             imported => ($rec) ? $rec->id : undef,
1345             import_error => $error,
1346             no_import => $$args{no_import},
1347             err_event => $evt
1348         });
1349         $$args{step} *= 2 unless $$args{step} == 256;
1350     }
1351
1352     $$args{progress}++;
1353 }
1354
1355
1356
1357
1358
1359 __PACKAGE__->register_method(
1360     api_name    => "open-ils.vandelay.bib_queue.owner.retrieve",
1361     method      => 'owner_queue_retrieve',
1362     api_level   => 1,
1363     argc        => 2,
1364     stream      => 1,
1365     record_type => 'bib'
1366 );
1367 __PACKAGE__->register_method(
1368     api_name    => "open-ils.vandelay.authority_queue.owner.retrieve",
1369     method      => 'owner_queue_retrieve',
1370     api_level   => 1,
1371     argc        => 2,
1372     stream      => 1,
1373     record_type => 'auth'
1374 );
1375
1376 sub owner_queue_retrieve {
1377     my($self, $conn, $auth, $owner_id, $filters) = @_;
1378     my $e = new_editor(authtoken => $auth, xact => 1);
1379     return $e->die_event unless $e->checkauth;
1380     $owner_id = $e->requestor->id; # XXX add support for viewing other's queues?
1381     my $queues;
1382     $filters ||= {};
1383     my $search = {owner => $owner_id};
1384     $search->{$_} = $filters->{$_} for keys %$filters;
1385
1386     if($self->{record_type} eq 'bib') {
1387         $queues = $e->search_vandelay_bib_queue(
1388             [$search, {order_by => {vbq => 'evergreen.lowercase(name)'}}]);
1389     } else {
1390         $queues = $e->search_vandelay_authority_queue(
1391             [$search, {order_by => {vaq => 'evergreen.lowercase(name)'}}]);
1392     }
1393     $conn->respond($_) for @$queues;
1394     $e->rollback;
1395     return undef;
1396 }
1397
1398 __PACKAGE__->register_method(
1399     api_name    => "open-ils.vandelay.bib_queue.delete",
1400     method      => "delete_queue",
1401     api_level   => 1,
1402     argc        => 2,
1403     record_type => 'bib'
1404 );
1405 __PACKAGE__->register_method(
1406     api_name    => "open-ils.vandelay.auth_queue.delete",
1407     method      => "delete_queue",
1408     api_level   => 1,
1409     argc        => 2,
1410     record_type => 'auth'
1411 );
1412
1413 sub delete_queue {
1414     my($self, $conn, $auth, $q_id) = @_;
1415     my $e = new_editor(xact => 1, authtoken => $auth);
1416     return $e->die_event unless $e->checkauth;
1417     if($self->{record_type} eq 'bib') {
1418         return $e->die_event unless $e->allowed('CREATE_BIB_IMPORT_QUEUE');
1419         my $queue = $e->retrieve_vandelay_bib_queue($q_id)
1420             or return $e->die_event;
1421         $e->delete_vandelay_bib_queue($queue)
1422             or return $e->die_event;
1423     } else {
1424            return $e->die_event unless $e->allowed('CREATE_AUTHORITY_IMPORT_QUEUE');
1425         my $queue = $e->retrieve_vandelay_authority_queue($q_id)
1426             or return $e->die_event;
1427         $e->delete_vandelay_authority_queue($queue)
1428             or return $e->die_event;
1429     }
1430     $e->commit;
1431     return 1;
1432 }
1433
1434
1435 __PACKAGE__->register_method(
1436     api_name    => "open-ils.vandelay.queued_bib_record.html",
1437     method      => 'queued_record_html',
1438     api_level   => 1,
1439     argc        => 2,
1440     stream      => 1,
1441     record_type => 'bib'
1442 );
1443 __PACKAGE__->register_method(
1444     api_name    => "open-ils.vandelay.queued_authority_record.html",
1445     method      => 'queued_record_html',
1446     api_level   => 1,
1447     argc        => 2,
1448     stream      => 1,
1449     record_type => 'auth'
1450 );
1451
1452 sub queued_record_html {
1453     my($self, $conn, $auth, $rec_id) = @_;
1454     my $e = new_editor(xact=>1,authtoken => $auth);
1455     return $e->die_event unless $e->checkauth;
1456     my $rec;
1457     if($self->{record_type} eq 'bib') {
1458         $rec = $e->retrieve_vandelay_queued_bib_record($rec_id)
1459             or return $e->die_event;
1460     } else {
1461         $rec = $e->retrieve_vandelay_queued_authority_record($rec_id)
1462             or return $e->die_event;
1463     }
1464
1465     $e->rollback;
1466     return $U->simplereq(
1467         'open-ils.search',
1468         'open-ils.search.biblio.record.html', undef, 1, $rec->marc);
1469 }
1470
1471
1472 __PACKAGE__->register_method(
1473     api_name    => "open-ils.vandelay.bib_queue.summary.retrieve",
1474     method      => 'retrieve_queue_summary',
1475     api_level   => 1,
1476     argc        => 2,
1477     stream      => 1,
1478     record_type => 'bib'
1479 );
1480 __PACKAGE__->register_method(
1481     api_name    => "open-ils.vandelay.auth_queue.summary.retrieve",
1482     method      => 'retrieve_queue_summary',
1483     api_level   => 1,
1484     argc        => 2,
1485     stream      => 1,
1486     record_type => 'auth'
1487 );
1488
1489 sub retrieve_queue_summary {
1490     my($self, $conn, $auth, $queue_id) = @_;
1491     my $e = new_editor(xact=>1, authtoken => $auth);
1492     return $e->die_event unless $e->checkauth;
1493
1494     my $queue;
1495     my $type = $self->{record_type};
1496     if($type eq 'bib') {
1497         $queue = $e->retrieve_vandelay_bib_queue($queue_id)
1498             or return $e->die_event;
1499     } else {
1500         $queue = $e->retrieve_vandelay_authority_queue($queue_id)
1501             or return $e->die_event;
1502     }
1503
1504     my $evt = check_queue_perms($e, $type, $queue);
1505     return $evt if $evt;
1506
1507     my $search = 'search_vandelay_queued_bib_record';
1508     $search =~ s/bib/authority/ if $type ne 'bib';
1509
1510     my $summary = {
1511         queue => $queue,
1512         total => scalar(@{$e->$search({queue => $queue_id}, {idlist=>1})}),
1513         imported => scalar(@{$e->$search({queue => $queue_id, import_time => {'!=' => undef}}, {idlist=>1})}),
1514     };
1515
1516     my $class = ($type eq 'bib') ? 'vqbr' : 'vqar';
1517     $summary->{rec_import_errors} = $e->json_query({
1518         select => {$class => [{alias => 'count', column => 'id', transform => 'count', aggregate => 1}]},
1519         from => $class,
1520         where => {queue => $queue_id, import_error => {'!=' => undef}}
1521     })->[0]->{count};
1522
1523     if($type eq 'bib') {
1524
1525         # count of all items attached to records in the queue in question
1526         my $query = {
1527             select => {vii => [{alias => 'count', column => 'id', transform => 'count', aggregate => 1}]},
1528             from => 'vii',
1529             where => {
1530                 record => {
1531                     in => {
1532                         select => {vqbr => ['id']},
1533                         from => 'vqbr',
1534                         where => {queue => $queue_id}
1535                     }
1536                 }
1537             }
1538         };
1539         $summary->{total_items} = $e->json_query($query)->[0]->{count};
1540
1541         # count of items we attempted to import, but errored, attached to records in the queue in question
1542         $query->{where}->{import_error} = {'!=' => undef};
1543         $summary->{item_import_errors} = $e->json_query($query)->[0]->{count};
1544
1545         # count of items we successfully imported attached to records in the queue in question
1546         delete $query->{where}->{import_error};
1547         $query->{where}->{import_time} = {'!=' => undef};
1548         $summary->{total_items_imported} = $e->json_query($query)->[0]->{count};
1549     }
1550
1551     return $summary;
1552 }
1553
1554 # --------------------------------------------------------------------------------
1555 # Given a list of queued record IDs, imports all items attached to those records
1556 # --------------------------------------------------------------------------------
1557 sub import_record_asset_list_impl {
1558     my($conn, $rec_ids, $requestor) = @_;
1559
1560     my $roe = new_editor(xact=> 1, requestor => $requestor);
1561
1562     # for speed, filter out any records have not been
1563     # imported or have no import items to load
1564     $rec_ids = $roe->json_query({
1565         select => {vqbr => ['id']},
1566         from => {vqbr => 'vii'},
1567         where => {'+vqbr' => {
1568             id => $rec_ids,
1569             import_time => {'!=' => undef}
1570         }},
1571         distinct => 1
1572     });
1573     $rec_ids = [map {$_->{id}} @$rec_ids];
1574
1575     my $report_args = {
1576         conn => $conn,
1577         total => scalar(@$rec_ids),
1578         step => 1, # how often to respond
1579         progress => 1,
1580         in_count => 0,
1581     };
1582
1583     for my $rec_id (@$rec_ids) {
1584         my $rec = $roe->retrieve_vandelay_queued_bib_record($rec_id);
1585         my $item_ids = $roe->search_vandelay_import_item(
1586             {record => $rec->id, import_error => undef},
1587             {idlist=>1}
1588         );
1589
1590         for my $item_id (@$item_ids) {
1591             my $e = new_editor(requestor => $requestor, xact => 1);
1592             my $item = $e->retrieve_vandelay_import_item($item_id);
1593             my ($copy, $vol, $evt);
1594
1595             $$report_args{import_item} = $item;
1596             $$report_args{e} = $e;
1597             $$report_args{import_error} = undef;
1598             $$report_args{evt} = undef;
1599
1600             if (my $copy_id = $item->internal_id) { # assignment
1601                 # copy matches an existing copy.  Overlay instead of create.
1602
1603                 $logger->info("vl: performing copy overlay for internal_id=$copy_id");
1604
1605                 my $qt = $e->json_query({
1606                     select => {vbq => ['queue_type']},
1607                     from => {vqbr => 'vbq'},
1608                     where => {'+vqbr' => {id => $rec_id}}
1609                 })->[0]->{queue_type};
1610
1611                 if ($qt eq 'acq') {
1612                     # internal_id for ACQ queues refers to acq.lineitem_detail.id
1613                     # pull the real copy id from the acq LID
1614
1615                     my $lid = $e->retrieve_acq_lineitem_detail($copy_id);
1616                     if (!$lid) {
1617                         $$report_args{evt} = $e->die_event;
1618                         respond_with_status($report_args);
1619                         next;
1620                     }
1621                     $copy_id = $lid->eg_copy_id;
1622                     $logger->info("vl: performing ACQ copy overlay for copy $copy_id");
1623                 }
1624
1625                 $copy = $e->search_asset_copy([
1626                     {id => $copy_id, deleted => 'f'},
1627                     {flesh => 1, flesh_fields => {acp => ['call_number']}}
1628                 ])->[0];
1629
1630                 if (!$copy) {
1631                     $$report_args{evt} = $e->die_event;
1632                     respond_with_status($report_args);
1633                     next;
1634                 }
1635
1636                 # prevent update of unrelated copies
1637                 if ($copy->call_number->record != $rec->imported_as) {
1638                     $logger->info("vl: attempt to overlay unrelated copy=$copy_id; rec=".$rec->imported_as);
1639
1640                     $evt = OpenILS::Event->new('INVALID_IMPORT_COPY_ID',
1641                         note => 'Cannot overlay copies for unlinked bib',
1642                         bre => $rec->imported_as,
1643                         copy_id => $copy_id
1644                     );
1645                     $$report_args{evt} = $evt;
1646                     respond_with_status($report_args);
1647                     next;
1648                 }
1649
1650                 # overlaying copies requires an extra permission
1651                 if (!$e->allowed("IMPORT_OVERLAY_COPY", $copy->call_number->owning_lib)) {
1652                     $$report_args{evt} = $e->die_event;
1653                     respond_with_status($report_args);
1654                     next;
1655                 }
1656
1657                 # are we updating the call-number?
1658                 if ($item->call_number and $item->call_number ne $copy->call_number->label) {
1659
1660                     my $count = $e->json_query({
1661                         select => {acp => [{
1662                             alias => 'count',
1663                             column => 'id',
1664                             transform => 'count',
1665                             aggregate => 1
1666                         }]},
1667                         from => 'acp',
1668                         where => {
1669                             deleted => 'f',
1670                             call_number => $copy->call_number->id
1671                         }
1672                     })->[0]->{count};
1673
1674                     if ($count == 1) {
1675
1676                         my $evol = $e->search_asset_call_number({
1677                             id => {'<>' => $copy->call_number->id},
1678                             label => $item->call_number,
1679                             owning_lib => $copy->call_number->owning_lib,
1680                             record => $copy->call_number->record,
1681                             prefix => $copy->call_number->prefix,
1682                             suffix => $copy->call_number->suffix,
1683                             deleted => 'f'
1684                         })->[0];
1685
1686                         if ($evol) {
1687                             # call number for overlayed copy changed to a
1688                             # label already in use by another call number.
1689                             # merge the old CN into the new CN
1690
1691                             $logger->info(
1692                                 "vl: moving copy to new call number ".
1693                                 $item->call_number);
1694
1695                             my ($mvol, $err) =
1696                                 OpenILS::Application::Cat::Merge::merge_volumes(
1697                                     $e, [$copy->call_number], $evol);
1698
1699                             if (!$mvol) {
1700                                 $$report_args{evt} = $err;
1701                                 respond_with_status($report_args);
1702                                 next;
1703                             }
1704
1705                             # update our copy *cough* of the copy to pick up
1706                             # any changes made my merge_volumes
1707                             $copy = $e->retrieve_asset_copy([
1708                                 $copy->id,
1709                                 {flesh => 1, flesh_fields => {acp => ['call_number']}}
1710                             ]);
1711
1712                         } else {
1713                             $logger->info(
1714                                 "vl: updating copy call number label".
1715                                 $item->call_number);
1716
1717                             $copy->call_number->label($item->call_number);
1718                             if (!$e->update_asset_call_number($copy->call_number)) {
1719                                 $$report_args{evt} = $e->die_event;
1720                                 respond_with_status($report_args);
1721                                 next;
1722                             }
1723                         }
1724
1725                     } else {
1726
1727                         # otherwise, move the copy to a new/existing
1728                         # call-number with the given label/owner
1729                         # note that overlay does not allow the owning_lib
1730                         # to be changed.  Should it?
1731
1732                         $logger->info("vl: moving copy to new callnumber in copy overlay");
1733
1734                         ($vol, $evt) =
1735                             OpenILS::Application::Cat::AssetCommon->find_or_create_volume(
1736                                 $e, $item->call_number,
1737                                 $copy->call_number->record,
1738                                 $copy->call_number->owning_lib
1739                             );
1740
1741                         if($evt) {
1742                             $$report_args{evt} = $evt;
1743                             respond_with_status($report_args);
1744                             next;
1745                         }
1746
1747                         $copy->call_number($vol);
1748                     }
1749                 } # cn-update
1750
1751                 # for every field that has a non-'' value, overlay the copy value
1752                 foreach (qw/ barcode location circ_lib status
1753                     circulate deposit deposit_amount ref holdable
1754                     price circ_as_type alert_message opac_visible circ_modifier/) {
1755
1756                     my $val = $item->$_();
1757                     $copy->$_($val) if $val and $val ne '';
1758                 }
1759
1760                 # de-flesh for update
1761                 $copy->call_number($copy->call_number->id);
1762                 $copy->ischanged(1);
1763
1764                 $evt = OpenILS::Application::Cat::AssetCommon->
1765                     update_fleshed_copies($e, {all => 1}, undef, [$copy]);
1766
1767                 if($evt) {
1768                     $$report_args{evt} = $evt;
1769                     respond_with_status($report_args);
1770                     next;
1771                 }
1772
1773             } else {
1774
1775                 # Creating a new copy
1776                 $logger->info("vl: creating new copy in import");
1777
1778                 # --------------------------------------------------------------------------------
1779                 # Find or create the volume
1780                 # --------------------------------------------------------------------------------
1781                 my ($vol, $evt) =
1782                     OpenILS::Application::Cat::AssetCommon->find_or_create_volume(
1783                         $e, $item->call_number, $rec->imported_as, $item->owning_lib);
1784
1785                 if($evt) {
1786                     $$report_args{evt} = $evt;
1787                     respond_with_status($report_args);
1788                     next;
1789                 }
1790
1791                 # --------------------------------------------------------------------------------
1792                 # Create the new copy
1793                 # --------------------------------------------------------------------------------
1794                 $copy = Fieldmapper::asset::copy->new;
1795                 $copy->loan_duration(2);
1796                 $copy->fine_level(2);
1797                 $copy->barcode($item->barcode);
1798                 $copy->location($item->location);
1799                 $copy->circ_lib($item->circ_lib || $item->owning_lib);
1800                 $copy->status( defined($item->status) ? $item->status : OILS_COPY_STATUS_IN_PROCESS );
1801                 $copy->circulate($item->circulate);
1802                 $copy->deposit($item->deposit);
1803                 $copy->deposit_amount($item->deposit_amount);
1804                 $copy->ref($item->ref);
1805                 $copy->holdable($item->holdable);
1806                 $copy->price($item->price);
1807                 $copy->circ_as_type($item->circ_as_type);
1808                 $copy->alert_message($item->alert_message);
1809                 $copy->opac_visible($item->opac_visible);
1810                 $copy->circ_modifier($item->circ_modifier);
1811
1812                 # --------------------------------------------------------------------------------
1813                 # Check for dupe barcode
1814                 # --------------------------------------------------------------------------------
1815                 if($evt = OpenILS::Application::Cat::AssetCommon->create_copy($e, $vol, $copy)) {
1816                     $$report_args{evt} = $evt;
1817                     $$report_args{import_error} = 'import.item.duplicate.barcode'
1818                         if $evt->{textcode} eq 'ITEM_BARCODE_EXISTS';
1819                     respond_with_status($report_args);
1820                     next;
1821                 }
1822
1823                 # --------------------------------------------------------------------------------
1824                 # create copy notes
1825                 # --------------------------------------------------------------------------------
1826                 $evt = OpenILS::Application::Cat::AssetCommon->create_copy_note(
1827                     $e, $copy, '', $item->pub_note, 1) if $item->pub_note;
1828
1829                 if($evt) {
1830                     $$report_args{evt} = $evt;
1831                     respond_with_status($report_args);
1832                     next;
1833                 }
1834
1835                 $evt = OpenILS::Application::Cat::AssetCommon->create_copy_note(
1836                     $e, $copy, '', $item->priv_note) if $item->priv_note;
1837
1838                 if($evt) {
1839                     $$report_args{evt} = $evt;
1840                     respond_with_status($report_args);
1841                     next;
1842                 }
1843             }
1844
1845             # set the import data on the import item
1846             $item->imported_as($copy->id); # $copy->id is set by create_copy() ^--
1847             $item->import_time('now');
1848
1849             unless($e->update_vandelay_import_item($item)) {
1850                 $$report_args{evt} = $e->die_event;
1851                 respond_with_status($report_args);
1852                 next;
1853             }
1854
1855             # --------------------------------------------------------------------------------
1856             # Item import succeeded
1857             # --------------------------------------------------------------------------------
1858             $e->commit;
1859             $$report_args{in_count}++;
1860             respond_with_status($report_args);
1861             $logger->info("vl: successfully imported item " . $item->barcode);
1862         }
1863     }
1864
1865     $roe->rollback;
1866     return undef;
1867 }
1868
1869
1870 sub respond_with_status {
1871     my $args = shift;
1872     my $e = $$args{e};
1873
1874     #  If the import failed, track the failure reason
1875
1876     my $error = $$args{import_error};
1877     my $evt = $$args{evt};
1878
1879     if($error or $evt) {
1880
1881         my $item = $$args{import_item};
1882         $logger->info("vl: unable to import item " . $item->barcode);
1883
1884         $error ||= 'general.unknown';
1885         $item->import_error($error);
1886
1887         if($evt) {
1888             my $detail = sprintf("%s : %s", $evt->{textcode}, substr($evt->{desc}, 0, 140));
1889             $item->error_detail($detail);
1890         }
1891
1892         # state of the editor is unknown at this point.  Force a rollback and start over.
1893         $e->rollback;
1894         $e = new_editor(xact => 1);
1895         $e->update_vandelay_import_item($item);
1896         $e->commit;
1897     }
1898
1899     if($$args{report_all} or ($$args{progress} % $$args{step}) == 0) {
1900         $$args{conn}->respond({
1901             total => $$args{total},
1902             progress => $$args{progress},
1903             success_count => $$args{success_count},
1904             err_event => $evt
1905         });
1906         $$args{step} *= 2 unless $$args{step} == 256;
1907     }
1908
1909     $$args{progress}++;
1910 }
1911
1912 __PACKAGE__->register_method(
1913     api_name    => "open-ils.vandelay.match_set.get_tree",
1914     method      => "match_set_get_tree",
1915     api_level   => 1,
1916     argc        => 2,
1917     signature   => {
1918         desc    => q/For a given vms object, return a tree of match set points
1919                     represented by a vmsp object with recursively fleshed
1920                     children./
1921     }
1922 );
1923
1924 sub match_set_get_tree {
1925     my ($self, $conn, $authtoken, $match_set_id) = @_;
1926
1927     $match_set_id = int($match_set_id) or return;
1928
1929     my $e = new_editor("authtoken" => $authtoken);
1930     $e->checkauth or return $e->die_event;
1931
1932     my $set = $e->retrieve_vandelay_match_set($match_set_id) or
1933         return $e->die_event;
1934
1935     $e->allowed("ADMIN_IMPORT_MATCH_SET", $set->owner) or
1936         return $e->die_event;
1937
1938     my $tree = $e->search_vandelay_match_set_point([
1939         {"match_set" => $match_set_id, "parent" => undef},
1940         {"flesh" => -1, "flesh_fields" => {"vmsp" => ["children"]}}
1941     ]) or return $e->die_event;
1942
1943     return pop @$tree;
1944 }
1945
1946
1947 __PACKAGE__->register_method(
1948     api_name    => "open-ils.vandelay.match_set.update",
1949     method      => "match_set_update_tree",
1950     api_level   => 1,
1951     argc        => 3,
1952     signature   => {
1953         desc => q/Replace any vmsp objects associated with a given (by ID) vms
1954                 with the given objects (recursively fleshed vmsp tree)./
1955     }
1956 );
1957
1958 sub _walk_new_vmsp {
1959     my ($e, $match_set_id, $node, $parent_id) = @_;
1960
1961     my $point = new Fieldmapper::vandelay::match_set_point;
1962     $point->parent($parent_id);
1963     $point->match_set($match_set_id);
1964     $point->$_($node->$_) for (qw/bool_op svf tag subfield negate quality/);
1965
1966     $e->create_vandelay_match_set_point($point) or return $e->die_event;
1967
1968     $parent_id = $e->data->id;
1969     if ($node->children && @{$node->children}) {
1970         for (@{$node->children}) {
1971             return $e->die_event if
1972                 _walk_new_vmsp($e, $match_set_id, $_, $parent_id);
1973         }
1974     }
1975
1976     return;
1977 }
1978
1979 sub match_set_update_tree {
1980     my ($self, $conn, $authtoken, $match_set_id, $tree) = @_;
1981
1982     my $e = new_editor("xact" => 1, "authtoken" => $authtoken);
1983     $e->checkauth or return $e->die_event;
1984
1985     my $set = $e->retrieve_vandelay_match_set($match_set_id) or
1986         return $e->die_event;
1987
1988     $e->allowed("ADMIN_IMPORT_MATCH_SET", $set->owner) or
1989         return $e->die_event;
1990
1991     my $existing = $e->search_vandelay_match_set_point([
1992         {"match_set" => $match_set_id},
1993         {"order_by" => {"vmsp" => "id DESC"}}
1994     ]) or return $e->die_event;
1995
1996     # delete points, working up from leaf points to the root
1997     while(@$existing) {
1998         for my $point (shift @$existing) {
1999             if( grep {$_->parent eq $point->id} @$existing) {
2000                 push(@$existing, $point);
2001             } else {
2002                 $e->delete_vandelay_match_set_point($point) or return $e->die_event;
2003             }
2004         }
2005     }
2006
2007     _walk_new_vmsp($e, $match_set_id, $tree);
2008
2009     $e->commit or return $e->die_event;
2010 }
2011
2012 __PACKAGE__->register_method(
2013     api_name    => 'open-ils.vandelay.bib_queue.to_bucket',
2014     method      => 'bib_queue_to_bucket',
2015     api_level   => 1,
2016     argc        => 2,
2017     signature   => {
2018         desc    => q/Add to or create a new bib container (bucket) with the successfully
2019                     imported contents of a vandelay queue.  Any user that has Vandelay
2020                     queue create permissions can append or create buckets from his-her own queues./,
2021         params  => [
2022             {desc => 'Authtoken', type => 'string'},
2023             {desc => 'Queue ID', type => 'number'},
2024             {desc => 'Bucket Name', type => 'string'}
2025         ],
2026         return  => {desc => q/
2027             {bucket => $bucket, addcount => number-of-items-added-to-bucket, item_count => total-bucket-item-count} on success,
2028             {add_count => 0} if there is nothing to do, and Event on error/}
2029     }
2030 );
2031
2032 sub bib_queue_to_bucket {
2033     my ($self, $conn, $auth, $q_id, $bucket_name) = @_;
2034
2035     my $e = new_editor(xact => 1, authtoken => $auth);
2036     return $e->die_event unless $e->checkauth;
2037
2038     my $queue = $e->retrieve_vandelay_bib_queue($q_id)
2039         or return $e->die_event;
2040
2041     return OpenILS::Event->new('BAD_PARAMS',
2042         note => q/Bucket creator must be queue owner/)
2043         unless $queue->owner == $e->requestor->id;
2044
2045     # find the bib IDs that will go into the bucket
2046     my $bib_ids = $e->json_query({
2047         select => {vqbr => ['imported_as']},
2048         from => 'vqbr',
2049         where => {queue => $q_id, imported_as => {'!=' => undef}}
2050     });
2051
2052     if (!@$bib_ids) { # no records to add
2053         $e->rollback;
2054         return {add_count => 0};
2055     }
2056
2057     # allow user to add to an existing bucket by name
2058     my $bucket = $e->search_container_biblio_record_entry_bucket({
2059         owner => $e->requestor->id,
2060         name => $bucket_name,
2061         btype => 'vandelay_queue'
2062     })->[0];
2063
2064     # if the bucket does not exist, create a new one
2065     if (!$bucket) {
2066         $bucket = Fieldmapper::container::biblio_record_entry_bucket->new;
2067         $bucket->name($bucket_name);
2068         $bucket->owner($e->requestor->id);
2069         $bucket->btype('vandelay_queue');
2070
2071         $e->create_container_biblio_record_entry_bucket($bucket)
2072             or return $e->die_event;
2073     }
2074
2075     # create the new bucket items
2076     for my $bib_id ( map {$_->{imported_as}} @$bib_ids ) {
2077         my $item = Fieldmapper::container::biblio_record_entry_bucket_item->new;
2078         $item->target_biblio_record_entry($bib_id);
2079         $item->bucket($bucket->id);
2080         $e->create_container_biblio_record_entry_bucket_item($item)
2081             or return $e->die_event;
2082     }
2083
2084     # re-fetch the bucket to pick up the correct create_time
2085     $bucket = $e->retrieve_container_biblio_record_entry_bucket($bucket->id)
2086         or return $e->die_event;
2087
2088     # get the total count of items in this bucket
2089     my $count = $e->json_query({
2090         select => {cbrebi => [{
2091             aggregate =>  1,
2092             transform => 'count',
2093             alias => 'count',
2094             column => 'id'
2095         }]},
2096         from => 'cbrebi',
2097         where => {bucket => $bucket->id}
2098     })->[0];
2099
2100     $e->commit;
2101
2102     return {
2103         bucket => $bucket,
2104         add_count => scalar(@$bib_ids), # items added to the bucket
2105         item_count => $count->{count} # total items in buckets
2106     };
2107 }
2108
2109 1;