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         require Exporter;
  37         @ISA    = qw(Exporter);
  38         @EXPORT = qw(
  39     GetZ3950BatchId
  40     GetWebserviceBatchId
  41     GetImportRecordMarc
  42     GetImportRecordMarcXML
  43     AddImportBatch
  44     GetImportBatch
  45     AddAuthToBatch
  46     AddBiblioToBatch
  47     AddItemsToImportBiblio
  48     ModAuthorityInBatch
  49     ModBiblioInBatch
  50
  51     BatchStageMarcRecords
  52     BatchFindDuplicates
  53     BatchCommitRecords
  54     BatchRevertRecords
  55     CleanBatch
  56     DeleteBatch
  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     $record_type,                $encoding,
 356     $marc_records,               $file_name,
 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 $marc_modification_template = shift;
 371     my $comments = shift;
 372     my $branch_code = shift;
 373     my $parse_items = shift;
 374     my $leave_as_staging = shift;
 375
 376     # optional callback to monitor status
 377     # of job
 378     my $progress_interval = 0;
 379     my $progress_callback = undef;
 380     if ($#_ == 1) {
 381         $progress_interval = shift;
 382         $progress_callback = shift;
 383         $progress_interval = 0 unless $progress_interval =~ /^\d+$/ and $progress_interval > 0;
 384         $progress_interval = 0 unless 'CODE' eq ref $progress_callback;
 385     }
 386
 387     my $batch_id = AddImportBatch( {
 388             overlay_action => 'create_new',
 389             import_status => 'staging',
 390             batch_type => 'batch',
 391             file_name => $file_name,
 392             comments => $comments,
 393             record_type => $record_type,
 394         } );
 395     if ($parse_items) {
 396         SetImportBatchItemAction($batch_id, 'always_add');
 397     } else {
 398         SetImportBatchItemAction($batch_id, 'ignore');
 399     }
 400
 401
 402     my $marc_type = C4::Context->preference('marcflavour');
 403     $marc_type .= 'AUTH' if ($marc_type eq 'UNIMARC' && $record_type eq 'auth');
 404     my @invalid_records = ();
 405     my $num_valid = 0;
 406     my $num_items = 0;
 407     # FIXME - for now, we're dealing only with bibs
 408     my $rec_num = 0;
 409     foreach my $marc_record (@$marc_records) {
 410         $rec_num++;
 411         if ($progress_interval and (0 == ($rec_num % $progress_interval))) {
 412             &$progress_callback($rec_num);
 413         }
 414
 415         ModifyRecordWithTemplate( $marc_modification_template, $marc_record ) if ( $marc_modification_template );
 416
 417         my $import_record_id;
 418         if (scalar($marc_record->fields()) == 0) {
 419             push @invalid_records, $marc_record;
 420         } else {
 421
 422             # Normalize the record so it doesn't have separated diacritics
 423             SetUTF8Flag($marc_record);
 424
 425             $num_valid++;
 426             if ($record_type eq 'biblio') {
 427                 $import_record_id = AddBiblioToBatch($batch_id, $rec_num, $marc_record, $encoding, int(rand(99999)), 0);
 428                 if ($parse_items) {
 429                     my @import_items_ids = AddItemsToImportBiblio($batch_id, $import_record_id, $marc_record, 0);
 430                     $num_items += scalar(@import_items_ids);
 431                 }
 432             } elsif ($record_type eq 'auth') {
 433                 $import_record_id = AddAuthToBatch($batch_id, $rec_num, $marc_record, $encoding, int(rand(99999)), 0, $marc_type);
 434             }
 435         }
 436     }
 437     unless ($leave_as_staging) {
 438         SetImportBatchStatus($batch_id, 'staged');
 439     }
 440     # FIXME branch_code, number of bibs, number of items
 441     _update_batch_record_counts($batch_id);
 442     return ($batch_id, $num_valid, $num_items, @invalid_records);
 443 }
 444
 445 =head2 AddItemsToImportBiblio
 446
 447   my @import_items_ids = AddItemsToImportBiblio($batch_id,
 448                 $import_record_id, $marc_record, $update_counts);
 449
 450 =cut
 451
 452 sub AddItemsToImportBiblio {
 453     my $batch_id = shift;
 454     my $import_record_id = shift;
 455     my $marc_record = shift;
 456     my $update_counts = @_ ? shift : 0;
 457
 458     my @import_items_ids = ();
 459
 460     my $dbh = C4::Context->dbh;
 461     my ($item_tag,$item_subfield) = &GetMarcFromKohaField("items.itemnumber",'');
 462     foreach my $item_field ($marc_record->field($item_tag)) {
 463         my $item_marc = MARC::Record->new();
 464         $item_marc->leader("00000    a              "); # must set Leader/09 to 'a'
 465         $item_marc->append_fields($item_field);
 466         $marc_record->delete_field($item_field);
 467         my $sth = $dbh->prepare_cached("INSERT INTO import_items (import_record_id, status, marcxml)
 468                                         VALUES (?, ?, ?)");
 469         $sth->bind_param(1, $import_record_id);
 470         $sth->bind_param(2, 'staged');
 471         $sth->bind_param(3, $item_marc->as_xml());
 472         $sth->execute();
 473         push @import_items_ids, $dbh->{'mysql_insertid'};
 474         $sth->finish();
 475     }
 476
 477     if ($#import_items_ids > -1) {
 478         _update_batch_record_counts($batch_id) if $update_counts;
 479         _update_import_record_marc($import_record_id, $marc_record, C4::Context->preference('marcflavour'));
 480     }
 481     return @import_items_ids;
 482 }
 483
 484 =head2 BatchFindDuplicates
 485
 486   my $num_with_matches = BatchFindDuplicates($batch_id, $matcher,
 487              $max_matches, $progress_interval, $progress_callback);
 488
 489 Goes through the records loaded in the batch and attempts to
 490 find duplicates for each one.  Sets the matching status
 491 of each record to "no_match" or "auto_match" as appropriate.
 492
 493 The $max_matches parameter is optional; if it is not supplied,
 494 it defaults to 10.
 495
 496 The $progress_interval and $progress_callback parameters are
 497 optional; if both are supplied, the sub referred to by
 498 $progress_callback will be invoked every $progress_interval
 499 records using the number of records processed as the
 500 singular argument.
 501
 502 =cut
 503
 504 sub BatchFindDuplicates {
 505     my $batch_id = shift;
 506     my $matcher = shift;
 507     my $max_matches = @_ ? shift : 10;
 508
 509     # optional callback to monitor status
 510     # of job
 511     my $progress_interval = 0;
 512     my $progress_callback = undef;
 513     if ($#_ == 1) {
 514         $progress_interval = shift;
 515         $progress_callback = shift;
 516         $progress_interval = 0 unless $progress_interval =~ /^\d+$/ and $progress_interval > 0;
 517         $progress_interval = 0 unless 'CODE' eq ref $progress_callback;
 518     }
 519
 520     my $dbh = C4::Context->dbh;
 521
 522     my $sth = $dbh->prepare("SELECT import_record_id, record_type, marc
 523                              FROM import_records
 524                              WHERE import_batch_id = ?");
 525     $sth->execute($batch_id);
 526     my $num_with_matches = 0;
 527     my $rec_num = 0;
 528     while (my $rowref = $sth->fetchrow_hashref) {
 529         $rec_num++;
 530         if ($progress_interval and (0 == ($rec_num % $progress_interval))) {
 531             &$progress_callback($rec_num);
 532         }
 533         my $marc_record = MARC::Record->new_from_usmarc($rowref->{'marc'});
 534         my @matches = ();
 535         if (defined $matcher) {
 536             @matches = $matcher->get_matches($marc_record, $max_matches);
 537         }
 538         if (scalar(@matches) > 0) {
 539             $num_with_matches++;
 540             SetImportRecordMatches($rowref->{'import_record_id'}, @matches);
 541             SetImportRecordOverlayStatus($rowref->{'import_record_id'}, 'auto_match');
 542         } else {
 543             SetImportRecordMatches($rowref->{'import_record_id'}, ());
 544             SetImportRecordOverlayStatus($rowref->{'import_record_id'}, 'no_match');
 545         }
 546     }
 547     $sth->finish();
 548     return $num_with_matches;
 549 }
 550
 551 =head2 BatchCommitRecords
 552
 553   my ($num_added, $num_updated, $num_items_added, $num_items_replaced, $num_items_errored, $num_ignored) =
 554         BatchCommitRecords($batch_id, $framework,
 555         $progress_interval, $progress_callback);
 556
 557 =cut
 558
 559 sub BatchCommitRecords {
 560     my $batch_id = shift;
 561     my $framework = shift;
 562
 563     # optional callback to monitor status
 564     # of job
 565     my $progress_interval = 0;
 566     my $progress_callback = undef;
 567     if ($#_ == 1) {
 568         $progress_interval = shift;
 569         $progress_callback = shift;
 570         $progress_interval = 0 unless $progress_interval =~ /^\d+$/ and $progress_interval > 0;
 571         $progress_interval = 0 unless 'CODE' eq ref $progress_callback;
 572     }
 573
 574     my $record_type;
 575     my $num_added = 0;
 576     my $num_updated = 0;
 577     my $num_items_added = 0;
 578     my $num_items_replaced = 0;
 579     my $num_items_errored = 0;
 580     my $num_ignored = 0;
 581     # commit (i.e., save, all records in the batch)
 582     SetImportBatchStatus('importing');
 583     my $overlay_action = GetImportBatchOverlayAction($batch_id);
 584     my $nomatch_action = GetImportBatchNoMatchAction($batch_id);
 585     my $item_action = GetImportBatchItemAction($batch_id);
 586     my $item_tag;
 587     my $item_subfield;
 588     my $dbh = C4::Context->dbh;
 589     my $sth = $dbh->prepare("SELECT import_records.import_record_id, record_type, status, overlay_status, marc, encoding
 590                              FROM import_records
 591                              LEFT JOIN import_auths ON (import_records.import_record_id=import_auths.import_record_id)
 592                              LEFT JOIN import_biblios ON (import_records.import_record_id=import_biblios.import_record_id)
 593                              WHERE import_batch_id = ?");
 594     $sth->execute($batch_id);
 595     my $marcflavour = C4::Context->preference('marcflavour');
 596     my $rec_num = 0;
 597     while (my $rowref = $sth->fetchrow_hashref) {
 598         $record_type = $rowref->{'record_type'};
 599         $rec_num++;
 600         if ($progress_interval and (0 == ($rec_num % $progress_interval))) {
 601             &$progress_callback($rec_num);
 602         }
 603         if ($rowref->{'status'} eq 'error' or $rowref->{'status'} eq 'imported') {
 604             $num_ignored++;
 605             next;
 606         }
 607
 608         my $marc_type;
 609         if ($marcflavour eq 'UNIMARC' && $record_type eq 'auth') {
 610             $marc_type = 'UNIMARCAUTH';
 611         } elsif ($marcflavour eq 'UNIMARC') {
 612             $marc_type = 'UNIMARC';
 613         } else {
 614             $marc_type = 'USMARC';
 615         }
 616         my $marc_record = MARC::Record->new_from_usmarc($rowref->{'marc'});
 617
 618         if ($record_type eq 'biblio') {
 619             # remove any item tags - rely on BatchCommitItems
 620             ($item_tag,$item_subfield) = &GetMarcFromKohaField("items.itemnumber",'');
 621             foreach my $item_field ($marc_record->field($item_tag)) {
 622                 $marc_record->delete_field($item_field);
 623             }
 624         }
 625
 626         my ($record_result, $item_result, $record_match) =
 627             _get_commit_action($overlay_action, $nomatch_action, $item_action,
 628                                $rowref->{'overlay_status'}, $rowref->{'import_record_id'}, $record_type);
 629
 630         my $recordid;
 631         my $query;
 632         if ($record_result eq 'create_new') {
 633             $num_added++;
 634             if ($record_type eq 'biblio') {
 635                 my $biblioitemnumber;
 636                 ($recordid, $biblioitemnumber) = AddBiblio($marc_record, $framework);
 637                 $query = "UPDATE import_biblios SET matched_biblionumber = ? WHERE import_record_id = ?";
 638                 if ($item_result eq 'create_new' || $item_result eq 'replace') {
 639                     my ($bib_items_added, $bib_items_replaced, $bib_items_errored) = BatchCommitItems($rowref->{'import_record_id'}, $recordid, $item_result);
 640                     $num_items_added += $bib_items_added;
 641                     $num_items_replaced += $bib_items_replaced;
 642                     $num_items_errored += $bib_items_errored;
 643                 }
 644             } else {
 645                 $recordid = AddAuthority($marc_record, undef, GuessAuthTypeCode($marc_record));
 646                 $query = "UPDATE import_auths SET matched_authid = ? WHERE import_record_id = ?";
 647             }
 648             my $sth = $dbh->prepare_cached($query);
 649             $sth->execute($recordid, $rowref->{'import_record_id'});
 650             $sth->finish();
 651             SetImportRecordStatus($rowref->{'import_record_id'}, 'imported');
 652         } elsif ($record_result eq 'replace') {
 653             $num_updated++;
 654             $recordid = $record_match;
 655             my $oldxml;
 656             if ($record_type eq 'biblio') {
 657                 my $oldbiblio = Koha::Biblios->find( $recordid );
 658                 $oldxml = GetXmlBiblio($recordid);
 659
 660                 # remove item fields so that they don't get
 661                 # added again if record is reverted
 662                 # 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.
 663                 my $old_marc = MARC::Record->new_from_xml(StripNonXmlChars($oldxml), 'UTF-8', $rowref->{'encoding'}, $marc_type);
 664                 foreach my $item_field ($old_marc->field($item_tag)) {
 665                     $old_marc->delete_field($item_field);
 666                 }
 667                 $oldxml = $old_marc->as_xml($marc_type);
 668
 669                 ModBiblio($marc_record, $recordid, $oldbiblio->frameworkcode);
 670                 $query = "UPDATE import_biblios SET matched_biblionumber = ? WHERE import_record_id = ?";
 671
 672                 if ($item_result eq 'create_new' || $item_result eq 'replace') {
 673                     my ($bib_items_added, $bib_items_replaced, $bib_items_errored) = BatchCommitItems($rowref->{'import_record_id'}, $recordid, $item_result);
 674                     $num_items_added += $bib_items_added;
 675                     $num_items_replaced += $bib_items_replaced;
 676                     $num_items_errored += $bib_items_errored;
 677                 }
 678             } else {
 679                 $oldxml = GetAuthorityXML($recordid);
 680
 681                 ModAuthority($recordid, $marc_record, GuessAuthTypeCode($marc_record));
 682                 $query = "UPDATE import_auths SET matched_authid = ? WHERE import_record_id = ?";
 683             }
 684             my $sth = $dbh->prepare_cached("UPDATE import_records SET marcxml_old = ? WHERE import_record_id = ?");
 685             $sth->execute($oldxml, $rowref->{'import_record_id'});
 686             $sth->finish();
 687             my $sth2 = $dbh->prepare_cached($query);
 688             $sth2->execute($recordid, $rowref->{'import_record_id'});
 689             $sth2->finish();
 690             SetImportRecordOverlayStatus($rowref->{'import_record_id'}, 'match_applied');
 691             SetImportRecordStatus($rowref->{'import_record_id'}, 'imported');
 692         } elsif ($record_result eq 'ignore') {
 693             $recordid = $record_match;
 694             $num_ignored++;
 695             $recordid = $record_match;
 696             if ($record_type eq 'biblio' and defined $recordid and ( $item_result eq 'create_new' || $item_result eq 'replace' ) ) {
 697                 my ($bib_items_added, $bib_items_replaced, $bib_items_errored) = BatchCommitItems($rowref->{'import_record_id'}, $recordid, $item_result);
 698                 $num_items_added += $bib_items_added;
 699          $num_items_replaced += $bib_items_replaced;
 700                 $num_items_errored += $bib_items_errored;
 701                 # still need to record the matched biblionumber so that the
 702                 # items can be reverted
 703                 my $sth2 = $dbh->prepare_cached("UPDATE import_biblios SET matched_biblionumber = ? WHERE import_record_id = ?");
 704                 $sth2->execute($recordid, $rowref->{'import_record_id'});
 705                 SetImportRecordOverlayStatus($rowref->{'import_record_id'}, 'match_applied');
 706             }
 707             SetImportRecordStatus($rowref->{'import_record_id'}, 'ignored');
 708         }
 709     }
 710     $sth->finish();
 711     SetImportBatchStatus($batch_id, 'imported');
 712     return ($num_added, $num_updated, $num_items_added, $num_items_replaced, $num_items_errored, $num_ignored);
 713 }
 714
 715 =head2 BatchCommitItems
 716
 717   ($num_items_added, $num_items_errored) =
 718          BatchCommitItems($import_record_id, $biblionumber);
 719
 720 =cut
 721
 722 sub BatchCommitItems {
 723     my ( $import_record_id, $biblionumber, $action ) = @_;
 724
 725     my $dbh = C4::Context->dbh;
 726
 727     my $num_items_added = 0;
 728     my $num_items_errored = 0;
 729     my $num_items_replaced = 0;
 730
 731     my $sth = $dbh->prepare( "
 732         SELECT import_items_id, import_items.marcxml, encoding
 733         FROM import_items
 734         JOIN import_records USING (import_record_id)
 735         WHERE import_record_id = ?
 736         ORDER BY import_items_id
 737     " );
 738     $sth->bind_param( 1, $import_record_id );
 739     $sth->execute();
 740
 741     while ( my $row = $sth->fetchrow_hashref() ) {
 742         my $item_marc = MARC::Record->new_from_xml( StripNonXmlChars( $row->{'marcxml'} ), 'UTF-8', $row->{'encoding'} );
 743
 744         # Delete date_due subfield as to not accidentally delete item checkout due dates
 745         my ( $MARCfield, $MARCsubfield ) = GetMarcFromKohaField( 'items.onloan', GetFrameworkCode($biblionumber) );
 746         $item_marc->field($MARCfield)->delete_subfield( code => $MARCsubfield );
 747
 748         my $item = TransformMarcToKoha( $item_marc );
 749
 750         my $duplicate_barcode = exists( $item->{'barcode'} ) && GetItemnumberFromBarcode( $item->{'barcode'} );
 751         my $duplicate_itemnumber = exists( $item->{'itemnumber'} );
 752
 753         my $updsth = $dbh->prepare("UPDATE import_items SET status = ?, itemnumber = ? WHERE import_items_id = ?");
 754         if ( $action eq "replace" && $duplicate_itemnumber ) {
 755             # Duplicate itemnumbers have precedence, that way we can update barcodes by overlaying
 756             ModItemFromMarc( $item_marc, $biblionumber, $item->{itemnumber} );
 757             $updsth->bind_param( 1, 'imported' );
 758             $updsth->bind_param( 2, $item->{itemnumber} );
 759             $updsth->bind_param( 3, $row->{'import_items_id'} );
 760             $updsth->execute();
 761             $updsth->finish();
 762             $num_items_replaced++;
 763         } elsif ( $action eq "replace" && $duplicate_barcode ) {
 764             my $itemnumber = GetItemnumberFromBarcode( $item->{'barcode'} );
 765             ModItemFromMarc( $item_marc, $biblionumber, $itemnumber );
 766             $updsth->bind_param( 1, 'imported' );
 767             $updsth->bind_param( 2, $item->{itemnumber} );
 768             $updsth->bind_param( 3, $row->{'import_items_id'} );
 769             $updsth->execute();
 770             $updsth->finish();
 771             $num_items_replaced++;
 772         } elsif ($duplicate_barcode) {
 773             $updsth->bind_param( 1, 'error' );
 774             $updsth->bind_param( 2, 'duplicate item barcode' );
 775             $updsth->bind_param( 3, $row->{'import_items_id'} );
 776             $updsth->execute();
 777             $num_items_errored++;
 778         } else {
 779             my ( $item_biblionumber, $biblioitemnumber, $itemnumber ) = AddItemFromMarc( $item_marc, $biblionumber );
 780             if( $itemnumber ) {
 781                 $updsth->bind_param( 1, 'imported' );
 782                 $updsth->bind_param( 2, $itemnumber );
 783                 $updsth->bind_param( 3, $row->{'import_items_id'} );
 784                 $updsth->execute();
 785                 $updsth->finish();
 786                 $num_items_added++;
 787             }
 788         }
 789     }
 790
 791     return ( $num_items_added, $num_items_replaced, $num_items_errored );
 792 }
 793
 794 =head2 BatchRevertRecords
 795
 796   my ($num_deleted, $num_errors, $num_reverted, $num_items_deleted,
 797       $num_ignored) = BatchRevertRecords($batch_id);
 798
 799 =cut
 800
 801 sub BatchRevertRecords {
 802     my $batch_id = shift;
 803
 804     $logger->trace("C4::ImportBatch::BatchRevertRecords( $batch_id )");
 805
 806     my $record_type;
 807     my $num_deleted = 0;
 808     my $num_errors = 0;
 809     my $num_reverted = 0;
 810     my $num_ignored = 0;
 811     my $num_items_deleted = 0;
 812     # commit (i.e., save, all records in the batch)
 813     SetImportBatchStatus('reverting');
 814     my $overlay_action = GetImportBatchOverlayAction($batch_id);
 815     my $nomatch_action = GetImportBatchNoMatchAction($batch_id);
 816     my $dbh = C4::Context->dbh;
 817     my $sth = $dbh->prepare("SELECT import_records.import_record_id, record_type, status, overlay_status, marcxml_old, encoding, matched_biblionumber, matched_authid
 818                              FROM import_records
 819                              LEFT JOIN import_auths ON (import_records.import_record_id=import_auths.import_record_id)
 820                              LEFT JOIN import_biblios ON (import_records.import_record_id=import_biblios.import_record_id)
 821                              WHERE import_batch_id = ?");
 822     $sth->execute($batch_id);
 823     my $marc_type;
 824     my $marcflavour = C4::Context->preference('marcflavour');
 825     while (my $rowref = $sth->fetchrow_hashref) {
 826         $record_type = $rowref->{'record_type'};
 827         if ($rowref->{'status'} eq 'error' or $rowref->{'status'} eq 'reverted') {
 828             $num_ignored++;
 829             next;
 830         }
 831         if ($marcflavour eq 'UNIMARC' && $record_type eq 'auth') {
 832             $marc_type = 'UNIMARCAUTH';
 833         } elsif ($marcflavour eq 'UNIMARC') {
 834             $marc_type = 'UNIMARC';
 835         } else {
 836             $marc_type = 'USMARC';
 837         }
 838
 839         my $record_result = _get_revert_action($overlay_action, $rowref->{'overlay_status'}, $rowref->{'status'});
 840
 841         if ($record_result eq 'delete') {
 842             my $error = undef;
 843             if  ($record_type eq 'biblio') {
 844                 $num_items_deleted += BatchRevertItems($rowref->{'import_record_id'}, $rowref->{'matched_biblionumber'});
 845                 $error = DelBiblio($rowref->{'matched_biblionumber'});
 846             } else {
 847                 DelAuthority({ authid => $rowref->{'matched_authid'} });
 848             }
 849             if (defined $error) {
 850                 $num_errors++;
 851             } else {
 852                 $num_deleted++;
 853                 SetImportRecordStatus($rowref->{'import_record_id'}, 'reverted');
 854             }
 855         } elsif ($record_result eq 'restore') {
 856             $num_reverted++;
 857             my $old_record = MARC::Record->new_from_xml(StripNonXmlChars($rowref->{'marcxml_old'}), 'UTF-8', $rowref->{'encoding'}, $marc_type);
 858             if ($record_type eq 'biblio') {
 859                 my $biblionumber = $rowref->{'matched_biblionumber'};
 860                 my $oldbiblio = Koha::Biblios->find( $biblionumber );
 861
 862                 $logger->info("C4::ImportBatch::BatchRevertRecords: Biblio record $biblionumber does not exist, restoration of this record was skipped") unless $oldbiblio;
 863                 next unless $oldbiblio; # Record has since been deleted. Deleted records should stay deleted.
 864
 865                 $num_items_deleted += BatchRevertItems($rowref->{'import_record_id'}, $rowref->{'matched_biblionumber'});
 866                 ModBiblio($old_record, $biblionumber, $oldbiblio->frameworkcode);
 867             } else {
 868                 my $authid = $rowref->{'matched_authid'};
 869                 ModAuthority($authid, $old_record, GuessAuthTypeCode($old_record));
 870             }
 871             SetImportRecordStatus($rowref->{'import_record_id'}, 'reverted');
 872         } elsif ($record_result eq 'ignore') {
 873             if ($record_type eq 'biblio') {
 874                 $num_items_deleted += BatchRevertItems($rowref->{'import_record_id'}, $rowref->{'matched_biblionumber'});
 875             }
 876             SetImportRecordStatus($rowref->{'import_record_id'}, 'reverted');
 877         }
 878         my $query;
 879         if ($record_type eq 'biblio') {
 880             # remove matched_biblionumber only if there is no 'imported' item left
 881             $query = "UPDATE import_biblios SET matched_biblionumber = NULL WHERE import_record_id = ?";
 882             $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')";
 883         } else {
 884             $query = "UPDATE import_auths SET matched_authid = NULL WHERE import_record_id = ?";
 885         }
 886         my $sth2 = $dbh->prepare_cached($query);
 887         $sth2->execute($rowref->{'import_record_id'});
 888     }
 889
 890     $sth->finish();
 891     SetImportBatchStatus($batch_id, 'reverted');
 892     return ($num_deleted, $num_errors, $num_reverted, $num_items_deleted, $num_ignored);
 893 }
 894
 895 =head2 BatchRevertItems
 896
 897   my $num_items_deleted = BatchRevertItems($import_record_id, $biblionumber);
 898
 899 =cut
 900
 901 sub BatchRevertItems {
 902     my ($import_record_id, $biblionumber) = @_;
 903
 904     my $dbh = C4::Context->dbh;
 905     my $num_items_deleted = 0;
 906
 907     my $sth = $dbh->prepare_cached("SELECT import_items_id, itemnumber
 908                                    FROM import_items
 909                                    JOIN items USING (itemnumber)
 910                                    WHERE import_record_id = ?");
 911     $sth->bind_param(1, $import_record_id);
 912     $sth->execute();
 913     while (my $row = $sth->fetchrow_hashref()) {
 914         my $error = DelItemCheck( $biblionumber, $row->{'itemnumber'});
 915         if ($error == 1){
 916             my $updsth = $dbh->prepare("UPDATE import_items SET status = ? WHERE import_items_id = ?");
 917             $updsth->bind_param(1, 'reverted');
 918             $updsth->bind_param(2, $row->{'import_items_id'});
 919             $updsth->execute();
 920             $updsth->finish();
 921             $num_items_deleted++;
 922         }
 923         else {
 924             next;
 925         }
 926     }
 927     $sth->finish();
 928     return $num_items_deleted;
 929 }
 930
 931 =head2 CleanBatch
 932
 933   CleanBatch($batch_id)
 934
 935 Deletes all staged records from the import batch
 936 and sets the status of the batch to 'cleaned'.  Note
 937 that deleting a stage record does *not* affect
 938 any record that has been committed to the database.
 939
 940 =cut
 941
 942 sub CleanBatch {
 943     my $batch_id = shift;
 944     return unless defined $batch_id;
 945
 946     C4::Context->dbh->do('DELETE FROM import_records WHERE import_batch_id = ?', {}, $batch_id);
 947     SetImportBatchStatus($batch_id, 'cleaned');
 948 }
 949
 950 =head2 DeleteBatch
 951
 952   DeleteBatch($batch_id)
 953
 954 Deletes the record from the database. This can only be done
 955 once the batch has been cleaned.
 956
 957 =cut
 958
 959 sub DeleteBatch {
 960     my $batch_id = shift;
 961     return unless defined $batch_id;
 962
 963     my $dbh = C4::Context->dbh;
 964     my $sth = $dbh->prepare('DELETE FROM import_batches WHERE import_batch_id = ?');
 965     $sth->execute( $batch_id );
 966 }
 967
 968 =head2 GetAllImportBatches
 969
 970   my $results = GetAllImportBatches();
 971
 972 Returns a references to an array of hash references corresponding
 973 to all import_batches rows (of batch_type 'batch'), sorted in
 974 ascending order by import_batch_id.
 975
 976 =cut
 977
 978 sub  GetAllImportBatches {
 979     my $dbh = C4::Context->dbh;
 980     my $sth = $dbh->prepare_cached("SELECT * FROM import_batches
 981                                     WHERE batch_type IN ('batch', 'webservice')
 982                                     ORDER BY import_batch_id ASC");
 983
 984     my $results = [];
 985     $sth->execute();
 986     while (my $row = $sth->fetchrow_hashref) {
 987         push @$results, $row;
 988     }
 989     $sth->finish();
 990     return $results;
 991 }
 992
 993 =head2 GetStagedWebserviceBatches
 994
 995   my $batch_ids = GetStagedWebserviceBatches();
 996
 997 Returns a references to an array of batch id's
 998 of batch_type 'webservice' that are not imported
 999
1000 =cut
1001
1002 my $PENDING_WEBSERVICE_BATCHES_QRY = <<EOQ;
1003 SELECT import_batch_id FROM import_batches
1004 WHERE batch_type = 'webservice'
1005 AND import_status = 'staged'
1006 EOQ
1007 sub  GetStagedWebserviceBatches {
1008     my $dbh = C4::Context->dbh;
1009     return $dbh->selectcol_arrayref($PENDING_WEBSERVICE_BATCHES_QRY);
1010 }
1011
1012 =head2 GetImportBatchRangeDesc
1013
1014   my $results = GetImportBatchRangeDesc($offset, $results_per_group);
1015
1016 Returns a reference to an array of hash references corresponding to
1017 import_batches rows (sorted in descending order by import_batch_id)
1018 start at the given offset.
1019
1020 =cut
1021
1022 sub GetImportBatchRangeDesc {
1023     my ($offset, $results_per_group) = @_;
1024
1025     my $dbh = C4::Context->dbh;
1026     my $query = "SELECT * FROM import_batches
1027                                     WHERE batch_type IN ('batch', 'webservice')
1028                                     ORDER BY import_batch_id DESC";
1029     my @params;
1030     if ($results_per_group){
1031         $query .= " LIMIT ?";
1032         push(@params, $results_per_group);
1033     }
1034     if ($offset){
1035         $query .= " OFFSET ?";
1036         push(@params, $offset);
1037     }
1038     my $sth = $dbh->prepare_cached($query);
1039     $sth->execute(@params);
1040     my $results = $sth->fetchall_arrayref({});
1041     $sth->finish();
1042     return $results;
1043 }
1044
1045 =head2 GetItemNumbersFromImportBatch
1046
1047   my @itemsnos = GetItemNumbersFromImportBatch($batch_id);
1048
1049 =cut
1050
1051 sub GetItemNumbersFromImportBatch {
1052     my ($batch_id) = @_;
1053     my $dbh = C4::Context->dbh;
1054     my $sql = q|
1055 SELECT itemnumber FROM import_items
1056 INNER JOIN items USING (itemnumber)
1057 INNER JOIN import_records USING (import_record_id)
1058 WHERE import_batch_id = ?|;
1059     my  $sth = $dbh->prepare( $sql );
1060     $sth->execute($batch_id);
1061     my @items ;
1062     while ( my ($itm) = $sth->fetchrow_array ) {
1063         push @items, $itm;
1064     }
1065     return @items;
1066 }
1067
1068 =head2 GetNumberOfImportBatches
1069
1070   my $count = GetNumberOfImportBatches();
1071
1072 =cut
1073
1074 sub GetNumberOfNonZ3950ImportBatches {
1075     my $dbh = C4::Context->dbh;
1076     my $sth = $dbh->prepare("SELECT COUNT(*) FROM import_batches WHERE batch_type != 'z3950'");
1077     $sth->execute();
1078     my ($count) = $sth->fetchrow_array();
1079     $sth->finish();
1080     return $count;
1081 }
1082
1083 =head2 GetImportBiblios
1084
1085   my $results = GetImportBiblios($importid);
1086
1087 =cut
1088
1089 sub GetImportBiblios {
1090     my ($import_record_id) = @_;
1091
1092     my $dbh = C4::Context->dbh;
1093     my $query = "SELECT * FROM import_biblios WHERE import_record_id = ?";
1094     return $dbh->selectall_arrayref(
1095         $query,
1096         { Slice => {} },
1097         $import_record_id
1098     );
1099
1100 }
1101
1102 =head2 GetImportRecordsRange
1103
1104   my $results = GetImportRecordsRange($batch_id, $offset, $results_per_group);
1105
1106 Returns a reference to an array of hash references corresponding to
1107 import_biblios/import_auths/import_records rows for a given batch
1108 starting at the given offset.
1109
1110 =cut
1111
1112 sub GetImportRecordsRange {
1113     my ( $batch_id, $offset, $results_per_group, $status, $parameters ) = @_;
1114
1115     my $dbh = C4::Context->dbh;
1116
1117     my $order_by = $parameters->{order_by} || 'import_record_id';
1118     ( $order_by ) = grep( /^$order_by$/, qw( import_record_id title status overlay_status ) ) ? $order_by : 'import_record_id';
1119
1120     my $order_by_direction =
1121       uc( $parameters->{order_by_direction} // 'ASC' ) eq 'DESC' ? 'DESC' : 'ASC';
1122
1123     $order_by .= " $order_by_direction, authorized_heading" if $order_by eq 'title';
1124
1125     my $query = "SELECT title, author, isbn, issn, authorized_heading, import_records.import_record_id,
1126                                            record_sequence, status, overlay_status,
1127                                            matched_biblionumber, matched_authid, record_type
1128                                     FROM   import_records
1129                                     LEFT JOIN import_auths ON (import_records.import_record_id=import_auths.import_record_id)
1130                                     LEFT JOIN import_biblios ON (import_records.import_record_id=import_biblios.import_record_id)
1131                                     WHERE  import_batch_id = ?";
1132     my @params;
1133     push(@params, $batch_id);
1134     if ($status) {
1135         $query .= " AND status=?";
1136         push(@params,$status);
1137     }
1138
1139     $query.=" ORDER BY $order_by $order_by_direction";
1140
1141     if($results_per_group){
1142         $query .= " LIMIT ?";
1143         push(@params, $results_per_group);
1144     }
1145     if($offset){
1146         $query .= " OFFSET ?";
1147         push(@params, $offset);
1148     }
1149     my $sth = $dbh->prepare_cached($query);
1150     $sth->execute(@params);
1151     my $results = $sth->fetchall_arrayref({});
1152     $sth->finish();
1153     return $results;
1154
1155 }
1156
1157 =head2 GetBestRecordMatch
1158
1159   my $record_id = GetBestRecordMatch($import_record_id);
1160
1161 =cut
1162
1163 sub GetBestRecordMatch {
1164     my ($import_record_id) = @_;
1165
1166     my $dbh = C4::Context->dbh;
1167     my $sth = $dbh->prepare("SELECT candidate_match_id
1168                              FROM   import_record_matches
1169                              JOIN   import_records ON ( import_record_matches.import_record_id = import_records.import_record_id )
1170                              LEFT JOIN biblio ON ( candidate_match_id = biblio.biblionumber )
1171                              LEFT JOIN auth_header ON ( candidate_match_id = auth_header.authid )
1172                              WHERE  import_record_matches.import_record_id = ? AND
1173                              (  (import_records.record_type = 'biblio' AND biblio.biblionumber IS NOT NULL) OR
1174                                 (import_records.record_type = 'auth' AND auth_header.authid IS NOT NULL) )
1175                              ORDER BY score DESC, candidate_match_id DESC");
1176     $sth->execute($import_record_id);
1177     my ($record_id) = $sth->fetchrow_array();
1178     $sth->finish();
1179     return $record_id;
1180 }
1181
1182 =head2 GetImportBatchStatus
1183
1184   my $status = GetImportBatchStatus($batch_id);
1185
1186 =cut
1187
1188 sub GetImportBatchStatus {
1189     my ($batch_id) = @_;
1190
1191     my $dbh = C4::Context->dbh;
1192     my $sth = $dbh->prepare("SELECT import_status FROM import_batches WHERE import_batch_id = ?");
1193     $sth->execute($batch_id);
1194     my ($status) = $sth->fetchrow_array();
1195     $sth->finish();
1196     return $status;
1197
1198 }
1199
1200 =head2 SetImportBatchStatus
1201
1202   SetImportBatchStatus($batch_id, $new_status);
1203
1204 =cut
1205
1206 sub SetImportBatchStatus {
1207     my ($batch_id, $new_status) = @_;
1208
1209     my $dbh = C4::Context->dbh;
1210     my $sth = $dbh->prepare("UPDATE import_batches SET import_status = ? WHERE import_batch_id = ?");
1211     $sth->execute($new_status, $batch_id);
1212     $sth->finish();
1213
1214 }
1215
1216 =head2 GetImportBatchOverlayAction
1217
1218   my $overlay_action = GetImportBatchOverlayAction($batch_id);
1219
1220 =cut
1221
1222 sub GetImportBatchOverlayAction {
1223     my ($batch_id) = @_;
1224
1225     my $dbh = C4::Context->dbh;
1226     my $sth = $dbh->prepare("SELECT overlay_action FROM import_batches WHERE import_batch_id = ?");
1227     $sth->execute($batch_id);
1228     my ($overlay_action) = $sth->fetchrow_array();
1229     $sth->finish();
1230     return $overlay_action;
1231
1232 }
1233
1234
1235 =head2 SetImportBatchOverlayAction
1236
1237   SetImportBatchOverlayAction($batch_id, $new_overlay_action);
1238
1239 =cut
1240
1241 sub SetImportBatchOverlayAction {
1242     my ($batch_id, $new_overlay_action) = @_;
1243
1244     my $dbh = C4::Context->dbh;
1245     my $sth = $dbh->prepare("UPDATE import_batches SET overlay_action = ? WHERE import_batch_id = ?");
1246     $sth->execute($new_overlay_action, $batch_id);
1247     $sth->finish();
1248
1249 }
1250
1251 =head2 GetImportBatchNoMatchAction
1252
1253   my $nomatch_action = GetImportBatchNoMatchAction($batch_id);
1254
1255 =cut
1256
1257 sub GetImportBatchNoMatchAction {
1258     my ($batch_id) = @_;
1259
1260     my $dbh = C4::Context->dbh;
1261     my $sth = $dbh->prepare("SELECT nomatch_action FROM import_batches WHERE import_batch_id = ?");
1262     $sth->execute($batch_id);
1263     my ($nomatch_action) = $sth->fetchrow_array();
1264     $sth->finish();
1265     return $nomatch_action;
1266
1267 }
1268
1269
1270 =head2 SetImportBatchNoMatchAction
1271
1272   SetImportBatchNoMatchAction($batch_id, $new_nomatch_action);
1273
1274 =cut
1275
1276 sub SetImportBatchNoMatchAction {
1277     my ($batch_id, $new_nomatch_action) = @_;
1278
1279     my $dbh = C4::Context->dbh;
1280     my $sth = $dbh->prepare("UPDATE import_batches SET nomatch_action = ? WHERE import_batch_id = ?");
1281     $sth->execute($new_nomatch_action, $batch_id);
1282     $sth->finish();
1283
1284 }
1285
1286 =head2 GetImportBatchItemAction
1287
1288   my $item_action = GetImportBatchItemAction($batch_id);
1289
1290 =cut
1291
1292 sub GetImportBatchItemAction {
1293     my ($batch_id) = @_;
1294
1295     my $dbh = C4::Context->dbh;
1296     my $sth = $dbh->prepare("SELECT item_action FROM import_batches WHERE import_batch_id = ?");
1297     $sth->execute($batch_id);
1298     my ($item_action) = $sth->fetchrow_array();
1299     $sth->finish();
1300     return $item_action;
1301
1302 }
1303
1304
1305 =head2 SetImportBatchItemAction
1306
1307   SetImportBatchItemAction($batch_id, $new_item_action);
1308
1309 =cut
1310
1311 sub SetImportBatchItemAction {
1312     my ($batch_id, $new_item_action) = @_;
1313
1314     my $dbh = C4::Context->dbh;
1315     my $sth = $dbh->prepare("UPDATE import_batches SET item_action = ? WHERE import_batch_id = ?");
1316     $sth->execute($new_item_action, $batch_id);
1317     $sth->finish();
1318
1319 }
1320
1321 =head2 GetImportBatchMatcher
1322
1323   my $matcher_id = GetImportBatchMatcher($batch_id);
1324
1325 =cut
1326
1327 sub GetImportBatchMatcher {
1328     my ($batch_id) = @_;
1329
1330     my $dbh = C4::Context->dbh;
1331     my $sth = $dbh->prepare("SELECT matcher_id FROM import_batches WHERE import_batch_id = ?");
1332     $sth->execute($batch_id);
1333     my ($matcher_id) = $sth->fetchrow_array();
1334     $sth->finish();
1335     return $matcher_id;
1336
1337 }
1338
1339
1340 =head2 SetImportBatchMatcher
1341
1342   SetImportBatchMatcher($batch_id, $new_matcher_id);
1343
1344 =cut
1345
1346 sub SetImportBatchMatcher {
1347     my ($batch_id, $new_matcher_id) = @_;
1348
1349     my $dbh = C4::Context->dbh;
1350     my $sth = $dbh->prepare("UPDATE import_batches SET matcher_id = ? WHERE import_batch_id = ?");
1351     $sth->execute($new_matcher_id, $batch_id);
1352     $sth->finish();
1353
1354 }
1355
1356 =head2 GetImportRecordOverlayStatus
1357
1358   my $overlay_status = GetImportRecordOverlayStatus($import_record_id);
1359
1360 =cut
1361
1362 sub GetImportRecordOverlayStatus {
1363     my ($import_record_id) = @_;
1364
1365     my $dbh = C4::Context->dbh;
1366     my $sth = $dbh->prepare("SELECT overlay_status FROM import_records WHERE import_record_id = ?");
1367     $sth->execute($import_record_id);
1368     my ($overlay_status) = $sth->fetchrow_array();
1369     $sth->finish();
1370     return $overlay_status;
1371
1372 }
1373
1374
1375 =head2 SetImportRecordOverlayStatus
1376
1377   SetImportRecordOverlayStatus($import_record_id, $new_overlay_status);
1378
1379 =cut
1380
1381 sub SetImportRecordOverlayStatus {
1382     my ($import_record_id, $new_overlay_status) = @_;
1383
1384     my $dbh = C4::Context->dbh;
1385     my $sth = $dbh->prepare("UPDATE import_records SET overlay_status = ? WHERE import_record_id = ?");
1386     $sth->execute($new_overlay_status, $import_record_id);
1387     $sth->finish();
1388
1389 }
1390
1391 =head2 GetImportRecordStatus
1392
1393   my $status = GetImportRecordStatus($import_record_id);
1394
1395 =cut
1396
1397 sub GetImportRecordStatus {
1398     my ($import_record_id) = @_;
1399
1400     my $dbh = C4::Context->dbh;
1401     my $sth = $dbh->prepare("SELECT status FROM import_records WHERE import_record_id = ?");
1402     $sth->execute($import_record_id);
1403     my ($status) = $sth->fetchrow_array();
1404     $sth->finish();
1405     return $status;
1406
1407 }
1408
1409
1410 =head2 SetImportRecordStatus
1411
1412   SetImportRecordStatus($import_record_id, $new_status);
1413
1414 =cut
1415
1416 sub SetImportRecordStatus {
1417     my ($import_record_id, $new_status) = @_;
1418
1419     my $dbh = C4::Context->dbh;
1420     my $sth = $dbh->prepare("UPDATE import_records SET status = ? WHERE import_record_id = ?");
1421     $sth->execute($new_status, $import_record_id);
1422     $sth->finish();
1423
1424 }
1425
1426 =head2 GetImportRecordMatches
1427
1428   my $results = GetImportRecordMatches($import_record_id, $best_only);
1429
1430 =cut
1431
1432 sub GetImportRecordMatches {
1433     my $import_record_id = shift;
1434     my $best_only = @_ ? shift : 0;
1435
1436     my $dbh = C4::Context->dbh;
1437     # FIXME currently biblio only
1438     my $sth = $dbh->prepare_cached("SELECT title, author, biblionumber,
1439                                     candidate_match_id, score, record_type
1440                                     FROM import_records
1441                                     JOIN import_record_matches USING (import_record_id)
1442                                     LEFT JOIN biblio ON (biblionumber = candidate_match_id)
1443                                     WHERE import_record_id = ?
1444                                     ORDER BY score DESC, biblionumber DESC");
1445     $sth->bind_param(1, $import_record_id);
1446     my $results = [];
1447     $sth->execute();
1448     while (my $row = $sth->fetchrow_hashref) {
1449         if ($row->{'record_type'} eq 'auth') {
1450             $row->{'authorized_heading'} = C4::AuthoritiesMarc::GetAuthorizedHeading( { authid => $row->{'candidate_match_id'} } );
1451         }
1452         next if ($row->{'record_type'} eq 'biblio' && not $row->{'biblionumber'});
1453         push @$results, $row;
1454         last if $best_only;
1455     }
1456     $sth->finish();
1457
1458     return $results;
1459
1460 }
1461
1462 =head2 SetImportRecordMatches
1463
1464   SetImportRecordMatches($import_record_id, @matches);
1465
1466 =cut
1467
1468 sub SetImportRecordMatches {
1469     my $import_record_id = shift;
1470     my @matches = @_;
1471
1472     my $dbh = C4::Context->dbh;
1473     my $delsth = $dbh->prepare("DELETE FROM import_record_matches WHERE import_record_id = ?");
1474     $delsth->execute($import_record_id);
1475     $delsth->finish();
1476
1477     my $sth = $dbh->prepare("INSERT INTO import_record_matches (import_record_id, candidate_match_id, score)
1478                                     VALUES (?, ?, ?)");
1479     foreach my $match (@matches) {
1480         $sth->execute($import_record_id, $match->{'record_id'}, $match->{'score'});
1481     }
1482 }
1483
1484 =head2 RecordsFromISO2709File
1485
1486     my ($errors, $records) = C4::ImportBatch::RecordsFromISO2709File($input_file, $record_type, $encoding);
1487
1488 Reads ISO2709 binary porridge from the given file and creates MARC::Record-objects out of it.
1489
1490 @PARAM1, String, absolute path to the ISO2709 file.
1491 @PARAM2, String, see stage_file.pl
1492 @PARAM3, String, should be utf8
1493
1494 Returns two array refs.
1495
1496 =cut
1497
1498 sub RecordsFromISO2709File {
1499     my ($input_file, $record_type, $encoding) = @_;
1500     my @errors;
1501
1502     my $marc_type = C4::Context->preference('marcflavour');
1503     $marc_type .= 'AUTH' if ($marc_type eq 'UNIMARC' && $record_type eq 'auth');
1504
1505     open IN, "<$input_file" or die "$0: cannot open input file $input_file: $!\n";
1506     my @marc_records;
1507     $/ = "\035";
1508     while (<IN>) {
1509         s/^\s+//;
1510         s/\s+$//;
1511         next unless $_; # skip if record has only whitespace, as might occur
1512                         # if file includes newlines between each MARC record
1513         my ($marc_record, $charset_guessed, $char_errors) = MarcToUTF8Record($_, $marc_type, $encoding);
1514         push @marc_records, $marc_record;
1515         if ($charset_guessed ne $encoding) {
1516             push @errors,
1517                 "Unexpected charset $charset_guessed, expecting $encoding";
1518         }
1519     }
1520     close IN;
1521     return ( \@errors, \@marc_records );
1522 }
1523
1524 =head2 RecordsFromMARCXMLFile
1525
1526     my ($errors, $records) = C4::ImportBatch::RecordsFromMARCXMLFile($input_file, $encoding);
1527
1528 Creates MARC::Record-objects out of the given MARCXML-file.
1529
1530 @PARAM1, String, absolute path to the ISO2709 file.
1531 @PARAM2, String, should be utf8
1532
1533 Returns two array refs.
1534
1535 =cut
1536
1537 sub RecordsFromMARCXMLFile {
1538     my ( $filename, $encoding ) = @_;
1539     my $batch = MARC::File::XML->in( $filename );
1540     my ( @marcRecords, @errors, $record );
1541     do {
1542         eval { $record = $batch->next( $encoding ); };
1543         if ($@) {
1544             push @errors, $@;
1545         }
1546         push @marcRecords, $record if $record;
1547     } while( $record );
1548     return (\@errors, \@marcRecords);
1549 }
1550
1551 =head2 RecordsFromMarcPlugin
1552
1553     Converts text of input_file into array of MARC records with to_marc plugin
1554
1555 =cut
1556
1557 sub RecordsFromMarcPlugin {
1558     my ($input_file, $plugin_class, $encoding) = @_;
1559     my ( $text, @return );
1560     return \@return if !$input_file || !$plugin_class;
1561
1562     # Read input file
1563     open IN, "<$input_file" or die "$0: cannot open input file $input_file: $!\n";
1564     $/ = "\035";
1565     while (<IN>) {
1566         s/^\s+//;
1567         s/\s+$//;
1568         next unless $_;
1569         $text .= $_;
1570     }
1571     close IN;
1572
1573     # Convert to large MARC blob with plugin
1574     $text = Koha::Plugins::Handler->run({
1575         class  => $plugin_class,
1576         method => 'to_marc',
1577         params => { data => $text },
1578     }) if $text;
1579
1580     # Convert to array of MARC records
1581     if( $text ) {
1582         my $marc_type = C4::Context->preference('marcflavour');
1583         foreach my $blob ( split(/\x1D/, $text) ) {
1584             next if $blob =~ /^\s*$/;
1585             my ($marcrecord) = MarcToUTF8Record($blob, $marc_type, $encoding);
1586             push @return, $marcrecord;
1587         }
1588     }
1589     return \@return;
1590 }
1591
1592 # internal functions
1593
1594 sub _create_import_record {
1595     my ($batch_id, $record_sequence, $marc_record, $record_type, $encoding, $z3950random, $marc_type) = @_;
1596
1597     my $dbh = C4::Context->dbh;
1598     my $sth = $dbh->prepare("INSERT INTO import_records (import_batch_id, record_sequence, marc, marcxml,
1599                                                          record_type, encoding, z3950random)
1600                                     VALUES (?, ?, ?, ?, ?, ?, ?)");
1601     $sth->execute($batch_id, $record_sequence, $marc_record->as_usmarc(), $marc_record->as_xml($marc_type),
1602                   $record_type, $encoding, $z3950random);
1603     my $import_record_id = $dbh->{'mysql_insertid'};
1604     $sth->finish();
1605     return $import_record_id;
1606 }
1607
1608 sub _update_import_record_marc {
1609     my ($import_record_id, $marc_record, $marc_type) = @_;
1610
1611     my $dbh = C4::Context->dbh;
1612     my $sth = $dbh->prepare("UPDATE import_records SET marc = ?, marcxml = ?
1613                              WHERE  import_record_id = ?");
1614     $sth->execute($marc_record->as_usmarc(), $marc_record->as_xml($marc_type), $import_record_id);
1615     $sth->finish();
1616 }
1617
1618 sub _add_auth_fields {
1619     my ($import_record_id, $marc_record) = @_;
1620
1621     my $controlnumber;
1622     if ($marc_record->field('001')) {
1623         $controlnumber = $marc_record->field('001')->data();
1624     }
1625     my $authorized_heading = C4::AuthoritiesMarc::GetAuthorizedHeading({ record => $marc_record });
1626     my $dbh = C4::Context->dbh;
1627     my $sth = $dbh->prepare("INSERT INTO import_auths (import_record_id, control_number, authorized_heading) VALUES (?, ?, ?)");
1628     $sth->execute($import_record_id, $controlnumber, $authorized_heading);
1629     $sth->finish();
1630 }
1631
1632 sub _add_biblio_fields {
1633     my ($import_record_id, $marc_record) = @_;
1634
1635     my ($title, $author, $isbn, $issn) = _parse_biblio_fields($marc_record);
1636     my $dbh = C4::Context->dbh;
1637     # FIXME no controlnumber, originalsource
1638     $isbn = C4::Koha::GetNormalizedISBN($isbn);
1639     my $sth = $dbh->prepare("INSERT INTO import_biblios (import_record_id, title, author, isbn, issn) VALUES (?, ?, ?, ?, ?)");
1640     $sth->execute($import_record_id, $title, $author, $isbn, $issn);
1641     $sth->finish();
1642
1643 }
1644
1645 sub _update_biblio_fields {
1646     my ($import_record_id, $marc_record) = @_;
1647
1648     my ($title, $author, $isbn, $issn) = _parse_biblio_fields($marc_record);
1649     my $dbh = C4::Context->dbh;
1650     # FIXME no controlnumber, originalsource
1651     # FIXME 2 - should regularize normalization of ISBN wherever it is done
1652     $isbn =~ s/\(.*$//;
1653     $isbn =~ tr/ -_//;
1654     $isbn = uc $isbn;
1655     my $sth = $dbh->prepare("UPDATE import_biblios SET title = ?, author = ?, isbn = ?, issn = ?
1656                              WHERE  import_record_id = ?");
1657     $sth->execute($title, $author, $isbn, $issn, $import_record_id);
1658     $sth->finish();
1659 }
1660
1661 sub _parse_biblio_fields {
1662     my ($marc_record) = @_;
1663
1664     my $dbh = C4::Context->dbh;
1665     my $bibliofields = TransformMarcToKoha($marc_record, '');
1666     return ($bibliofields->{'title'}, $bibliofields->{'author'}, $bibliofields->{'isbn'}, $bibliofields->{'issn'});
1667
1668 }
1669
1670 sub _update_batch_record_counts {
1671     my ($batch_id) = @_;
1672
1673     my $dbh = C4::Context->dbh;
1674     my $sth = $dbh->prepare_cached("UPDATE import_batches SET
1675                                         num_records = (
1676                                             SELECT COUNT(*)
1677                                             FROM import_records
1678                                             WHERE import_batch_id = import_batches.import_batch_id),
1679                                         num_items = (
1680                                             SELECT COUNT(*)
1681                                             FROM import_records
1682                                             JOIN import_items USING (import_record_id)
1683                                             WHERE import_batch_id = import_batches.import_batch_id
1684                                             AND record_type = 'biblio')
1685                                     WHERE import_batch_id = ?");
1686     $sth->bind_param(1, $batch_id);
1687     $sth->execute();
1688     $sth->finish();
1689 }
1690
1691 sub _get_commit_action {
1692     my ($overlay_action, $nomatch_action, $item_action, $overlay_status, $import_record_id, $record_type) = @_;
1693
1694     if ($record_type eq 'biblio') {
1695         my ($bib_result, $bib_match, $item_result);
1696
1697         if ($overlay_status ne 'no_match') {
1698             $bib_match = GetBestRecordMatch($import_record_id);
1699             if ($overlay_action eq 'replace') {
1700                 $bib_result  = defined($bib_match) ? 'replace' : 'create_new';
1701             } elsif ($overlay_action eq 'create_new') {
1702                 $bib_result  = 'create_new';
1703             } elsif ($overlay_action eq 'ignore') {
1704                 $bib_result  = 'ignore';
1705             }
1706          if($item_action eq 'always_add' or $item_action eq 'add_only_for_matches'){
1707                 $item_result = 'create_new';
1708        }
1709       elsif($item_action eq 'replace'){
1710           $item_result = 'replace';
1711           }
1712       else {
1713              $item_result = 'ignore';
1714            }
1715         } else {
1716             $bib_result = $nomatch_action;
1717             $item_result = ($item_action eq 'always_add' or $item_action eq 'add_only_for_new')     ? 'create_new' : 'ignore';
1718         }
1719         return ($bib_result, $item_result, $bib_match);
1720     } else { # must be auths
1721         my ($auth_result, $auth_match);
1722
1723         if ($overlay_status ne 'no_match') {
1724             $auth_match = GetBestRecordMatch($import_record_id);
1725             if ($overlay_action eq 'replace') {
1726                 $auth_result  = defined($auth_match) ? 'replace' : 'create_new';
1727             } elsif ($overlay_action eq 'create_new') {
1728                 $auth_result  = 'create_new';
1729             } elsif ($overlay_action eq 'ignore') {
1730                 $auth_result  = 'ignore';
1731             }
1732         } else {
1733             $auth_result = $nomatch_action;
1734         }
1735
1736         return ($auth_result, undef, $auth_match);
1737
1738     }
1739 }
1740
1741 sub _get_revert_action {
1742     my ($overlay_action, $overlay_status, $status) = @_;
1743
1744     my $bib_result;
1745
1746     if ($status eq 'ignored') {
1747         $bib_result = 'ignore';
1748     } else {
1749         if ($overlay_action eq 'create_new') {
1750             $bib_result = 'delete';
1751         } else {
1752             $bib_result = ($overlay_status eq 'match_applied') ? 'restore' : 'delete';
1753         }
1754     }
1755     return $bib_result;
1756 }
1757
1758 1;
1759 __END__
1760
1761 =head1 AUTHOR
1762
1763 Koha Development Team <http://koha-community.org/>
1764
1765 Galen Charlton <galen.charlton@liblime.com>
1766
1767 =cut