C4/ImportBatch.pm

   1 package C4::ImportBatch;
   2
   3 # Copyright (C) 2007 LibLime, 2012 C & P Bibliography Services
   4 #
   5 # This file is part of Koha.
   6 #
   7 # Koha is free software; you can redistribute it and/or modify it
   8 # under the terms of the GNU General Public License as published by
   9 # the Free Software Foundation; either version 3 of the License, or
  10 # (at your option) any later version.
  11 #
  12 # Koha is distributed in the hope that it will be useful, but
  13 # WITHOUT ANY WARRANTY; without even the implied warranty of
  14 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  15 # GNU General Public License for more details.
  16 #
  17 # You should have received a copy of the GNU General Public License
  18 # along with Koha; if not, see <http://www.gnu.org/licenses>.
  19
  20 use strict;
  21 use warnings;
  22
  23 use C4::Context;
  24 use C4::Koha;
  25 use C4::Biblio;
  26 use C4::Items;
  27 use C4::Charset;
  28 use C4::AuthoritiesMarc;
  29 use C4::MarcModificationTemplates;
  30 use Koha::Plugins::Handler;
  31 use Koha::Logger;
  32
  33 use vars qw(@ISA @EXPORT @EXPORT_OK %EXPORT_TAGS);
  34
  35 BEGIN {
  36         # set the version for version checking
  37         require Exporter;
  38         @ISA    = qw(Exporter);
  39         @EXPORT = qw(
  40     GetZ3950BatchId
  41     GetWebserviceBatchId
  42     GetImportRecordMarc
  43     GetImportRecordMarcXML
  44     AddImportBatch
  45     GetImportBatch
  46     AddAuthToBatch
  47     AddBiblioToBatch
  48     AddItemsToImportBiblio
  49     ModAuthorityInBatch
  50     ModBiblioInBatch
  51
  52     BatchStageMarcRecords
  53     BatchFindDuplicates
  54     BatchCommitRecords
  55     BatchRevertRecords
  56     CleanBatch
  57
  58     GetAllImportBatches
  59     GetStagedWebserviceBatches
  60     GetImportBatchRangeDesc
  61     GetNumberOfNonZ3950ImportBatches
  62     GetImportBiblios
  63     GetImportRecordsRange
  64         GetItemNumbersFromImportBatch
  65
  66     GetImportBatchStatus
  67     SetImportBatchStatus
  68     GetImportBatchOverlayAction
  69     SetImportBatchOverlayAction
  70     GetImportBatchNoMatchAction
  71     SetImportBatchNoMatchAction
  72     GetImportBatchItemAction
  73     SetImportBatchItemAction
  74     GetImportBatchMatcher
  75     SetImportBatchMatcher
  76     GetImportRecordOverlayStatus
  77     SetImportRecordOverlayStatus
  78     GetImportRecordStatus
  79     SetImportRecordStatus
  80     GetImportRecordMatches
  81     SetImportRecordMatches
  82         );
  83 }
  84
  85 our $logger = Koha::Logger->get( { category => 'C4.ImportBatch' } );
  86
  87 =head1 NAME
  88
  89 C4::ImportBatch - manage batches of imported MARC records
  90
  91 =head1 SYNOPSIS
  92
  93 use C4::ImportBatch;
  94
  95 =head1 FUNCTIONS
  96
  97 =head2 GetZ3950BatchId
  98
  99   my $batchid = GetZ3950BatchId($z3950server);
 100
 101 Retrieves the ID of the import batch for the Z39.50
 102 reservoir for the given target.  If necessary,
 103 creates the import batch.
 104
 105 =cut
 106
 107 sub GetZ3950BatchId {
 108     my ($z3950server) = @_;
 109
 110     my $dbh = C4::Context->dbh;
 111     my $sth = $dbh->prepare("SELECT import_batch_id FROM import_batches
 112                              WHERE  batch_type = 'z3950'
 113                              AND    file_name = ?");
 114     $sth->execute($z3950server);
 115     my $rowref = $sth->fetchrow_arrayref();
 116     $sth->finish();
 117     if (defined $rowref) {
 118         return $rowref->[0];
 119     } else {
 120         my $batch_id = AddImportBatch( {
 121                 overlay_action => 'create_new',
 122                 import_status => 'staged',
 123                 batch_type => 'z3950',
 124                 file_name => $z3950server,
 125             } );
 126         return $batch_id;
 127     }
 128
 129 }
 130
 131 =head2 GetWebserviceBatchId
 132
 133   my $batchid = GetWebserviceBatchId();
 134
 135 Retrieves the ID of the import batch for webservice.
 136 If necessary, creates the import batch.
 137
 138 =cut
 139
 140 my $WEBSERVICE_BASE_QRY = <<EOQ;
 141 SELECT import_batch_id FROM import_batches
 142 WHERE  batch_type = 'webservice'
 143 AND    import_status = 'staged'
 144 EOQ
 145 sub GetWebserviceBatchId {
 146     my ($params) = @_;
 147
 148     my $dbh = C4::Context->dbh;
 149     my $sql = $WEBSERVICE_BASE_QRY;
 150     my @args;
 151     foreach my $field (qw(matcher_id overlay_action nomatch_action item_action)) {
 152         if (my $val = $params->{$field}) {
 153             $sql .= " AND $field = ?";
 154             push @args, $val;
 155         }
 156     }
 157     my $id = $dbh->selectrow_array($sql, undef, @args);
 158     return $id if $id;
 159
 160     $params->{batch_type} = 'webservice';
 161     $params->{import_status} = 'staged';
 162     return AddImportBatch($params);
 163 }
 164
 165 =head2 GetImportRecordMarc
 166
 167   my ($marcblob, $encoding) = GetImportRecordMarc($import_record_id);
 168
 169 =cut
 170
 171 sub GetImportRecordMarc {
 172     my ($import_record_id) = @_;
 173
 174     my $dbh = C4::Context->dbh;
 175     my ( $marc, $encoding ) = $dbh->selectrow_array(q|
 176         SELECT marc, encoding
 177         FROM import_records
 178         WHERE import_record_id = ?
 179     |, undef, $import_record_id );
 180
 181     return $marc, $encoding;
 182 }
 183
 184 sub GetRecordFromImportBiblio {
 185     my ( $import_record_id, $embed_items ) = @_;
 186
 187     my ($marc) = GetImportRecordMarc($import_record_id);
 188     my $record = MARC::Record->new_from_usmarc($marc);
 189
 190     EmbedItemsInImportBiblio( $record, $import_record_id ) if $embed_items;
 191
 192     return $record;
 193 }
 194
 195 sub EmbedItemsInImportBiblio {
 196     my ( $record, $import_record_id ) = @_;
 197     my ( $itemtag, $itemsubfield ) = GetMarcFromKohaField("items.itemnumber", '');
 198     my $dbh = C4::Context->dbh;
 199     my $import_items = $dbh->selectall_arrayref(q|
 200         SELECT import_items.marcxml
 201         FROM import_items
 202         WHERE import_record_id = ?
 203     |, { Slice => {} }, $import_record_id );
 204     my @item_fields;
 205     for my $import_item ( @$import_items ) {
 206         my $item_marc = MARC::Record::new_from_xml($import_item->{marcxml});
 207         push @item_fields, $item_marc->field($itemtag);
 208     }
 209     $record->append_fields(@item_fields);
 210     return $record;
 211 }
 212
 213 =head2 GetImportRecordMarcXML
 214
 215   my $marcxml = GetImportRecordMarcXML($import_record_id);
 216
 217 =cut
 218
 219 sub GetImportRecordMarcXML {
 220     my ($import_record_id) = @_;
 221
 222     my $dbh = C4::Context->dbh;
 223     my $sth = $dbh->prepare("SELECT marcxml FROM import_records WHERE import_record_id = ?");
 224     $sth->execute($import_record_id);
 225     my ($marcxml) = $sth->fetchrow();
 226     $sth->finish();
 227     return $marcxml;
 228
 229 }
 230
 231 =head2 AddImportBatch
 232
 233   my $batch_id = AddImportBatch($params_hash);
 234
 235 =cut
 236
 237 sub AddImportBatch {
 238     my ($params) = @_;
 239
 240     my (@fields, @vals);
 241     foreach (qw( matcher_id template_id branchcode
 242                  overlay_action nomatch_action item_action
 243                  import_status batch_type file_name comments record_type )) {
 244         if (exists $params->{$_}) {
 245             push @fields, $_;
 246             push @vals, $params->{$_};
 247         }
 248     }
 249     my $dbh = C4::Context->dbh;
 250     $dbh->do("INSERT INTO import_batches (".join( ',', @fields).")
 251                                   VALUES (".join( ',', map '?', @fields).")",
 252              undef,
 253              @vals);
 254     return $dbh->{'mysql_insertid'};
 255 }
 256
 257 =head2 GetImportBatch
 258
 259   my $row = GetImportBatch($batch_id);
 260
 261 Retrieve a hashref of an import_batches row.
 262
 263 =cut
 264
 265 sub GetImportBatch {
 266     my ($batch_id) = @_;
 267
 268     my $dbh = C4::Context->dbh;
 269     my $sth = $dbh->prepare_cached("SELECT * FROM import_batches WHERE import_batch_id = ?");
 270     $sth->bind_param(1, $batch_id);
 271     $sth->execute();
 272     my $result = $sth->fetchrow_hashref;
 273     $sth->finish();
 274     return $result;
 275
 276 }
 277
 278 =head2 AddBiblioToBatch
 279
 280   my $import_record_id = AddBiblioToBatch($batch_id, $record_sequence,
 281                 $marc_record, $encoding, $z3950random, $update_counts);
 282
 283 =cut
 284
 285 sub AddBiblioToBatch {
 286     my $batch_id = shift;
 287     my $record_sequence = shift;
 288     my $marc_record = shift;
 289     my $encoding = shift;
 290     my $z3950random = shift;
 291     my $update_counts = @_ ? shift : 1;
 292
 293     my $import_record_id = _create_import_record($batch_id, $record_sequence, $marc_record, 'biblio', $encoding, $z3950random, C4::Context->preference('marcflavour'));
 294     _add_biblio_fields($import_record_id, $marc_record);
 295     _update_batch_record_counts($batch_id) if $update_counts;
 296     return $import_record_id;
 297 }
 298
 299 =head2 ModBiblioInBatch
 300
 301   ModBiblioInBatch($import_record_id, $marc_record);
 302
 303 =cut
 304
 305 sub ModBiblioInBatch {
 306     my ($import_record_id, $marc_record) = @_;
 307
 308     _update_import_record_marc($import_record_id, $marc_record, C4::Context->preference('marcflavour'));
 309     _update_biblio_fields($import_record_id, $marc_record);
 310
 311 }
 312
 313 =head2 AddAuthToBatch
 314
 315   my $import_record_id = AddAuthToBatch($batch_id, $record_sequence,
 316                 $marc_record, $encoding, $z3950random, $update_counts, [$marc_type]);
 317
 318 =cut
 319
 320 sub AddAuthToBatch {
 321     my $batch_id = shift;
 322     my $record_sequence = shift;
 323     my $marc_record = shift;
 324     my $encoding = shift;
 325     my $z3950random = shift;
 326     my $update_counts = @_ ? shift : 1;
 327     my $marc_type = shift || C4::Context->preference('marcflavour');
 328
 329     $marc_type = 'UNIMARCAUTH' if $marc_type eq 'UNIMARC';
 330
 331     my $import_record_id = _create_import_record($batch_id, $record_sequence, $marc_record, 'auth', $encoding, $z3950random, $marc_type);
 332     _add_auth_fields($import_record_id, $marc_record);
 333     _update_batch_record_counts($batch_id) if $update_counts;
 334     return $import_record_id;
 335 }
 336
 337 =head2 ModAuthInBatch
 338
 339   ModAuthInBatch($import_record_id, $marc_record);
 340
 341 =cut
 342
 343 sub ModAuthInBatch {
 344     my ($import_record_id, $marc_record) = @_;
 345
 346     my $marcflavour = C4::Context->preference('marcflavour');
 347     _update_import_record_marc($import_record_id, $marc_record, $marcflavour eq 'UNIMARC' ? 'UNIMARCAUTH' : 'USMARC');
 348
 349 }
 350
 351 =head2 BatchStageMarcRecords
 352
 353 ( $batch_id, $num_records, $num_items, @invalid_records ) =
 354   BatchStageMarcRecords(
 355     $encoding,                   $marc_records,
 356     $file_name,                  $to_marc_plugin,
 357     $marc_modification_template, $comments,
 358     $branch_code,                $parse_items,
 359     $leave_as_staging,           $progress_interval,
 360     $progress_callback
 361   );
 362
 363 =cut
 364
 365 sub BatchStageMarcRecords {
 366     my $record_type = shift;
 367     my $encoding = shift;
 368     my $marc_records = shift;
 369     my $file_name = shift;
 370     my $to_marc_plugin = shift;
 371     my $marc_modification_template = shift;
 372     my $comments = shift;
 373     my $branch_code = shift;
 374     my $parse_items = shift;
 375     my $leave_as_staging = shift;
 376
 377     # optional callback to monitor status
 378     # of job
 379     my $progress_interval = 0;
 380     my $progress_callback = undef;
 381     if ($#_ == 1) {
 382         $progress_interval = shift;
 383         $progress_callback = shift;
 384         $progress_interval = 0 unless $progress_interval =~ /^\d+$/ and $progress_interval > 0;
 385         $progress_interval = 0 unless 'CODE' eq ref $progress_callback;
 386     }
 387
 388     my $batch_id = AddImportBatch( {
 389             overlay_action => 'create_new',
 390             import_status => 'staging',
 391             batch_type => 'batch',
 392             file_name => $file_name,
 393             comments => $comments,
 394             record_type => $record_type,
 395         } );
 396     if ($parse_items) {
 397         SetImportBatchItemAction($batch_id, 'always_add');
 398     } else {
 399         SetImportBatchItemAction($batch_id, 'ignore');
 400     }
 401
 402     $marc_records = Koha::Plugins::Handler->run(
 403         {
 404             class  => $to_marc_plugin,
 405             method => 'to_marc',
 406             params => { data => $marc_records }
 407         }
 408     ) if $to_marc_plugin;
 409
 410     my $marc_type = C4::Context->preference('marcflavour');
 411     $marc_type .= 'AUTH' if ($marc_type eq 'UNIMARC' && $record_type eq 'auth');
 412     my @invalid_records = ();
 413     my $num_valid = 0;
 414     my $num_items = 0;
 415     # FIXME - for now, we're dealing only with bibs
 416     my $rec_num = 0;
 417     foreach my $marc_blob (split(/\x1D/, $marc_records)) {
 418         $marc_blob =~ s/^\s+//g;
 419         $marc_blob =~ s/\s+$//g;
 420         next unless $marc_blob;
 421         $rec_num++;
 422         if ($progress_interval and (0 == ($rec_num % $progress_interval))) {
 423             &$progress_callback($rec_num);
 424         }
 425         my ($marc_record, $charset_guessed, $char_errors) =
 426             MarcToUTF8Record($marc_blob, $marc_type, $encoding);
 427
 428         $encoding = $charset_guessed unless $encoding;
 429
 430         ModifyRecordWithTemplate( $marc_modification_template, $marc_record ) if ( $marc_modification_template );
 431
 432         my $import_record_id;
 433         if (scalar($marc_record->fields()) == 0) {
 434             push @invalid_records, $marc_blob;
 435         } else {
 436
 437             # Normalize the record so it doesn't have separated diacritics
 438             SetUTF8Flag($marc_record);
 439
 440             $num_valid++;
 441             if ($record_type eq 'biblio') {
 442                 $import_record_id = AddBiblioToBatch($batch_id, $rec_num, $marc_record, $encoding, int(rand(99999)), 0);
 443                 if ($parse_items) {
 444                     my @import_items_ids = AddItemsToImportBiblio($batch_id, $import_record_id, $marc_record, 0);
 445                     $num_items += scalar(@import_items_ids);
 446                 }
 447             } elsif ($record_type eq 'auth') {
 448                 $import_record_id = AddAuthToBatch($batch_id, $rec_num, $marc_record, $encoding, int(rand(99999)), 0, $marc_type);
 449             }
 450         }
 451     }
 452     unless ($leave_as_staging) {
 453         SetImportBatchStatus($batch_id, 'staged');
 454     }
 455     # FIXME branch_code, number of bibs, number of items
 456     _update_batch_record_counts($batch_id);
 457     return ($batch_id, $num_valid, $num_items, @invalid_records);
 458 }
 459
 460 =head2 AddItemsToImportBiblio
 461
 462   my @import_items_ids = AddItemsToImportBiblio($batch_id,
 463                 $import_record_id, $marc_record, $update_counts);
 464
 465 =cut
 466
 467 sub AddItemsToImportBiblio {
 468     my $batch_id = shift;
 469     my $import_record_id = shift;
 470     my $marc_record = shift;
 471     my $update_counts = @_ ? shift : 0;
 472
 473     my @import_items_ids = ();
 474
 475     my $dbh = C4::Context->dbh;
 476     my ($item_tag,$item_subfield) = &GetMarcFromKohaField("items.itemnumber",'');
 477     foreach my $item_field ($marc_record->field($item_tag)) {
 478         my $item_marc = MARC::Record->new();
 479         $item_marc->leader("00000    a              "); # must set Leader/09 to 'a'
 480         $item_marc->append_fields($item_field);
 481         $marc_record->delete_field($item_field);
 482         my $sth = $dbh->prepare_cached("INSERT INTO import_items (import_record_id, status, marcxml)
 483                                         VALUES (?, ?, ?)");
 484         $sth->bind_param(1, $import_record_id);
 485         $sth->bind_param(2, 'staged');
 486         $sth->bind_param(3, $item_marc->as_xml());
 487         $sth->execute();
 488         push @import_items_ids, $dbh->{'mysql_insertid'};
 489         $sth->finish();
 490     }
 491
 492     if ($#import_items_ids > -1) {
 493         _update_batch_record_counts($batch_id) if $update_counts;
 494         _update_import_record_marc($import_record_id, $marc_record, C4::Context->preference('marcflavour'));
 495     }
 496     return @import_items_ids;
 497 }
 498
 499 =head2 BatchFindDuplicates
 500
 501   my $num_with_matches = BatchFindDuplicates($batch_id, $matcher,
 502              $max_matches, $progress_interval, $progress_callback);
 503
 504 Goes through the records loaded in the batch and attempts to
 505 find duplicates for each one.  Sets the matching status
 506 of each record to "no_match" or "auto_match" as appropriate.
 507
 508 The $max_matches parameter is optional; if it is not supplied,
 509 it defaults to 10.
 510
 511 The $progress_interval and $progress_callback parameters are
 512 optional; if both are supplied, the sub referred to by
 513 $progress_callback will be invoked every $progress_interval
 514 records using the number of records processed as the
 515 singular argument.
 516
 517 =cut
 518
 519 sub BatchFindDuplicates {
 520     my $batch_id = shift;
 521     my $matcher = shift;
 522     my $max_matches = @_ ? shift : 10;
 523
 524     # optional callback to monitor status
 525     # of job
 526     my $progress_interval = 0;
 527     my $progress_callback = undef;
 528     if ($#_ == 1) {
 529         $progress_interval = shift;
 530         $progress_callback = shift;
 531         $progress_interval = 0 unless $progress_interval =~ /^\d+$/ and $progress_interval > 0;
 532         $progress_interval = 0 unless 'CODE' eq ref $progress_callback;
 533     }
 534
 535     my $dbh = C4::Context->dbh;
 536
 537     my $sth = $dbh->prepare("SELECT import_record_id, record_type, marc
 538                              FROM import_records
 539                              WHERE import_batch_id = ?");
 540     $sth->execute($batch_id);
 541     my $num_with_matches = 0;
 542     my $rec_num = 0;
 543     while (my $rowref = $sth->fetchrow_hashref) {
 544         $rec_num++;
 545         if ($progress_interval and (0 == ($rec_num % $progress_interval))) {
 546             &$progress_callback($rec_num);
 547         }
 548         my $marc_record = MARC::Record->new_from_usmarc($rowref->{'marc'});
 549         my @matches = ();
 550         if (defined $matcher) {
 551             @matches = $matcher->get_matches($marc_record, $max_matches);
 552         }
 553         if (scalar(@matches) > 0) {
 554             $num_with_matches++;
 555             SetImportRecordMatches($rowref->{'import_record_id'}, @matches);
 556             SetImportRecordOverlayStatus($rowref->{'import_record_id'}, 'auto_match');
 557         } else {
 558             SetImportRecordMatches($rowref->{'import_record_id'}, ());
 559             SetImportRecordOverlayStatus($rowref->{'import_record_id'}, 'no_match');
 560         }
 561     }
 562     $sth->finish();
 563     return $num_with_matches;
 564 }
 565
 566 =head2 BatchCommitRecords
 567
 568   my ($num_added, $num_updated, $num_items_added, $num_items_replaced, $num_items_errored, $num_ignored) =
 569         BatchCommitRecords($batch_id, $framework,
 570         $progress_interval, $progress_callback);
 571
 572 =cut
 573
 574 sub BatchCommitRecords {
 575     my $batch_id = shift;
 576     my $framework = shift;
 577
 578     # optional callback to monitor status
 579     # of job
 580     my $progress_interval = 0;
 581     my $progress_callback = undef;
 582     if ($#_ == 1) {
 583         $progress_interval = shift;
 584         $progress_callback = shift;
 585         $progress_interval = 0 unless $progress_interval =~ /^\d+$/ and $progress_interval > 0;
 586         $progress_interval = 0 unless 'CODE' eq ref $progress_callback;
 587     }
 588
 589     my $record_type;
 590     my $num_added = 0;
 591     my $num_updated = 0;
 592     my $num_items_added = 0;
 593     my $num_items_replaced = 0;
 594     my $num_items_errored = 0;
 595     my $num_ignored = 0;
 596     # commit (i.e., save, all records in the batch)
 597     SetImportBatchStatus('importing');
 598     my $overlay_action = GetImportBatchOverlayAction($batch_id);
 599     my $nomatch_action = GetImportBatchNoMatchAction($batch_id);
 600     my $item_action = GetImportBatchItemAction($batch_id);
 601     my $item_tag;
 602     my $item_subfield;
 603     my $dbh = C4::Context->dbh;
 604     my $sth = $dbh->prepare("SELECT import_records.import_record_id, record_type, status, overlay_status, marc, encoding
 605                              FROM import_records
 606                              LEFT JOIN import_auths ON (import_records.import_record_id=import_auths.import_record_id)
 607                              LEFT JOIN import_biblios ON (import_records.import_record_id=import_biblios.import_record_id)
 608                              WHERE import_batch_id = ?");
 609     $sth->execute($batch_id);
 610     my $marcflavour = C4::Context->preference('marcflavour');
 611     my $rec_num = 0;
 612     while (my $rowref = $sth->fetchrow_hashref) {
 613         $record_type = $rowref->{'record_type'};
 614         $rec_num++;
 615         if ($progress_interval and (0 == ($rec_num % $progress_interval))) {
 616             &$progress_callback($rec_num);
 617         }
 618         if ($rowref->{'status'} eq 'error' or $rowref->{'status'} eq 'imported') {
 619             $num_ignored++;
 620             next;
 621         }
 622
 623         my $marc_type;
 624         if ($marcflavour eq 'UNIMARC' && $record_type eq 'auth') {
 625             $marc_type = 'UNIMARCAUTH';
 626         } elsif ($marcflavour eq 'UNIMARC') {
 627             $marc_type = 'UNIMARC';
 628         } else {
 629             $marc_type = 'USMARC';
 630         }
 631         my $marc_record = MARC::Record->new_from_usmarc($rowref->{'marc'});
 632
 633         if ($record_type eq 'biblio') {
 634             # remove any item tags - rely on BatchCommitItems
 635             ($item_tag,$item_subfield) = &GetMarcFromKohaField("items.itemnumber",'');
 636             foreach my $item_field ($marc_record->field($item_tag)) {
 637                 $marc_record->delete_field($item_field);
 638             }
 639         }
 640
 641         my ($record_result, $item_result, $record_match) =
 642             _get_commit_action($overlay_action, $nomatch_action, $item_action,
 643                                $rowref->{'overlay_status'}, $rowref->{'import_record_id'}, $record_type);
 644
 645         my $recordid;
 646         my $query;
 647         if ($record_result eq 'create_new') {
 648             $num_added++;
 649             if ($record_type eq 'biblio') {
 650                 my $biblioitemnumber;
 651                 ($recordid, $biblioitemnumber) = AddBiblio($marc_record, $framework);
 652                 $query = "UPDATE import_biblios SET matched_biblionumber = ? WHERE import_record_id = ?";
 653                 if ($item_result eq 'create_new' || $item_result eq 'replace') {
 654                     my ($bib_items_added, $bib_items_replaced, $bib_items_errored) = BatchCommitItems($rowref->{'import_record_id'}, $recordid, $item_result);
 655                     $num_items_added += $bib_items_added;
 656                     $num_items_replaced += $bib_items_replaced;
 657                     $num_items_errored += $bib_items_errored;
 658                 }
 659             } else {
 660                 $recordid = AddAuthority($marc_record, undef, GuessAuthTypeCode($marc_record));
 661                 $query = "UPDATE import_auths SET matched_authid = ? WHERE import_record_id = ?";
 662             }
 663             my $sth = $dbh->prepare_cached($query);
 664             $sth->execute($recordid, $rowref->{'import_record_id'});
 665             $sth->finish();
 666             SetImportRecordStatus($rowref->{'import_record_id'}, 'imported');
 667         } elsif ($record_result eq 'replace') {
 668             $num_updated++;
 669             $recordid = $record_match;
 670             my $oldxml;
 671             if ($record_type eq 'biblio') {
 672                 my $oldbiblio = GetBiblio($recordid);
 673                 $oldxml = GetXmlBiblio($recordid);
 674
 675                 # remove item fields so that they don't get
 676                 # added again if record is reverted
 677                 # FIXME: GetXmlBiblio output should not contain item info any more! So the next foreach should not be needed. Does not hurt either; may remove old 952s that should not have been there anymore.
 678                 my $old_marc = MARC::Record->new_from_xml(StripNonXmlChars($oldxml), 'UTF-8', $rowref->{'encoding'}, $marc_type);
 679                 foreach my $item_field ($old_marc->field($item_tag)) {
 680                     $old_marc->delete_field($item_field);
 681                 }
 682                 $oldxml = $old_marc->as_xml($marc_type);
 683
 684                 ModBiblio($marc_record, $recordid, $oldbiblio->{'frameworkcode'});
 685                 $query = "UPDATE import_biblios SET matched_biblionumber = ? WHERE import_record_id = ?";
 686
 687                 if ($item_result eq 'create_new' || $item_result eq 'replace') {
 688                     my ($bib_items_added, $bib_items_replaced, $bib_items_errored) = BatchCommitItems($rowref->{'import_record_id'}, $recordid, $item_result);
 689                     $num_items_added += $bib_items_added;
 690                     $num_items_replaced += $bib_items_replaced;
 691                     $num_items_errored += $bib_items_errored;
 692                 }
 693             } else {
 694                 $oldxml = GetAuthorityXML($recordid);
 695
 696                 ModAuthority($recordid, $marc_record, GuessAuthTypeCode($marc_record));
 697                 $query = "UPDATE import_auths SET matched_authid = ? WHERE import_record_id = ?";
 698             }
 699             my $sth = $dbh->prepare_cached("UPDATE import_records SET marcxml_old = ? WHERE import_record_id = ?");
 700             $sth->execute($oldxml, $rowref->{'import_record_id'});
 701             $sth->finish();
 702             my $sth2 = $dbh->prepare_cached($query);
 703             $sth2->execute($recordid, $rowref->{'import_record_id'});
 704             $sth2->finish();
 705             SetImportRecordOverlayStatus($rowref->{'import_record_id'}, 'match_applied');
 706             SetImportRecordStatus($rowref->{'import_record_id'}, 'imported');
 707         } elsif ($record_result eq 'ignore') {
 708             $recordid = $record_match;
 709             $num_ignored++;
 710             $recordid = $record_match;
 711             if ($record_type eq 'biblio' and defined $recordid and ( $item_result eq 'create_new' || $item_result eq 'replace' ) ) {
 712                 my ($bib_items_added, $bib_items_replaced, $bib_items_errored) = BatchCommitItems($rowref->{'import_record_id'}, $recordid, $item_result);
 713                 $num_items_added += $bib_items_added;
 714          $num_items_replaced += $bib_items_replaced;
 715                 $num_items_errored += $bib_items_errored;
 716                 # still need to record the matched biblionumber so that the
 717                 # items can be reverted
 718                 my $sth2 = $dbh->prepare_cached("UPDATE import_biblios SET matched_biblionumber = ? WHERE import_record_id = ?");
 719                 $sth2->execute($recordid, $rowref->{'import_record_id'});
 720                 SetImportRecordOverlayStatus($rowref->{'import_record_id'}, 'match_applied');
 721             }
 722             SetImportRecordStatus($rowref->{'import_record_id'}, 'ignored');
 723         }
 724     }
 725     $sth->finish();
 726     SetImportBatchStatus($batch_id, 'imported');
 727     return ($num_added, $num_updated, $num_items_added, $num_items_replaced, $num_items_errored, $num_ignored);
 728 }
 729
 730 =head2 BatchCommitItems
 731
 732   ($num_items_added, $num_items_errored) =
 733          BatchCommitItems($import_record_id, $biblionumber);
 734
 735 =cut
 736
 737 sub BatchCommitItems {
 738     my ( $import_record_id, $biblionumber, $action ) = @_;
 739
 740     my $dbh = C4::Context->dbh;
 741
 742     my $num_items_added = 0;
 743     my $num_items_errored = 0;
 744     my $num_items_replaced = 0;
 745
 746     my $sth = $dbh->prepare( "
 747         SELECT import_items_id, import_items.marcxml, encoding
 748         FROM import_items
 749         JOIN import_records USING (import_record_id)
 750         WHERE import_record_id = ?
 751         ORDER BY import_items_id
 752     " );
 753     $sth->bind_param( 1, $import_record_id );
 754     $sth->execute();
 755
 756     while ( my $row = $sth->fetchrow_hashref() ) {
 757         my $item_marc = MARC::Record->new_from_xml( StripNonXmlChars( $row->{'marcxml'} ), 'UTF-8', $row->{'encoding'} );
 758
 759         # Delete date_due subfield as to not accidentally delete item checkout due dates
 760         my ( $MARCfield, $MARCsubfield ) = GetMarcFromKohaField( 'items.onloan', GetFrameworkCode($biblionumber) );
 761         $item_marc->field($MARCfield)->delete_subfield( code => $MARCsubfield );
 762
 763         my $item = TransformMarcToKoha( $dbh, $item_marc );
 764
 765         my $duplicate_barcode = exists( $item->{'barcode'} ) && GetItemnumberFromBarcode( $item->{'barcode'} );
 766         my $duplicate_itemnumber = exists( $item->{'itemnumber'} );
 767
 768         my $updsth = $dbh->prepare("UPDATE import_items SET status = ?, itemnumber = ? WHERE import_items_id = ?");
 769         if ( $action eq "replace" && $duplicate_itemnumber ) {
 770             # Duplicate itemnumbers have precedence, that way we can update barcodes by overlaying
 771             ModItemFromMarc( $item_marc, $biblionumber, $item->{itemnumber} );
 772             $updsth->bind_param( 1, 'imported' );
 773             $updsth->bind_param( 2, $item->{itemnumber} );
 774             $updsth->bind_param( 3, $row->{'import_items_id'} );
 775             $updsth->execute();
 776             $updsth->finish();
 777             $num_items_replaced++;
 778         } elsif ( $action eq "replace" && $duplicate_barcode ) {
 779             my $itemnumber = GetItemnumberFromBarcode( $item->{'barcode'} );
 780             ModItemFromMarc( $item_marc, $biblionumber, $itemnumber );
 781             $updsth->bind_param( 1, 'imported' );
 782             $updsth->bind_param( 2, $item->{itemnumber} );
 783             $updsth->bind_param( 3, $row->{'import_items_id'} );
 784             $updsth->execute();
 785             $updsth->finish();
 786             $num_items_replaced++;
 787         } elsif ($duplicate_barcode) {
 788             $updsth->bind_param( 1, 'error' );
 789             $updsth->bind_param( 2, 'duplicate item barcode' );
 790             $updsth->bind_param( 3, $row->{'import_items_id'} );
 791             $updsth->execute();
 792             $num_items_errored++;
 793         } else {
 794             my ( $item_biblionumber, $biblioitemnumber, $itemnumber ) = AddItemFromMarc( $item_marc, $biblionumber );
 795             $updsth->bind_param( 1, 'imported' );
 796             $updsth->bind_param( 2, $itemnumber );
 797             $updsth->bind_param( 3, $row->{'import_items_id'} );
 798             $updsth->execute();
 799             $updsth->finish();
 800             $num_items_added++;
 801         }
 802     }
 803
 804     return ( $num_items_added, $num_items_replaced, $num_items_errored );
 805 }
 806
 807 =head2 BatchRevertRecords
 808
 809   my ($num_deleted, $num_errors, $num_reverted, $num_items_deleted,
 810       $num_ignored) = BatchRevertRecords($batch_id);
 811
 812 =cut
 813
 814 sub BatchRevertRecords {
 815     my $batch_id = shift;
 816
 817     $logger->trace("C4::ImportBatch::BatchRevertRecords( $batch_id )");
 818
 819     my $record_type;
 820     my $num_deleted = 0;
 821     my $num_errors = 0;
 822     my $num_reverted = 0;
 823     my $num_ignored = 0;
 824     my $num_items_deleted = 0;
 825     # commit (i.e., save, all records in the batch)
 826     SetImportBatchStatus('reverting');
 827     my $overlay_action = GetImportBatchOverlayAction($batch_id);
 828     my $nomatch_action = GetImportBatchNoMatchAction($batch_id);
 829     my $dbh = C4::Context->dbh;
 830     my $sth = $dbh->prepare("SELECT import_records.import_record_id, record_type, status, overlay_status, marcxml_old, encoding, matched_biblionumber, matched_authid
 831                              FROM import_records
 832                              LEFT JOIN import_auths ON (import_records.import_record_id=import_auths.import_record_id)
 833                              LEFT JOIN import_biblios ON (import_records.import_record_id=import_biblios.import_record_id)
 834                              WHERE import_batch_id = ?");
 835     $sth->execute($batch_id);
 836     my $marc_type;
 837     my $marcflavour = C4::Context->preference('marcflavour');
 838     while (my $rowref = $sth->fetchrow_hashref) {
 839         $record_type = $rowref->{'record_type'};
 840         if ($rowref->{'status'} eq 'error' or $rowref->{'status'} eq 'reverted') {
 841             $num_ignored++;
 842             next;
 843         }
 844         if ($marcflavour eq 'UNIMARC' && $record_type eq 'auth') {
 845             $marc_type = 'UNIMARCAUTH';
 846         } elsif ($marcflavour eq 'UNIMARC') {
 847             $marc_type = 'UNIMARC';
 848         } else {
 849             $marc_type = 'USMARC';
 850         }
 851
 852         my $record_result = _get_revert_action($overlay_action, $rowref->{'overlay_status'}, $rowref->{'status'});
 853
 854         if ($record_result eq 'delete') {
 855             my $error = undef;
 856             if  ($record_type eq 'biblio') {
 857                 $num_items_deleted += BatchRevertItems($rowref->{'import_record_id'}, $rowref->{'matched_biblionumber'});
 858                 $error = DelBiblio($rowref->{'matched_biblionumber'});
 859             } else {
 860                 my $deletedauthid = DelAuthority($rowref->{'matched_authid'});
 861             }
 862             if (defined $error) {
 863                 $num_errors++;
 864             } else {
 865                 $num_deleted++;
 866                 SetImportRecordStatus($rowref->{'import_record_id'}, 'reverted');
 867             }
 868         } elsif ($record_result eq 'restore') {
 869             $num_reverted++;
 870             my $old_record = MARC::Record->new_from_xml(StripNonXmlChars($rowref->{'marcxml_old'}), 'UTF-8', $rowref->{'encoding'}, $marc_type);
 871             if ($record_type eq 'biblio') {
 872                 my $biblionumber = $rowref->{'matched_biblionumber'};
 873                 my $oldbiblio = GetBiblio($biblionumber);
 874
 875                 $logger->info("C4::ImportBatch::BatchRevertRecords: Biblio record $biblionumber does not exist, restoration of this record was skipped") unless $oldbiblio;
 876                 next unless $oldbiblio; # Record has since been deleted. Deleted records should stay deleted.
 877
 878                 $num_items_deleted += BatchRevertItems($rowref->{'import_record_id'}, $rowref->{'matched_biblionumber'});
 879                 ModBiblio($old_record, $biblionumber, $oldbiblio->{'frameworkcode'});
 880             } else {
 881                 my $authid = $rowref->{'matched_authid'};
 882                 ModAuthority($authid, $old_record, GuessAuthTypeCode($old_record));
 883             }
 884             SetImportRecordStatus($rowref->{'import_record_id'}, 'reverted');
 885         } elsif ($record_result eq 'ignore') {
 886             if ($record_type eq 'biblio') {
 887                 $num_items_deleted += BatchRevertItems($rowref->{'import_record_id'}, $rowref->{'matched_biblionumber'});
 888             }
 889             SetImportRecordStatus($rowref->{'import_record_id'}, 'reverted');
 890         }
 891         my $query;
 892         if ($record_type eq 'biblio') {
 893             # remove matched_biblionumber only if there is no 'imported' item left
 894             $query = "UPDATE import_biblios SET matched_biblionumber = NULL WHERE import_record_id = ?";
 895             $query = "UPDATE import_biblios SET matched_biblionumber = NULL WHERE import_record_id = ?  AND NOT EXISTS (SELECT * FROM import_items WHERE import_items.import_record_id=import_biblios.import_record_id and status='imported')";
 896         } else {
 897             $query = "UPDATE import_auths SET matched_authid = NULL WHERE import_record_id = ?";
 898         }
 899         my $sth2 = $dbh->prepare_cached($query);
 900         $sth2->execute($rowref->{'import_record_id'});
 901     }
 902
 903     $sth->finish();
 904     SetImportBatchStatus($batch_id, 'reverted');
 905     return ($num_deleted, $num_errors, $num_reverted, $num_items_deleted, $num_ignored);
 906 }
 907
 908 =head2 BatchRevertItems
 909
 910   my $num_items_deleted = BatchRevertItems($import_record_id, $biblionumber);
 911
 912 =cut
 913
 914 sub BatchRevertItems {
 915     my ($import_record_id, $biblionumber) = @_;
 916
 917     my $dbh = C4::Context->dbh;
 918     my $num_items_deleted = 0;
 919
 920     my $sth = $dbh->prepare_cached("SELECT import_items_id, itemnumber
 921                                    FROM import_items
 922                                    JOIN items USING (itemnumber)
 923                                    WHERE import_record_id = ?");
 924     $sth->bind_param(1, $import_record_id);
 925     $sth->execute();
 926     while (my $row = $sth->fetchrow_hashref()) {
 927         my $error = DelItemCheck($dbh, $biblionumber, $row->{'itemnumber'});
 928         if ($error == 1){
 929             my $updsth = $dbh->prepare("UPDATE import_items SET status = ? WHERE import_items_id = ?");
 930             $updsth->bind_param(1, 'reverted');
 931             $updsth->bind_param(2, $row->{'import_items_id'});
 932             $updsth->execute();
 933             $updsth->finish();
 934             $num_items_deleted++;
 935         }
 936         else {
 937             next;
 938         }
 939     }
 940     $sth->finish();
 941     return $num_items_deleted;
 942 }
 943
 944 =head2 CleanBatch
 945
 946   CleanBatch($batch_id)
 947
 948 Deletes all staged records from the import batch
 949 and sets the status of the batch to 'cleaned'.  Note
 950 that deleting a stage record does *not* affect
 951 any record that has been committed to the database.
 952
 953 =cut
 954
 955 sub CleanBatch {
 956     my $batch_id = shift;
 957     return unless defined $batch_id;
 958
 959     C4::Context->dbh->do('DELETE FROM import_records WHERE import_batch_id = ?', {}, $batch_id);
 960     SetImportBatchStatus($batch_id, 'cleaned');
 961 }
 962
 963 =head2 GetAllImportBatches
 964
 965   my $results = GetAllImportBatches();
 966
 967 Returns a references to an array of hash references corresponding
 968 to all import_batches rows (of batch_type 'batch'), sorted in
 969 ascending order by import_batch_id.
 970
 971 =cut
 972
 973 sub  GetAllImportBatches {
 974     my $dbh = C4::Context->dbh;
 975     my $sth = $dbh->prepare_cached("SELECT * FROM import_batches
 976                                     WHERE batch_type IN ('batch', 'webservice')
 977                                     ORDER BY import_batch_id ASC");
 978
 979     my $results = [];
 980     $sth->execute();
 981     while (my $row = $sth->fetchrow_hashref) {
 982         push @$results, $row;
 983     }
 984     $sth->finish();
 985     return $results;
 986 }
 987
 988 =head2 GetStagedWebserviceBatches
 989
 990   my $batch_ids = GetStagedWebserviceBatches();
 991
 992 Returns a references to an array of batch id's
 993 of batch_type 'webservice' that are not imported
 994
 995 =cut
 996
 997 my $PENDING_WEBSERVICE_BATCHES_QRY = <<EOQ;
 998 SELECT import_batch_id FROM import_batches
 999 WHERE batch_type = 'webservice'
1000 AND import_status = 'staged'
1001 EOQ
1002 sub  GetStagedWebserviceBatches {
1003     my $dbh = C4::Context->dbh;
1004     return $dbh->selectcol_arrayref($PENDING_WEBSERVICE_BATCHES_QRY);
1005 }
1006
1007 =head2 GetImportBatchRangeDesc
1008
1009   my $results = GetImportBatchRangeDesc($offset, $results_per_group);
1010
1011 Returns a reference to an array of hash references corresponding to
1012 import_batches rows (sorted in descending order by import_batch_id)
1013 start at the given offset.
1014
1015 =cut
1016
1017 sub GetImportBatchRangeDesc {
1018     my ($offset, $results_per_group) = @_;
1019
1020     my $dbh = C4::Context->dbh;
1021     my $query = "SELECT * FROM import_batches
1022                                     WHERE batch_type IN ('batch', 'webservice')
1023                                     ORDER BY import_batch_id DESC";
1024     my @params;
1025     if ($results_per_group){
1026         $query .= " LIMIT ?";
1027         push(@params, $results_per_group);
1028     }
1029     if ($offset){
1030         $query .= " OFFSET ?";
1031         push(@params, $offset);
1032     }
1033     my $sth = $dbh->prepare_cached($query);
1034     $sth->execute(@params);
1035     my $results = $sth->fetchall_arrayref({});
1036     $sth->finish();
1037     return $results;
1038 }
1039
1040 =head2 GetItemNumbersFromImportBatch
1041
1042   my @itemsnos = GetItemNumbersFromImportBatch($batch_id);
1043
1044 =cut
1045
1046 sub GetItemNumbersFromImportBatch {
1047         my ($batch_id) = @_;
1048         my $dbh = C4::Context->dbh;
1049         my $sth = $dbh->prepare("SELECT itemnumber FROM import_batches,import_records,import_items WHERE import_batches.import_batch_id=import_records.import_batch_id AND import_records.import_record_id=import_items.import_record_id AND import_batches.import_batch_id=?");
1050         $sth->execute($batch_id);
1051         my @items ;
1052         while ( my ($itm) = $sth->fetchrow_array ) {
1053                 push @items, $itm;
1054         }
1055         return @items;
1056 }
1057
1058 =head2 GetNumberOfImportBatches
1059
1060   my $count = GetNumberOfImportBatches();
1061
1062 =cut
1063
1064 sub GetNumberOfNonZ3950ImportBatches {
1065     my $dbh = C4::Context->dbh;
1066     my $sth = $dbh->prepare("SELECT COUNT(*) FROM import_batches WHERE batch_type != 'z3950'");
1067     $sth->execute();
1068     my ($count) = $sth->fetchrow_array();
1069     $sth->finish();
1070     return $count;
1071 }
1072
1073 =head2 GetImportBiblios
1074
1075   my $results = GetImportBiblios($importid);
1076
1077 =cut
1078
1079 sub GetImportBiblios {
1080     my ($import_record_id) = @_;
1081
1082     my $dbh = C4::Context->dbh;
1083     my $query = "SELECT * FROM import_biblios WHERE import_record_id = ?";
1084     return $dbh->selectall_arrayref(
1085         $query,
1086         { Slice => {} },
1087         $import_record_id
1088     );
1089
1090 }
1091
1092 =head2 GetImportRecordsRange
1093
1094   my $results = GetImportRecordsRange($batch_id, $offset, $results_per_group);
1095
1096 Returns a reference to an array of hash references corresponding to
1097 import_biblios/import_auths/import_records rows for a given batch
1098 starting at the given offset.
1099
1100 =cut
1101
1102 sub GetImportRecordsRange {
1103     my ( $batch_id, $offset, $results_per_group, $status, $parameters ) = @_;
1104
1105     my $dbh = C4::Context->dbh;
1106
1107     my $order_by = $parameters->{order_by} || 'import_record_id';
1108     ( $order_by ) = grep( /^$order_by$/, qw( import_record_id title status overlay_status ) ) ? $order_by : 'import_record_id';
1109
1110     my $order_by_direction =
1111       uc( $parameters->{order_by_direction} ) eq 'DESC' ? 'DESC' : 'ASC';
1112
1113     $order_by .= " $order_by_direction, authorized_heading" if $order_by eq 'title';
1114
1115     my $query = "SELECT title, author, isbn, issn, authorized_heading, import_records.import_record_id,
1116                                            record_sequence, status, overlay_status,
1117                                            matched_biblionumber, matched_authid, record_type
1118                                     FROM   import_records
1119                                     LEFT JOIN import_auths ON (import_records.import_record_id=import_auths.import_record_id)
1120                                     LEFT JOIN import_biblios ON (import_records.import_record_id=import_biblios.import_record_id)
1121                                     WHERE  import_batch_id = ?";
1122     my @params;
1123     push(@params, $batch_id);
1124     if ($status) {
1125         $query .= " AND status=?";
1126         push(@params,$status);
1127     }
1128
1129     $query.=" ORDER BY $order_by $order_by_direction";
1130
1131     if($results_per_group){
1132         $query .= " LIMIT ?";
1133         push(@params, $results_per_group);
1134     }
1135     if($offset){
1136         $query .= " OFFSET ?";
1137         push(@params, $offset);
1138     }
1139     my $sth = $dbh->prepare_cached($query);
1140     $sth->execute(@params);
1141     my $results = $sth->fetchall_arrayref({});
1142     $sth->finish();
1143     return $results;
1144
1145 }
1146
1147 =head2 GetBestRecordMatch
1148
1149   my $record_id = GetBestRecordMatch($import_record_id);
1150
1151 =cut
1152
1153 sub GetBestRecordMatch {
1154     my ($import_record_id) = @_;
1155
1156     my $dbh = C4::Context->dbh;
1157     my $sth = $dbh->prepare("SELECT candidate_match_id
1158                              FROM   import_record_matches
1159                              JOIN   import_records ON ( import_record_matches.import_record_id = import_records.import_record_id )
1160                              LEFT JOIN biblio ON ( candidate_match_id = biblio.biblionumber )
1161                              LEFT JOIN auth_header ON ( candidate_match_id = auth_header.authid )
1162                              WHERE  import_record_matches.import_record_id = ? AND
1163                              (  (import_records.record_type = 'biblio' AND biblio.biblionumber IS NOT NULL) OR
1164                                 (import_records.record_type = 'auth' AND auth_header.authid IS NOT NULL) )
1165                              ORDER BY score DESC, candidate_match_id DESC");
1166     $sth->execute($import_record_id);
1167     my ($record_id) = $sth->fetchrow_array();
1168     $sth->finish();
1169     return $record_id;
1170 }
1171
1172 =head2 GetImportBatchStatus
1173
1174   my $status = GetImportBatchStatus($batch_id);
1175
1176 =cut
1177
1178 sub GetImportBatchStatus {
1179     my ($batch_id) = @_;
1180
1181     my $dbh = C4::Context->dbh;
1182     my $sth = $dbh->prepare("SELECT import_status FROM import_batches WHERE import_batch_id = ?");
1183     $sth->execute($batch_id);
1184     my ($status) = $sth->fetchrow_array();
1185     $sth->finish();
1186     return $status;
1187
1188 }
1189
1190 =head2 SetImportBatchStatus
1191
1192   SetImportBatchStatus($batch_id, $new_status);
1193
1194 =cut
1195
1196 sub SetImportBatchStatus {
1197     my ($batch_id, $new_status) = @_;
1198
1199     my $dbh = C4::Context->dbh;
1200     my $sth = $dbh->prepare("UPDATE import_batches SET import_status = ? WHERE import_batch_id = ?");
1201     $sth->execute($new_status, $batch_id);
1202     $sth->finish();
1203
1204 }
1205
1206 =head2 GetImportBatchOverlayAction
1207
1208   my $overlay_action = GetImportBatchOverlayAction($batch_id);
1209
1210 =cut
1211
1212 sub GetImportBatchOverlayAction {
1213     my ($batch_id) = @_;
1214
1215     my $dbh = C4::Context->dbh;
1216     my $sth = $dbh->prepare("SELECT overlay_action FROM import_batches WHERE import_batch_id = ?");
1217     $sth->execute($batch_id);
1218     my ($overlay_action) = $sth->fetchrow_array();
1219     $sth->finish();
1220     return $overlay_action;
1221
1222 }
1223
1224
1225 =head2 SetImportBatchOverlayAction
1226
1227   SetImportBatchOverlayAction($batch_id, $new_overlay_action);
1228
1229 =cut
1230
1231 sub SetImportBatchOverlayAction {
1232     my ($batch_id, $new_overlay_action) = @_;
1233
1234     my $dbh = C4::Context->dbh;
1235     my $sth = $dbh->prepare("UPDATE import_batches SET overlay_action = ? WHERE import_batch_id = ?");
1236     $sth->execute($new_overlay_action, $batch_id);
1237     $sth->finish();
1238
1239 }
1240
1241 =head2 GetImportBatchNoMatchAction
1242
1243   my $nomatch_action = GetImportBatchNoMatchAction($batch_id);
1244
1245 =cut
1246
1247 sub GetImportBatchNoMatchAction {
1248     my ($batch_id) = @_;
1249
1250     my $dbh = C4::Context->dbh;
1251     my $sth = $dbh->prepare("SELECT nomatch_action FROM import_batches WHERE import_batch_id = ?");
1252     $sth->execute($batch_id);
1253     my ($nomatch_action) = $sth->fetchrow_array();
1254     $sth->finish();
1255     return $nomatch_action;
1256
1257 }
1258
1259
1260 =head2 SetImportBatchNoMatchAction
1261
1262   SetImportBatchNoMatchAction($batch_id, $new_nomatch_action);
1263
1264 =cut
1265
1266 sub SetImportBatchNoMatchAction {
1267     my ($batch_id, $new_nomatch_action) = @_;
1268
1269     my $dbh = C4::Context->dbh;
1270     my $sth = $dbh->prepare("UPDATE import_batches SET nomatch_action = ? WHERE import_batch_id = ?");
1271     $sth->execute($new_nomatch_action, $batch_id);
1272     $sth->finish();
1273
1274 }
1275
1276 =head2 GetImportBatchItemAction
1277
1278   my $item_action = GetImportBatchItemAction($batch_id);
1279
1280 =cut
1281
1282 sub GetImportBatchItemAction {
1283     my ($batch_id) = @_;
1284
1285     my $dbh = C4::Context->dbh;
1286     my $sth = $dbh->prepare("SELECT item_action FROM import_batches WHERE import_batch_id = ?");
1287     $sth->execute($batch_id);
1288     my ($item_action) = $sth->fetchrow_array();
1289     $sth->finish();
1290     return $item_action;
1291
1292 }
1293
1294
1295 =head2 SetImportBatchItemAction
1296
1297   SetImportBatchItemAction($batch_id, $new_item_action);
1298
1299 =cut
1300
1301 sub SetImportBatchItemAction {
1302     my ($batch_id, $new_item_action) = @_;
1303
1304     my $dbh = C4::Context->dbh;
1305     my $sth = $dbh->prepare("UPDATE import_batches SET item_action = ? WHERE import_batch_id = ?");
1306     $sth->execute($new_item_action, $batch_id);
1307     $sth->finish();
1308
1309 }
1310
1311 =head2 GetImportBatchMatcher
1312
1313   my $matcher_id = GetImportBatchMatcher($batch_id);
1314
1315 =cut
1316
1317 sub GetImportBatchMatcher {
1318     my ($batch_id) = @_;
1319
1320     my $dbh = C4::Context->dbh;
1321     my $sth = $dbh->prepare("SELECT matcher_id FROM import_batches WHERE import_batch_id = ?");
1322     $sth->execute($batch_id);
1323     my ($matcher_id) = $sth->fetchrow_array();
1324     $sth->finish();
1325     return $matcher_id;
1326
1327 }
1328
1329
1330 =head2 SetImportBatchMatcher
1331
1332   SetImportBatchMatcher($batch_id, $new_matcher_id);
1333
1334 =cut
1335
1336 sub SetImportBatchMatcher {
1337     my ($batch_id, $new_matcher_id) = @_;
1338
1339     my $dbh = C4::Context->dbh;
1340     my $sth = $dbh->prepare("UPDATE import_batches SET matcher_id = ? WHERE import_batch_id = ?");
1341     $sth->execute($new_matcher_id, $batch_id);
1342     $sth->finish();
1343
1344 }
1345
1346 =head2 GetImportRecordOverlayStatus
1347
1348   my $overlay_status = GetImportRecordOverlayStatus($import_record_id);
1349
1350 =cut
1351
1352 sub GetImportRecordOverlayStatus {
1353     my ($import_record_id) = @_;
1354
1355     my $dbh = C4::Context->dbh;
1356     my $sth = $dbh->prepare("SELECT overlay_status FROM import_records WHERE import_record_id = ?");
1357     $sth->execute($import_record_id);
1358     my ($overlay_status) = $sth->fetchrow_array();
1359     $sth->finish();
1360     return $overlay_status;
1361
1362 }
1363
1364
1365 =head2 SetImportRecordOverlayStatus
1366
1367   SetImportRecordOverlayStatus($import_record_id, $new_overlay_status);
1368
1369 =cut
1370
1371 sub SetImportRecordOverlayStatus {
1372     my ($import_record_id, $new_overlay_status) = @_;
1373
1374     my $dbh = C4::Context->dbh;
1375     my $sth = $dbh->prepare("UPDATE import_records SET overlay_status = ? WHERE import_record_id = ?");
1376     $sth->execute($new_overlay_status, $import_record_id);
1377     $sth->finish();
1378
1379 }
1380
1381 =head2 GetImportRecordStatus
1382
1383   my $status = GetImportRecordStatus($import_record_id);
1384
1385 =cut
1386
1387 sub GetImportRecordStatus {
1388     my ($import_record_id) = @_;
1389
1390     my $dbh = C4::Context->dbh;
1391     my $sth = $dbh->prepare("SELECT status FROM import_records WHERE import_record_id = ?");
1392     $sth->execute($import_record_id);
1393     my ($status) = $sth->fetchrow_array();
1394     $sth->finish();
1395     return $status;
1396
1397 }
1398
1399
1400 =head2 SetImportRecordStatus
1401
1402   SetImportRecordStatus($import_record_id, $new_status);
1403
1404 =cut
1405
1406 sub SetImportRecordStatus {
1407     my ($import_record_id, $new_status) = @_;
1408
1409     my $dbh = C4::Context->dbh;
1410     my $sth = $dbh->prepare("UPDATE import_records SET status = ? WHERE import_record_id = ?");
1411     $sth->execute($new_status, $import_record_id);
1412     $sth->finish();
1413
1414 }
1415
1416 =head2 GetImportRecordMatches
1417
1418   my $results = GetImportRecordMatches($import_record_id, $best_only);
1419
1420 =cut
1421
1422 sub GetImportRecordMatches {
1423     my $import_record_id = shift;
1424     my $best_only = @_ ? shift : 0;
1425
1426     my $dbh = C4::Context->dbh;
1427     # FIXME currently biblio only
1428     my $sth = $dbh->prepare_cached("SELECT title, author, biblionumber,
1429                                     candidate_match_id, score, record_type
1430                                     FROM import_records
1431                                     JOIN import_record_matches USING (import_record_id)
1432                                     LEFT JOIN biblio ON (biblionumber = candidate_match_id)
1433                                     WHERE import_record_id = ?
1434                                     ORDER BY score DESC, biblionumber DESC");
1435     $sth->bind_param(1, $import_record_id);
1436     my $results = [];
1437     $sth->execute();
1438     while (my $row = $sth->fetchrow_hashref) {
1439         if ($row->{'record_type'} eq 'auth') {
1440             $row->{'authorized_heading'} = C4::AuthoritiesMarc::GetAuthorizedHeading( { authid => $row->{'candidate_match_id'} } );
1441         }
1442         next if ($row->{'record_type'} eq 'biblio' && not $row->{'biblionumber'});
1443         push @$results, $row;
1444         last if $best_only;
1445     }
1446     $sth->finish();
1447
1448     return $results;
1449
1450 }
1451
1452
1453 =head2 SetImportRecordMatches
1454
1455   SetImportRecordMatches($import_record_id, @matches);
1456
1457 =cut
1458
1459 sub SetImportRecordMatches {
1460     my $import_record_id = shift;
1461     my @matches = @_;
1462
1463     my $dbh = C4::Context->dbh;
1464     my $delsth = $dbh->prepare("DELETE FROM import_record_matches WHERE import_record_id = ?");
1465     $delsth->execute($import_record_id);
1466     $delsth->finish();
1467
1468     my $sth = $dbh->prepare("INSERT INTO import_record_matches (import_record_id, candidate_match_id, score)
1469                                     VALUES (?, ?, ?)");
1470     foreach my $match (@matches) {
1471         $sth->execute($import_record_id, $match->{'record_id'}, $match->{'score'});
1472     }
1473 }
1474
1475
1476 # internal functions
1477
1478 sub _create_import_record {
1479     my ($batch_id, $record_sequence, $marc_record, $record_type, $encoding, $z3950random, $marc_type) = @_;
1480
1481     my $dbh = C4::Context->dbh;
1482     my $sth = $dbh->prepare("INSERT INTO import_records (import_batch_id, record_sequence, marc, marcxml,
1483                                                          record_type, encoding, z3950random)
1484                                     VALUES (?, ?, ?, ?, ?, ?, ?)");
1485     $sth->execute($batch_id, $record_sequence, $marc_record->as_usmarc(), $marc_record->as_xml($marc_type),
1486                   $record_type, $encoding, $z3950random);
1487     my $import_record_id = $dbh->{'mysql_insertid'};
1488     $sth->finish();
1489     return $import_record_id;
1490 }
1491
1492 sub _update_import_record_marc {
1493     my ($import_record_id, $marc_record, $marc_type) = @_;
1494
1495     my $dbh = C4::Context->dbh;
1496     my $sth = $dbh->prepare("UPDATE import_records SET marc = ?, marcxml = ?
1497                              WHERE  import_record_id = ?");
1498     $sth->execute($marc_record->as_usmarc(), $marc_record->as_xml($marc_type), $import_record_id);
1499     $sth->finish();
1500 }
1501
1502 sub _add_auth_fields {
1503     my ($import_record_id, $marc_record) = @_;
1504
1505     my $controlnumber;
1506     if ($marc_record->field('001')) {
1507         $controlnumber = $marc_record->field('001')->data();
1508     }
1509     my $authorized_heading = C4::AuthoritiesMarc::GetAuthorizedHeading({ record => $marc_record });
1510     my $dbh = C4::Context->dbh;
1511     my $sth = $dbh->prepare("INSERT INTO import_auths (import_record_id, control_number, authorized_heading) VALUES (?, ?, ?)");
1512     $sth->execute($import_record_id, $controlnumber, $authorized_heading);
1513     $sth->finish();
1514 }
1515
1516 sub _add_biblio_fields {
1517     my ($import_record_id, $marc_record) = @_;
1518
1519     my ($title, $author, $isbn, $issn) = _parse_biblio_fields($marc_record);
1520     my $dbh = C4::Context->dbh;
1521     # FIXME no controlnumber, originalsource
1522     $isbn = C4::Koha::GetNormalizedISBN($isbn);
1523     my $sth = $dbh->prepare("INSERT INTO import_biblios (import_record_id, title, author, isbn, issn) VALUES (?, ?, ?, ?, ?)");
1524     $sth->execute($import_record_id, $title, $author, $isbn, $issn);
1525     $sth->finish();
1526
1527 }
1528
1529 sub _update_biblio_fields {
1530     my ($import_record_id, $marc_record) = @_;
1531
1532     my ($title, $author, $isbn, $issn) = _parse_biblio_fields($marc_record);
1533     my $dbh = C4::Context->dbh;
1534     # FIXME no controlnumber, originalsource
1535     # FIXME 2 - should regularize normalization of ISBN wherever it is done
1536     $isbn =~ s/\(.*$//;
1537     $isbn =~ tr/ -_//;
1538     $isbn = uc $isbn;
1539     my $sth = $dbh->prepare("UPDATE import_biblios SET title = ?, author = ?, isbn = ?, issn = ?
1540                              WHERE  import_record_id = ?");
1541     $sth->execute($title, $author, $isbn, $issn, $import_record_id);
1542     $sth->finish();
1543 }
1544
1545 sub _parse_biblio_fields {
1546     my ($marc_record) = @_;
1547
1548     my $dbh = C4::Context->dbh;
1549     my $bibliofields = TransformMarcToKoha($dbh, $marc_record, '');
1550     return ($bibliofields->{'title'}, $bibliofields->{'author'}, $bibliofields->{'isbn'}, $bibliofields->{'issn'});
1551
1552 }
1553
1554 sub _update_batch_record_counts {
1555     my ($batch_id) = @_;
1556
1557     my $dbh = C4::Context->dbh;
1558     my $sth = $dbh->prepare_cached("UPDATE import_batches SET
1559                                         num_records = (
1560                                             SELECT COUNT(*)
1561                                             FROM import_records
1562                                             WHERE import_batch_id = import_batches.import_batch_id),
1563                                         num_items = (
1564                                             SELECT COUNT(*)
1565                                             FROM import_records
1566                                             JOIN import_items USING (import_record_id)
1567                                             WHERE import_batch_id = import_batches.import_batch_id
1568                                             AND record_type = 'biblio')
1569                                     WHERE import_batch_id = ?");
1570     $sth->bind_param(1, $batch_id);
1571     $sth->execute();
1572     $sth->finish();
1573 }
1574
1575 sub _get_commit_action {
1576     my ($overlay_action, $nomatch_action, $item_action, $overlay_status, $import_record_id, $record_type) = @_;
1577
1578     if ($record_type eq 'biblio') {
1579         my ($bib_result, $bib_match, $item_result);
1580
1581         if ($overlay_status ne 'no_match') {
1582             $bib_match = GetBestRecordMatch($import_record_id);
1583             if ($overlay_action eq 'replace') {
1584                 $bib_result  = defined($bib_match) ? 'replace' : 'create_new';
1585             } elsif ($overlay_action eq 'create_new') {
1586                 $bib_result  = 'create_new';
1587             } elsif ($overlay_action eq 'ignore') {
1588                 $bib_result  = 'ignore';
1589             }
1590          if($item_action eq 'always_add' or $item_action eq 'add_only_for_matches'){
1591                 $item_result = 'create_new';
1592        }
1593       elsif($item_action eq 'replace'){
1594           $item_result = 'replace';
1595           }
1596       else {
1597              $item_result = 'ignore';
1598            }
1599         } else {
1600             $bib_result = $nomatch_action;
1601             $item_result = ($item_action eq 'always_add' or $item_action eq 'add_only_for_new')     ? 'create_new' : 'ignore';
1602         }
1603         return ($bib_result, $item_result, $bib_match);
1604     } else { # must be auths
1605         my ($auth_result, $auth_match);
1606
1607         if ($overlay_status ne 'no_match') {
1608             $auth_match = GetBestRecordMatch($import_record_id);
1609             if ($overlay_action eq 'replace') {
1610                 $auth_result  = defined($auth_match) ? 'replace' : 'create_new';
1611             } elsif ($overlay_action eq 'create_new') {
1612                 $auth_result  = 'create_new';
1613             } elsif ($overlay_action eq 'ignore') {
1614                 $auth_result  = 'ignore';
1615             }
1616         } else {
1617             $auth_result = $nomatch_action;
1618         }
1619
1620         return ($auth_result, undef, $auth_match);
1621
1622     }
1623 }
1624
1625 sub _get_revert_action {
1626     my ($overlay_action, $overlay_status, $status) = @_;
1627
1628     my $bib_result;
1629
1630     if ($status eq 'ignored') {
1631         $bib_result = 'ignore';
1632     } else {
1633         if ($overlay_action eq 'create_new') {
1634             $bib_result = 'delete';
1635         } else {
1636             $bib_result = ($overlay_status eq 'match_applied') ? 'restore' : 'delete';
1637         }
1638     }
1639     return $bib_result;
1640 }
1641
1642 1;
1643 __END__
1644
1645 =head1 AUTHOR
1646
1647 Koha Development Team <http://koha-community.org/>
1648
1649 Galen Charlton <galen.charlton@liblime.com>
1650
1651 =cut