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($VERSION @ISA @EXPORT @EXPORT_OK %EXPORT_TAGS);
  34
  35 BEGIN {
  36         # set the version for version checking
  37     $VERSION = 3.07.00.049;
  38         require Exporter;
  39         @ISA    = qw(Exporter);
  40         @EXPORT = qw(
  41     GetZ3950BatchId
  42     GetWebserviceBatchId
  43     GetImportRecordMarc
  44     GetImportRecordMarcXML
  45     AddImportBatch
  46     GetImportBatch
  47     AddAuthToBatch
  48     AddBiblioToBatch
  49     AddItemsToImportBiblio
  50     ModAuthorityInBatch
  51     ModBiblioInBatch
  52
  53     BatchStageMarcRecords
  54     BatchFindDuplicates
  55     BatchCommitRecords
  56     BatchRevertRecords
  57     CleanBatch
  58
  59     GetAllImportBatches
  60     GetStagedWebserviceBatches
  61     GetImportBatchRangeDesc
  62     GetNumberOfNonZ3950ImportBatches
  63     GetImportBiblios
  64     GetImportRecordsRange
  65         GetItemNumbersFromImportBatch
  66
  67     GetImportBatchStatus
  68     SetImportBatchStatus
  69     GetImportBatchOverlayAction
  70     SetImportBatchOverlayAction
  71     GetImportBatchNoMatchAction
  72     SetImportBatchNoMatchAction
  73     GetImportBatchItemAction
  74     SetImportBatchItemAction
  75     GetImportBatchMatcher
  76     SetImportBatchMatcher
  77     GetImportRecordOverlayStatus
  78     SetImportRecordOverlayStatus
  79     GetImportRecordStatus
  80     SetImportRecordStatus
  81     GetImportRecordMatches
  82     SetImportRecordMatches
  83         );
  84 }
  85
  86 =head1 NAME
  87
  88 C4::ImportBatch - manage batches of imported MARC records
  89
  90 =head1 SYNOPSIS
  91
  92 use C4::ImportBatch;
  93
  94 =head1 FUNCTIONS
  95
  96 =head2 GetZ3950BatchId
  97
  98   my $batchid = GetZ3950BatchId($z3950server);
  99
 100 Retrieves the ID of the import batch for the Z39.50
 101 reservoir for the given target.  If necessary,
 102 creates the import batch.
 103
 104 =cut
 105
 106 sub GetZ3950BatchId {
 107     my ($z3950server) = @_;
 108
 109     my $dbh = C4::Context->dbh;
 110     my $sth = $dbh->prepare("SELECT import_batch_id FROM import_batches
 111                              WHERE  batch_type = 'z3950'
 112                              AND    file_name = ?");
 113     $sth->execute($z3950server);
 114     my $rowref = $sth->fetchrow_arrayref();
 115     $sth->finish();
 116     if (defined $rowref) {
 117         return $rowref->[0];
 118     } else {
 119         my $batch_id = AddImportBatch( {
 120                 overlay_action => 'create_new',
 121                 import_status => 'staged',
 122                 batch_type => 'z3950',
 123                 file_name => $z3950server,
 124             } );
 125         return $batch_id;
 126     }
 127
 128 }
 129
 130 =head2 GetWebserviceBatchId
 131
 132   my $batchid = GetWebserviceBatchId();
 133
 134 Retrieves the ID of the import batch for webservice.
 135 If necessary, creates the import batch.
 136
 137 =cut
 138
 139 my $WEBSERVICE_BASE_QRY = <<EOQ;
 140 SELECT import_batch_id FROM import_batches
 141 WHERE  batch_type = 'webservice'
 142 AND    import_status = 'staged'
 143 EOQ
 144 sub GetWebserviceBatchId {
 145     my ($params) = @_;
 146
 147     my $dbh = C4::Context->dbh;
 148     my $sql = $WEBSERVICE_BASE_QRY;
 149     my @args;
 150     foreach my $field (qw(matcher_id overlay_action nomatch_action item_action)) {
 151         if (my $val = $params->{$field}) {
 152             $sql .= " AND $field = ?";
 153             push @args, $val;
 154         }
 155     }
 156     my $id = $dbh->selectrow_array($sql, undef, @args);
 157     return $id if $id;
 158
 159     $params->{batch_type} = 'webservice';
 160     $params->{import_status} = 'staged';
 161     return AddImportBatch($params);
 162 }
 163
 164 =head2 GetImportRecordMarc
 165
 166   my ($marcblob, $encoding) = GetImportRecordMarc($import_record_id);
 167
 168 =cut
 169
 170 sub GetImportRecordMarc {
 171     my ($import_record_id) = @_;
 172
 173     my $dbh = C4::Context->dbh;
 174     my ( $marc, $encoding ) = $dbh->selectrow_array(q|
 175         SELECT marc, encoding
 176         FROM import_records
 177         WHERE import_record_id = ?
 178     |, undef, $import_record_id );
 179
 180     return $marc, $encoding;
 181 }
 182
 183 sub GetRecordFromImportBiblio {
 184     my ( $import_record_id, $embed_items ) = @_;
 185
 186     my ($marc) = GetImportRecordMarc($import_record_id);
 187     my $record = MARC::Record->new_from_usmarc($marc);
 188
 189     EmbedItemsInImportBiblio( $record, $import_record_id ) if $embed_items;
 190
 191     return $record;
 192 }
 193
 194 sub EmbedItemsInImportBiblio {
 195     my ( $record, $import_record_id ) = @_;
 196     my ( $itemtag, $itemsubfield ) = GetMarcFromKohaField("items.itemnumber", '');
 197     my $dbh = C4::Context->dbh;
 198     my $import_items = $dbh->selectall_arrayref(q|
 199         SELECT import_items.marcxml
 200         FROM import_items
 201         WHERE import_record_id = ?
 202     |, { Slice => {} }, $import_record_id );
 203     my @item_fields;
 204     for my $import_item ( @$import_items ) {
 205         my $item_marc = MARC::Record::new_from_xml($import_item->{marcxml});
 206         push @item_fields, $item_marc->field($itemtag);
 207     }
 208     $record->append_fields(@item_fields);
 209     return $record;
 210 }
 211
 212 =head2 GetImportRecordMarcXML
 213
 214   my $marcxml = GetImportRecordMarcXML($import_record_id);
 215
 216 =cut
 217
 218 sub GetImportRecordMarcXML {
 219     my ($import_record_id) = @_;
 220
 221     my $dbh = C4::Context->dbh;
 222     my $sth = $dbh->prepare("SELECT marcxml FROM import_records WHERE import_record_id = ?");
 223     $sth->execute($import_record_id);
 224     my ($marcxml) = $sth->fetchrow();
 225     $sth->finish();
 226     return $marcxml;
 227
 228 }
 229
 230 =head2 AddImportBatch
 231
 232   my $batch_id = AddImportBatch($params_hash);
 233
 234 =cut
 235
 236 sub AddImportBatch {
 237     my ($params) = @_;
 238
 239     my (@fields, @vals);
 240     foreach (qw( matcher_id template_id branchcode
 241                  overlay_action nomatch_action item_action
 242                  import_status batch_type file_name comments record_type )) {
 243         if (exists $params->{$_}) {
 244             push @fields, $_;
 245             push @vals, $params->{$_};
 246         }
 247     }
 248     my $dbh = C4::Context->dbh;
 249     $dbh->do("INSERT INTO import_batches (".join( ',', @fields).")
 250                                   VALUES (".join( ',', map '?', @fields).")",
 251              undef,
 252              @vals);
 253     return $dbh->{'mysql_insertid'};
 254 }
 255
 256 =head2 GetImportBatch
 257
 258   my $row = GetImportBatch($batch_id);
 259
 260 Retrieve a hashref of an import_batches row.
 261
 262 =cut
 263
 264 sub GetImportBatch {
 265     my ($batch_id) = @_;
 266
 267     my $dbh = C4::Context->dbh;
 268     my $sth = $dbh->prepare_cached("SELECT * FROM import_batches WHERE import_batch_id = ?");
 269     $sth->bind_param(1, $batch_id);
 270     $sth->execute();
 271     my $result = $sth->fetchrow_hashref;
 272     $sth->finish();
 273     return $result;
 274
 275 }
 276
 277 =head2 AddBiblioToBatch
 278
 279   my $import_record_id = AddBiblioToBatch($batch_id, $record_sequence,
 280                 $marc_record, $encoding, $z3950random, $update_counts);
 281
 282 =cut
 283
 284 sub AddBiblioToBatch {
 285     my $batch_id = shift;
 286     my $record_sequence = shift;
 287     my $marc_record = shift;
 288     my $encoding = shift;
 289     my $z3950random = shift;
 290     my $update_counts = @_ ? shift : 1;
 291
 292     my $import_record_id = _create_import_record($batch_id, $record_sequence, $marc_record, 'biblio', $encoding, $z3950random, C4::Context->preference('marcflavour'));
 293     _add_biblio_fields($import_record_id, $marc_record);
 294     _update_batch_record_counts($batch_id) if $update_counts;
 295     return $import_record_id;
 296 }
 297
 298 =head2 ModBiblioInBatch
 299
 300   ModBiblioInBatch($import_record_id, $marc_record);
 301
 302 =cut
 303
 304 sub ModBiblioInBatch {
 305     my ($import_record_id, $marc_record) = @_;
 306
 307     _update_import_record_marc($import_record_id, $marc_record, C4::Context->preference('marcflavour'));
 308     _update_biblio_fields($import_record_id, $marc_record);
 309
 310 }
 311
 312 =head2 AddAuthToBatch
 313
 314   my $import_record_id = AddAuthToBatch($batch_id, $record_sequence,
 315                 $marc_record, $encoding, $z3950random, $update_counts, [$marc_type]);
 316
 317 =cut
 318
 319 sub AddAuthToBatch {
 320     my $batch_id = shift;
 321     my $record_sequence = shift;
 322     my $marc_record = shift;
 323     my $encoding = shift;
 324     my $z3950random = shift;
 325     my $update_counts = @_ ? shift : 1;
 326     my $marc_type = shift || C4::Context->preference('marcflavour');
 327
 328     $marc_type = 'UNIMARCAUTH' if $marc_type eq 'UNIMARC';
 329
 330     my $import_record_id = _create_import_record($batch_id, $record_sequence, $marc_record, 'auth', $encoding, $z3950random, $marc_type);
 331     _add_auth_fields($import_record_id, $marc_record);
 332     _update_batch_record_counts($batch_id) if $update_counts;
 333     return $import_record_id;
 334 }
 335
 336 =head2 ModAuthInBatch
 337
 338   ModAuthInBatch($import_record_id, $marc_record);
 339
 340 =cut
 341
 342 sub ModAuthInBatch {
 343     my ($import_record_id, $marc_record) = @_;
 344
 345     my $marcflavour = C4::Context->preference('marcflavour');
 346     _update_import_record_marc($import_record_id, $marc_record, $marcflavour eq 'UNIMARC' ? 'UNIMARCAUTH' : 'USMARC');
 347
 348 }
 349
 350 =head2 BatchStageMarcRecords
 351
 352 ( $batch_id, $num_records, $num_items, @invalid_records ) =
 353   BatchStageMarcRecords(
 354     $encoding,                   $marc_records,
 355     $file_name,                  $to_marc_plugin,
 356     $marc_modification_template, $comments,
 357     $branch_code,                $parse_items,
 358     $leave_as_staging,           $progress_interval,
 359     $progress_callback
 360   );
 361
 362 =cut
 363
 364 sub BatchStageMarcRecords {
 365     my $record_type = shift;
 366     my $encoding = shift;
 367     my $marc_records = shift;
 368     my $file_name = shift;
 369     my $to_marc_plugin = 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     $marc_records = Koha::Plugins::Handler->run(
 402         {
 403             class  => $to_marc_plugin,
 404             method => 'to_marc',
 405             params => { data => $marc_records }
 406         }
 407     ) if $to_marc_plugin;
 408
 409     my $marc_type = C4::Context->preference('marcflavour');
 410     $marc_type .= 'AUTH' if ($marc_type eq 'UNIMARC' && $record_type eq 'auth');
 411     my @invalid_records = ();
 412     my $num_valid = 0;
 413     my $num_items = 0;
 414     # FIXME - for now, we're dealing only with bibs
 415     my $rec_num = 0;
 416     foreach my $marc_blob (split(/\x1D/, $marc_records)) {
 417         $marc_blob =~ s/^\s+//g;
 418         $marc_blob =~ s/\s+$//g;
 419         next unless $marc_blob;
 420         $rec_num++;
 421         if ($progress_interval and (0 == ($rec_num % $progress_interval))) {
 422             &$progress_callback($rec_num);
 423         }
 424         my ($marc_record, $charset_guessed, $char_errors) =
 425             MarcToUTF8Record($marc_blob, $marc_type, $encoding);
 426
 427         $encoding = $charset_guessed unless $encoding;
 428
 429         ModifyRecordWithTemplate( $marc_modification_template, $marc_record ) if ( $marc_modification_template );
 430
 431         my $import_record_id;
 432         if (scalar($marc_record->fields()) == 0) {
 433             push @invalid_records, $marc_blob;
 434         } else {
 435
 436             # Normalize the record so it doesn't have separated diacritics
 437             SetUTF8Flag($marc_record);
 438
 439             $num_valid++;
 440             if ($record_type eq 'biblio') {
 441                 $import_record_id = AddBiblioToBatch($batch_id, $rec_num, $marc_record, $encoding, int(rand(99999)), 0);
 442                 if ($parse_items) {
 443                     my @import_items_ids = AddItemsToImportBiblio($batch_id, $import_record_id, $marc_record, 0);
 444                     $num_items += scalar(@import_items_ids);
 445                 }
 446             } elsif ($record_type eq 'auth') {
 447                 $import_record_id = AddAuthToBatch($batch_id, $rec_num, $marc_record, $encoding, int(rand(99999)), 0, $marc_type);
 448             }
 449         }
 450     }
 451     unless ($leave_as_staging) {
 452         SetImportBatchStatus($batch_id, 'staged');
 453     }
 454     # FIXME branch_code, number of bibs, number of items
 455     _update_batch_record_counts($batch_id);
 456     return ($batch_id, $num_valid, $num_items, @invalid_records);
 457 }
 458
 459 =head2 AddItemsToImportBiblio
 460
 461   my @import_items_ids = AddItemsToImportBiblio($batch_id,
 462                 $import_record_id, $marc_record, $update_counts);
 463
 464 =cut
 465
 466 sub AddItemsToImportBiblio {
 467     my $batch_id = shift;
 468     my $import_record_id = shift;
 469     my $marc_record = shift;
 470     my $update_counts = @_ ? shift : 0;
 471
 472     my @import_items_ids = ();
 473
 474     my $dbh = C4::Context->dbh;
 475     my ($item_tag,$item_subfield) = &GetMarcFromKohaField("items.itemnumber",'');
 476     foreach my $item_field ($marc_record->field($item_tag)) {
 477         my $item_marc = MARC::Record->new();
 478         $item_marc->leader("00000    a              "); # must set Leader/09 to 'a'
 479         $item_marc->append_fields($item_field);
 480         $marc_record->delete_field($item_field);
 481         my $sth = $dbh->prepare_cached("INSERT INTO import_items (import_record_id, status, marcxml)
 482                                         VALUES (?, ?, ?)");
 483         $sth->bind_param(1, $import_record_id);
 484         $sth->bind_param(2, 'staged');
 485         $sth->bind_param(3, $item_marc->as_xml());
 486         $sth->execute();
 487         push @import_items_ids, $dbh->{'mysql_insertid'};
 488         $sth->finish();
 489     }
 490
 491     if ($#import_items_ids > -1) {
 492         _update_batch_record_counts($batch_id) if $update_counts;
 493         _update_import_record_marc($import_record_id, $marc_record, C4::Context->preference('marcflavour'));
 494     }
 495     return @import_items_ids;
 496 }
 497
 498 =head2 BatchFindDuplicates
 499
 500   my $num_with_matches = BatchFindDuplicates($batch_id, $matcher,
 501              $max_matches, $progress_interval, $progress_callback);
 502
 503 Goes through the records loaded in the batch and attempts to
 504 find duplicates for each one.  Sets the matching status
 505 of each record to "no_match" or "auto_match" as appropriate.
 506
 507 The $max_matches parameter is optional; if it is not supplied,
 508 it defaults to 10.
 509
 510 The $progress_interval and $progress_callback parameters are
 511 optional; if both are supplied, the sub referred to by
 512 $progress_callback will be invoked every $progress_interval
 513 records using the number of records processed as the
 514 singular argument.
 515
 516 =cut
 517
 518 sub BatchFindDuplicates {
 519     my $batch_id = shift;
 520     my $matcher = shift;
 521     my $max_matches = @_ ? shift : 10;
 522
 523     # optional callback to monitor status
 524     # of job
 525     my $progress_interval = 0;
 526     my $progress_callback = undef;
 527     if ($#_ == 1) {
 528         $progress_interval = shift;
 529         $progress_callback = shift;
 530         $progress_interval = 0 unless $progress_interval =~ /^\d+$/ and $progress_interval > 0;
 531         $progress_interval = 0 unless 'CODE' eq ref $progress_callback;
 532     }
 533
 534     my $dbh = C4::Context->dbh;
 535
 536     my $sth = $dbh->prepare("SELECT import_record_id, record_type, marc
 537                              FROM import_records
 538                              WHERE import_batch_id = ?");
 539     $sth->execute($batch_id);
 540     my $num_with_matches = 0;
 541     my $rec_num = 0;
 542     while (my $rowref = $sth->fetchrow_hashref) {
 543         $rec_num++;
 544         if ($progress_interval and (0 == ($rec_num % $progress_interval))) {
 545             &$progress_callback($rec_num);
 546         }
 547         my $marc_record = MARC::Record->new_from_usmarc($rowref->{'marc'});
 548         my @matches = ();
 549         if (defined $matcher) {
 550             @matches = $matcher->get_matches($marc_record, $max_matches);
 551         }
 552         if (scalar(@matches) > 0) {
 553             $num_with_matches++;
 554             SetImportRecordMatches($rowref->{'import_record_id'}, @matches);
 555             SetImportRecordOverlayStatus($rowref->{'import_record_id'}, 'auto_match');
 556         } else {
 557             SetImportRecordMatches($rowref->{'import_record_id'}, ());
 558             SetImportRecordOverlayStatus($rowref->{'import_record_id'}, 'no_match');
 559         }
 560     }
 561     $sth->finish();
 562     return $num_with_matches;
 563 }
 564
 565 =head2 BatchCommitRecords
 566
 567   my ($num_added, $num_updated, $num_items_added, $num_items_replaced, $num_items_errored, $num_ignored) =
 568         BatchCommitRecords($batch_id, $framework,
 569         $progress_interval, $progress_callback);
 570
 571 =cut
 572
 573 sub BatchCommitRecords {
 574     my $batch_id = shift;
 575     my $framework = shift;
 576
 577     # optional callback to monitor status
 578     # of job
 579     my $progress_interval = 0;
 580     my $progress_callback = undef;
 581     if ($#_ == 1) {
 582         $progress_interval = shift;
 583         $progress_callback = shift;
 584         $progress_interval = 0 unless $progress_interval =~ /^\d+$/ and $progress_interval > 0;
 585         $progress_interval = 0 unless 'CODE' eq ref $progress_callback;
 586     }
 587
 588     my $record_type;
 589     my $num_added = 0;
 590     my $num_updated = 0;
 591     my $num_items_added = 0;
 592     my $num_items_replaced = 0;
 593     my $num_items_errored = 0;
 594     my $num_ignored = 0;
 595     # commit (i.e., save, all records in the batch)
 596     SetImportBatchStatus('importing');
 597     my $overlay_action = GetImportBatchOverlayAction($batch_id);
 598     my $nomatch_action = GetImportBatchNoMatchAction($batch_id);
 599     my $item_action = GetImportBatchItemAction($batch_id);
 600     my $item_tag;
 601     my $item_subfield;
 602     my $dbh = C4::Context->dbh;
 603     my $sth = $dbh->prepare("SELECT import_records.import_record_id, record_type, status, overlay_status, marc, encoding
 604                              FROM import_records
 605                              LEFT JOIN import_auths ON (import_records.import_record_id=import_auths.import_record_id)
 606                              LEFT JOIN import_biblios ON (import_records.import_record_id=import_biblios.import_record_id)
 607                              WHERE import_batch_id = ?");
 608     $sth->execute($batch_id);
 609     my $marcflavour = C4::Context->preference('marcflavour');
 610     my $rec_num = 0;
 611     while (my $rowref = $sth->fetchrow_hashref) {
 612         $record_type = $rowref->{'record_type'};
 613         $rec_num++;
 614         if ($progress_interval and (0 == ($rec_num % $progress_interval))) {
 615             &$progress_callback($rec_num);
 616         }
 617         if ($rowref->{'status'} eq 'error' or $rowref->{'status'} eq 'imported') {
 618             $num_ignored++;
 619             next;
 620         }
 621
 622         my $marc_type;
 623         if ($marcflavour eq 'UNIMARC' && $record_type eq 'auth') {
 624             $marc_type = 'UNIMARCAUTH';
 625         } elsif ($marcflavour eq 'UNIMARC') {
 626             $marc_type = 'UNIMARC';
 627         } else {
 628             $marc_type = 'USMARC';
 629         }
 630         my $marc_record = MARC::Record->new_from_usmarc($rowref->{'marc'});
 631
 632         if ($record_type eq 'biblio') {
 633             # remove any item tags - rely on BatchCommitItems
 634             ($item_tag,$item_subfield) = &GetMarcFromKohaField("items.itemnumber",'');
 635             foreach my $item_field ($marc_record->field($item_tag)) {
 636                 $marc_record->delete_field($item_field);
 637             }
 638         }
 639
 640         my ($record_result, $item_result, $record_match) =
 641             _get_commit_action($overlay_action, $nomatch_action, $item_action,
 642                                $rowref->{'overlay_status'}, $rowref->{'import_record_id'}, $record_type);
 643
 644         my $recordid;
 645         my $query;
 646         if ($record_result eq 'create_new') {
 647             $num_added++;
 648             if ($record_type eq 'biblio') {
 649                 my $biblioitemnumber;
 650                 ($recordid, $biblioitemnumber) = AddBiblio($marc_record, $framework);
 651                 $query = "UPDATE import_biblios SET matched_biblionumber = ? WHERE import_record_id = ?";
 652                 if ($item_result eq 'create_new' || $item_result eq 'replace') {
 653                     my ($bib_items_added, $bib_items_replaced, $bib_items_errored) = BatchCommitItems($rowref->{'import_record_id'}, $recordid, $item_result);
 654                     $num_items_added += $bib_items_added;
 655                     $num_items_replaced += $bib_items_replaced;
 656                     $num_items_errored += $bib_items_errored;
 657                 }
 658             } else {
 659                 $recordid = AddAuthority($marc_record, undef, GuessAuthTypeCode($marc_record));
 660                 $query = "UPDATE import_auths SET matched_authid = ? WHERE import_record_id = ?";
 661             }
 662             my $sth = $dbh->prepare_cached($query);
 663             $sth->execute($recordid, $rowref->{'import_record_id'});
 664             $sth->finish();
 665             SetImportRecordStatus($rowref->{'import_record_id'}, 'imported');
 666         } elsif ($record_result eq 'replace') {
 667             $num_updated++;
 668             $recordid = $record_match;
 669             my $oldxml;
 670             if ($record_type eq 'biblio') {
 671                 my $oldbiblio = GetBiblio($recordid);
 672                 $oldxml = GetXmlBiblio($recordid);
 673
 674                 # remove item fields so that they don't get
 675                 # added again if record is reverted
 676                 # 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.
 677                 my $old_marc = MARC::Record->new_from_xml(StripNonXmlChars($oldxml), 'UTF-8', $rowref->{'encoding'}, $marc_type);
 678                 foreach my $item_field ($old_marc->field($item_tag)) {
 679                     $old_marc->delete_field($item_field);
 680                 }
 681                 $oldxml = $old_marc->as_xml($marc_type);
 682
 683                 ModBiblio($marc_record, $recordid, $oldbiblio->{'frameworkcode'});
 684                 $query = "UPDATE import_biblios SET matched_biblionumber = ? WHERE import_record_id = ?";
 685
 686                 if ($item_result eq 'create_new' || $item_result eq 'replace') {
 687                     my ($bib_items_added, $bib_items_replaced, $bib_items_errored) = BatchCommitItems($rowref->{'import_record_id'}, $recordid, $item_result);
 688                     $num_items_added += $bib_items_added;
 689                     $num_items_replaced += $bib_items_replaced;
 690                     $num_items_errored += $bib_items_errored;
 691                 }
 692             } else {
 693                 $oldxml = GetAuthorityXML($recordid);
 694
 695                 ModAuthority($recordid, $marc_record, GuessAuthTypeCode($marc_record));
 696                 $query = "UPDATE import_auths SET matched_authid = ? WHERE import_record_id = ?";
 697             }
 698             my $sth = $dbh->prepare_cached("UPDATE import_records SET marcxml_old = ? WHERE import_record_id = ?");
 699             $sth->execute($oldxml, $rowref->{'import_record_id'});
 700             $sth->finish();
 701             my $sth2 = $dbh->prepare_cached($query);
 702             $sth2->execute($recordid, $rowref->{'import_record_id'});
 703             $sth2->finish();
 704             SetImportRecordOverlayStatus($rowref->{'import_record_id'}, 'match_applied');
 705             SetImportRecordStatus($rowref->{'import_record_id'}, 'imported');
 706         } elsif ($record_result eq 'ignore') {
 707             $recordid = $record_match;
 708             $num_ignored++;
 709             $recordid = $record_match;
 710             if ($record_type eq 'biblio' and defined $recordid and ( $item_result eq 'create_new' || $item_result eq 'replace' ) ) {
 711                 my ($bib_items_added, $bib_items_replaced, $bib_items_errored) = BatchCommitItems($rowref->{'import_record_id'}, $recordid, $item_result);
 712                 $num_items_added += $bib_items_added;
 713          $num_items_replaced += $bib_items_replaced;
 714                 $num_items_errored += $bib_items_errored;
 715                 # still need to record the matched biblionumber so that the
 716                 # items can be reverted
 717                 my $sth2 = $dbh->prepare_cached("UPDATE import_biblios SET matched_biblionumber = ? WHERE import_record_id = ?");
 718                 $sth2->execute($recordid, $rowref->{'import_record_id'});
 719                 SetImportRecordOverlayStatus($rowref->{'import_record_id'}, 'match_applied');
 720             }
 721             SetImportRecordStatus($rowref->{'import_record_id'}, 'ignored');
 722         }
 723     }
 724     $sth->finish();
 725     SetImportBatchStatus($batch_id, 'imported');
 726     return ($num_added, $num_updated, $num_items_added, $num_items_replaced, $num_items_errored, $num_ignored);
 727 }
 728
 729 =head2 BatchCommitItems
 730
 731   ($num_items_added, $num_items_errored) =
 732          BatchCommitItems($import_record_id, $biblionumber);
 733
 734 =cut
 735
 736 sub BatchCommitItems {
 737     my ( $import_record_id, $biblionumber, $action ) = @_;
 738
 739     my $dbh = C4::Context->dbh;
 740
 741     my $num_items_added = 0;
 742     my $num_items_errored = 0;
 743     my $num_items_replaced = 0;
 744
 745     my $sth = $dbh->prepare( "
 746         SELECT import_items_id, import_items.marcxml, encoding
 747         FROM import_items
 748         JOIN import_records USING (import_record_id)
 749         WHERE import_record_id = ?
 750         ORDER BY import_items_id
 751     " );
 752     $sth->bind_param( 1, $import_record_id );
 753     $sth->execute();
 754
 755     while ( my $row = $sth->fetchrow_hashref() ) {
 756         my $item_marc = MARC::Record->new_from_xml( StripNonXmlChars( $row->{'marcxml'} ), 'UTF-8', $row->{'encoding'} );
 757
 758         # Delete date_due subfield as to not accidentally delete item checkout due dates
 759         my ( $MARCfield, $MARCsubfield ) = GetMarcFromKohaField( 'items.onloan', GetFrameworkCode($biblionumber) );
 760         $item_marc->field($MARCfield)->delete_subfield( code => $MARCsubfield );
 761
 762         my $item = TransformMarcToKoha( $dbh, $item_marc );
 763
 764         my $duplicate_barcode = exists( $item->{'barcode'} ) && GetItemnumberFromBarcode( $item->{'barcode'} );
 765         my $duplicate_itemnumber = exists( $item->{'itemnumber'} );
 766
 767         my $updsth = $dbh->prepare("UPDATE import_items SET status = ?, itemnumber = ? WHERE import_items_id = ?");
 768         if ( $action eq "replace" && $duplicate_itemnumber ) {
 769             # Duplicate itemnumbers have precedence, that way we can update barcodes by overlaying
 770             ModItemFromMarc( $item_marc, $biblionumber, $item->{itemnumber} );
 771             $updsth->bind_param( 1, 'imported' );
 772             $updsth->bind_param( 2, $item->{itemnumber} );
 773             $updsth->bind_param( 3, $row->{'import_items_id'} );
 774             $updsth->execute();
 775             $updsth->finish();
 776             $num_items_replaced++;
 777         } elsif ( $action eq "replace" && $duplicate_barcode ) {
 778             my $itemnumber = GetItemnumberFromBarcode( $item->{'barcode'} );
 779             ModItemFromMarc( $item_marc, $biblionumber, $itemnumber );
 780             $updsth->bind_param( 1, 'imported' );
 781             $updsth->bind_param( 2, $item->{itemnumber} );
 782             $updsth->bind_param( 3, $row->{'import_items_id'} );
 783             $updsth->execute();
 784             $updsth->finish();
 785             $num_items_replaced++;
 786         } elsif ($duplicate_barcode) {
 787             $updsth->bind_param( 1, 'error' );
 788             $updsth->bind_param( 2, 'duplicate item barcode' );
 789             $updsth->bind_param( 3, $row->{'import_items_id'} );
 790             $updsth->execute();
 791             $num_items_errored++;
 792         } else {
 793             my ( $item_biblionumber, $biblioitemnumber, $itemnumber ) = AddItemFromMarc( $item_marc, $biblionumber );
 794             $updsth->bind_param( 1, 'imported' );
 795             $updsth->bind_param( 2, $itemnumber );
 796             $updsth->bind_param( 3, $row->{'import_items_id'} );
 797             $updsth->execute();
 798             $updsth->finish();
 799             $num_items_added++;
 800         }
 801     }
 802
 803     return ( $num_items_added, $num_items_replaced, $num_items_errored );
 804 }
 805
 806 =head2 BatchRevertRecords
 807
 808   my ($num_deleted, $num_errors, $num_reverted, $num_items_deleted,
 809       $num_ignored) = BatchRevertRecords($batch_id);
 810
 811 =cut
 812
 813 sub BatchRevertRecords {
 814     my $batch_id = shift;
 815
 816     my $logger = Koha::Logger->get( { category => 'C4.ImportBatch.BatchRevertRecords' } );
 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("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