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::Items;
  31 use Koha::Plugins::Handler;
  32 use Koha::Logger;
  33
  34 use vars qw(@ISA @EXPORT @EXPORT_OK %EXPORT_TAGS);
  35
  36 BEGIN {
  37         require Exporter;
  38         @ISA    = qw(Exporter);
  39         @EXPORT = qw(
  40     GetZ3950BatchId
  41     GetWebserviceBatchId
  42     GetImportRecordMarc
  43     GetImportRecordMarcXML
  44     AddImportBatch
  45     GetImportBatch
  46     AddAuthToBatch
  47     AddBiblioToBatch
  48     AddItemsToImportBiblio
  49     ModAuthorityInBatch
  50     ModBiblioInBatch
  51
  52     BatchStageMarcRecords
  53     BatchFindDuplicates
  54     BatchCommitRecords
  55     BatchRevertRecords
  56     CleanBatch
  57     DeleteBatch
  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 our $logger = Koha::Logger->get( { category => 'C4.ImportBatch' } );
  87
  88 =head1 NAME
  89
  90 C4::ImportBatch - manage batches of imported MARC records
  91
  92 =head1 SYNOPSIS
  93
  94 use C4::ImportBatch;
  95
  96 =head1 FUNCTIONS
  97
  98 =head2 GetZ3950BatchId
  99
 100   my $batchid = GetZ3950BatchId($z3950server);
 101
 102 Retrieves the ID of the import batch for the Z39.50
 103 reservoir for the given target.  If necessary,
 104 creates the import batch.
 105
 106 =cut
 107
 108 sub GetZ3950BatchId {
 109     my ($z3950server) = @_;
 110
 111     my $dbh = C4::Context->dbh;
 112     my $sth = $dbh->prepare("SELECT import_batch_id FROM import_batches
 113                              WHERE  batch_type = 'z3950'
 114                              AND    file_name = ?");
 115     $sth->execute($z3950server);
 116     my $rowref = $sth->fetchrow_arrayref();
 117     $sth->finish();
 118     if (defined $rowref) {
 119         return $rowref->[0];
 120     } else {
 121         my $batch_id = AddImportBatch( {
 122                 overlay_action => 'create_new',
 123                 import_status => 'staged',
 124                 batch_type => 'z3950',
 125                 file_name => $z3950server,
 126             } );
 127         return $batch_id;
 128     }
 129
 130 }
 131
 132 =head2 GetWebserviceBatchId
 133
 134   my $batchid = GetWebserviceBatchId();
 135
 136 Retrieves the ID of the import batch for webservice.
 137 If necessary, creates the import batch.
 138
 139 =cut
 140
 141 my $WEBSERVICE_BASE_QRY = <<EOQ;
 142 SELECT import_batch_id FROM import_batches
 143 WHERE  batch_type = 'webservice'
 144 AND    import_status = 'staged'
 145 EOQ
 146 sub GetWebserviceBatchId {
 147     my ($params) = @_;
 148
 149     my $dbh = C4::Context->dbh;
 150     my $sql = $WEBSERVICE_BASE_QRY;
 151     my @args;
 152     foreach my $field (qw(matcher_id overlay_action nomatch_action item_action)) {
 153         if (my $val = $params->{$field}) {
 154             $sql .= " AND $field = ?";
 155             push @args, $val;
 156         }
 157     }
 158     my $id = $dbh->selectrow_array($sql, undef, @args);
 159     return $id if $id;
 160
 161     $params->{batch_type} = 'webservice';
 162     $params->{import_status} = 'staged';
 163     return AddImportBatch($params);
 164 }
 165
 166 =head2 GetImportRecordMarc
 167
 168   my ($marcblob, $encoding) = GetImportRecordMarc($import_record_id);
 169
 170 =cut
 171
 172 sub GetImportRecordMarc {
 173     my ($import_record_id) = @_;
 174
 175     my $dbh = C4::Context->dbh;
 176     my ( $marc, $encoding ) = $dbh->selectrow_array(q|
 177         SELECT marc, encoding
 178         FROM import_records
 179         WHERE import_record_id = ?
 180     |, undef, $import_record_id );
 181
 182     return $marc, $encoding;
 183 }
 184
 185 sub GetRecordFromImportBiblio {
 186     my ( $import_record_id, $embed_items ) = @_;
 187
 188     my ($marc) = GetImportRecordMarc($import_record_id);
 189     my $record = MARC::Record->new_from_usmarc($marc);
 190
 191     EmbedItemsInImportBiblio( $record, $import_record_id ) if $embed_items;
 192
 193     return $record;
 194 }
 195
 196 sub EmbedItemsInImportBiblio {
 197     my ( $record, $import_record_id ) = @_;
 198     my ( $itemtag, $itemsubfield ) = GetMarcFromKohaField( "items.itemnumber" );
 199     my $dbh = C4::Context->dbh;
 200     my $import_items = $dbh->selectall_arrayref(q|
 201         SELECT import_items.marcxml
 202         FROM import_items
 203         WHERE import_record_id = ?
 204     |, { Slice => {} }, $import_record_id );
 205     my @item_fields;
 206     for my $import_item ( @$import_items ) {
 207         my $item_marc = MARC::Record::new_from_xml($import_item->{marcxml}, 'UTF-8');
 208         push @item_fields, $item_marc->field($itemtag);
 209     }
 210     $record->append_fields(@item_fields);
 211     return $record;
 212 }
 213
 214 =head2 GetImportRecordMarcXML
 215
 216   my $marcxml = GetImportRecordMarcXML($import_record_id);
 217
 218 =cut
 219
 220 sub GetImportRecordMarcXML {
 221     my ($import_record_id) = @_;
 222
 223     my $dbh = C4::Context->dbh;
 224     my $sth = $dbh->prepare("SELECT marcxml FROM import_records WHERE import_record_id = ?");
 225     $sth->execute($import_record_id);
 226     my ($marcxml) = $sth->fetchrow();
 227     $sth->finish();
 228     return $marcxml;
 229
 230 }
 231
 232 =head2 AddImportBatch
 233
 234   my $batch_id = AddImportBatch($params_hash);
 235
 236 =cut
 237
 238 sub AddImportBatch {
 239     my ($params) = @_;
 240
 241     my (@fields, @vals);
 242     foreach (qw( matcher_id template_id branchcode
 243                  overlay_action nomatch_action item_action
 244                  import_status batch_type file_name comments record_type )) {
 245         if (exists $params->{$_}) {
 246             push @fields, $_;
 247             push @vals, $params->{$_};
 248         }
 249     }
 250     my $dbh = C4::Context->dbh;
 251     $dbh->do("INSERT INTO import_batches (".join( ',', @fields).")
 252                                   VALUES (".join( ',', map '?', @fields).")",
 253              undef,
 254              @vals);
 255     return $dbh->{'mysql_insertid'};
 256 }
 257
 258 =head2 GetImportBatch
 259
 260   my $row = GetImportBatch($batch_id);
 261
 262 Retrieve a hashref of an import_batches row.
 263
 264 =cut
 265
 266 sub GetImportBatch {
 267     my ($batch_id) = @_;
 268
 269     my $dbh = C4::Context->dbh;
 270     my $sth = $dbh->prepare_cached("SELECT * FROM import_batches WHERE import_batch_id = ?");
 271     $sth->bind_param(1, $batch_id);
 272     $sth->execute();
 273     my $result = $sth->fetchrow_hashref;
 274     $sth->finish();
 275     return $result;
 276
 277 }
 278
 279 =head2 AddBiblioToBatch
 280
 281   my $import_record_id = AddBiblioToBatch($batch_id, $record_sequence,
 282                 $marc_record, $encoding, $update_counts);
 283
 284 =cut
 285
 286 sub AddBiblioToBatch {
 287     my $batch_id = shift;
 288     my $record_sequence = shift;
 289     my $marc_record = shift;
 290     my $encoding = shift;
 291     my $update_counts = @_ ? shift : 1;
 292
 293     my $import_record_id = _create_import_record($batch_id, $record_sequence, $marc_record, 'biblio', $encoding, 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, $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 $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, $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     $record_type,                $encoding,
 355     $marc_records,               $file_name,
 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 $marc_modification_template = shift;
 370     my $comments = shift;
 371     my $branch_code = shift;
 372     my $parse_items = shift;
 373     my $leave_as_staging = shift;
 374
 375     # optional callback to monitor status
 376     # of job
 377     my $progress_interval = 0;
 378     my $progress_callback = undef;
 379     if ($#_ == 1) {
 380         $progress_interval = shift;
 381         $progress_callback = shift;
 382         $progress_interval = 0 unless $progress_interval =~ /^\d+$/ and $progress_interval > 0;
 383         $progress_interval = 0 unless 'CODE' eq ref $progress_callback;
 384     }
 385
 386     my $batch_id = AddImportBatch( {
 387             overlay_action => 'create_new',
 388             import_status => 'staging',
 389             batch_type => 'batch',
 390             file_name => $file_name,
 391             comments => $comments,
 392             record_type => $record_type,
 393         } );
 394     if ($parse_items) {
 395         SetImportBatchItemAction($batch_id, 'always_add');
 396     } else {
 397         SetImportBatchItemAction($batch_id, 'ignore');
 398     }
 399
 400
 401     my $marc_type = C4::Context->preference('marcflavour');
 402     $marc_type .= 'AUTH' if ($marc_type eq 'UNIMARC' && $record_type eq 'auth');
 403     my @invalid_records = ();
 404     my $num_valid = 0;
 405     my $num_items = 0;
 406     # FIXME - for now, we're dealing only with bibs
 407     my $rec_num = 0;
 408     foreach my $marc_record (@$marc_records) {
 409         $rec_num++;
 410         if ($progress_interval and (0 == ($rec_num % $progress_interval))) {
 411             &$progress_callback($rec_num);
 412         }
 413
 414         ModifyRecordWithTemplate( $marc_modification_template, $marc_record ) if ( $marc_modification_template );
 415
 416         my $import_record_id;
 417         if (scalar($marc_record->fields()) == 0) {
 418             push @invalid_records, $marc_record;
 419         } else {
 420
 421             # Normalize the record so it doesn't have separated diacritics
 422             SetUTF8Flag($marc_record);
 423
 424             $num_valid++;
 425             if ($record_type eq 'biblio') {
 426                 $import_record_id = AddBiblioToBatch($batch_id, $rec_num, $marc_record, $encoding, int(rand(99999)), 0);
 427                 if ($parse_items) {
 428                     my @import_items_ids = AddItemsToImportBiblio($batch_id, $import_record_id, $marc_record, 0);
 429                     $num_items += scalar(@import_items_ids);
 430                 }
 431             } elsif ($record_type eq 'auth') {
 432                 $import_record_id = AddAuthToBatch($batch_id, $rec_num, $marc_record, $encoding, int(rand(99999)), 0, $marc_type);
 433             }
 434         }
 435     }
 436     unless ($leave_as_staging) {
 437         SetImportBatchStatus($batch_id, 'staged');
 438     }
 439     # FIXME branch_code, number of bibs, number of items
 440     _update_batch_record_counts($batch_id);
 441     return ($batch_id, $num_valid, $num_items, @invalid_records);
 442 }
 443
 444 =head2 AddItemsToImportBiblio
 445
 446   my @import_items_ids = AddItemsToImportBiblio($batch_id,
 447                 $import_record_id, $marc_record, $update_counts);
 448
 449 =cut
 450
 451 sub AddItemsToImportBiblio {
 452     my $batch_id = shift;
 453     my $import_record_id = shift;
 454     my $marc_record = shift;
 455     my $update_counts = @_ ? shift : 0;
 456
 457     my @import_items_ids = ();
 458
 459     my $dbh = C4::Context->dbh;
 460     my ($item_tag,$item_subfield) = &GetMarcFromKohaField( "items.itemnumber" );
 461     foreach my $item_field ($marc_record->field($item_tag)) {
 462         my $item_marc = MARC::Record->new();
 463         $item_marc->leader("00000    a              "); # must set Leader/09 to 'a'
 464         $item_marc->append_fields($item_field);
 465         $marc_record->delete_field($item_field);
 466         my $sth = $dbh->prepare_cached("INSERT INTO import_items (import_record_id, status, marcxml)
 467                                         VALUES (?, ?, ?)");
 468         $sth->bind_param(1, $import_record_id);
 469         $sth->bind_param(2, 'staged');
 470         $sth->bind_param(3, $item_marc->as_xml("USMARC"));
 471         $sth->execute();
 472         push @import_items_ids, $dbh->{'mysql_insertid'};
 473         $sth->finish();
 474     }
 475
 476     if ($#import_items_ids > -1) {
 477         _update_batch_record_counts($batch_id) if $update_counts;
 478         _update_import_record_marc($import_record_id, $marc_record, C4::Context->preference('marcflavour'));
 479     }
 480     return @import_items_ids;
 481 }
 482
 483 =head2 BatchFindDuplicates
 484
 485   my $num_with_matches = BatchFindDuplicates($batch_id, $matcher,
 486              $max_matches, $progress_interval, $progress_callback);
 487
 488 Goes through the records loaded in the batch and attempts to
 489 find duplicates for each one.  Sets the matching status
 490 of each record to "no_match" or "auto_match" as appropriate.
 491
 492 The $max_matches parameter is optional; if it is not supplied,
 493 it defaults to 10.
 494
 495 The $progress_interval and $progress_callback parameters are
 496 optional; if both are supplied, the sub referred to by
 497 $progress_callback will be invoked every $progress_interval
 498 records using the number of records processed as the
 499 singular argument.
 500
 501 =cut
 502
 503 sub BatchFindDuplicates {
 504     my $batch_id = shift;
 505     my $matcher = shift;
 506     my $max_matches = @_ ? shift : 10;
 507
 508     # optional callback to monitor status
 509     # of job
 510     my $progress_interval = 0;
 511     my $progress_callback = undef;
 512     if ($#_ == 1) {
 513         $progress_interval = shift;
 514         $progress_callback = shift;
 515         $progress_interval = 0 unless $progress_interval =~ /^\d+$/ and $progress_interval > 0;
 516         $progress_interval = 0 unless 'CODE' eq ref $progress_callback;
 517     }
 518
 519     my $dbh = C4::Context->dbh;
 520
 521     my $sth = $dbh->prepare("SELECT import_record_id, record_type, marc
 522                              FROM import_records
 523                              WHERE import_batch_id = ?");
 524     $sth->execute($batch_id);
 525     my $num_with_matches = 0;
 526     my $rec_num = 0;
 527     while (my $rowref = $sth->fetchrow_hashref) {
 528         $rec_num++;
 529         if ($progress_interval and (0 == ($rec_num % $progress_interval))) {
 530             &$progress_callback($rec_num);
 531         }
 532         my $marc_record = MARC::Record->new_from_usmarc($rowref->{'marc'});
 533         my @matches = ();
 534         if (defined $matcher) {
 535             @matches = $matcher->get_matches($marc_record, $max_matches);
 536         }
 537         if (scalar(@matches) > 0) {
 538             $num_with_matches++;
 539             SetImportRecordMatches($rowref->{'import_record_id'}, @matches);
 540             SetImportRecordOverlayStatus($rowref->{'import_record_id'}, 'auto_match');
 541         } else {
 542             SetImportRecordMatches($rowref->{'import_record_id'}, ());
 543             SetImportRecordOverlayStatus($rowref->{'import_record_id'}, 'no_match');
 544         }
 545     }
 546     $sth->finish();
 547     return $num_with_matches;
 548 }
 549
 550 =head2 BatchCommitRecords
 551
 552   my ($num_added, $num_updated, $num_items_added, $num_items_replaced, $num_items_errored, $num_ignored) =
 553         BatchCommitRecords($batch_id, $framework,
 554         $progress_interval, $progress_callback);
 555
 556 =cut
 557
 558 sub BatchCommitRecords {
 559     my $batch_id = shift;
 560     my $framework = shift;
 561
 562     # optional callback to monitor status
 563     # of job
 564     my $progress_interval = 0;
 565     my $progress_callback = undef;
 566     if ($#_ == 1) {
 567         $progress_interval = shift;
 568         $progress_callback = shift;
 569         $progress_interval = 0 unless $progress_interval =~ /^\d+$/ and $progress_interval > 0;
 570         $progress_interval = 0 unless 'CODE' eq ref $progress_callback;
 571     }
 572
 573     my $record_type;
 574     my $num_added = 0;
 575     my $num_updated = 0;
 576     my $num_items_added = 0;
 577     my $num_items_replaced = 0;
 578     my $num_items_errored = 0;
 579     my $num_ignored = 0;
 580     # commit (i.e., save, all records in the batch)
 581     SetImportBatchStatus('importing');
 582     my $overlay_action = GetImportBatchOverlayAction($batch_id);
 583     my $nomatch_action = GetImportBatchNoMatchAction($batch_id);
 584     my $item_action = GetImportBatchItemAction($batch_id);
 585     my $item_tag;
 586     my $item_subfield;
 587     my $dbh = C4::Context->dbh;
 588     my $sth = $dbh->prepare("SELECT import_records.import_record_id, record_type, status, overlay_status, marc, encoding
 589                              FROM import_records
 590                              LEFT JOIN import_auths ON (import_records.import_record_id=import_auths.import_record_id)
 591                              LEFT JOIN import_biblios ON (import_records.import_record_id=import_biblios.import_record_id)
 592                              WHERE import_batch_id = ?");
 593     $sth->execute($batch_id);
 594     my $marcflavour = C4::Context->preference('marcflavour');
 595     my $rec_num = 0;
 596     while (my $rowref = $sth->fetchrow_hashref) {
 597         $record_type = $rowref->{'record_type'};
 598         $rec_num++;
 599         if ($progress_interval and (0 == ($rec_num % $progress_interval))) {
 600             &$progress_callback($rec_num);
 601         }
 602         if ($rowref->{'status'} eq 'error' or $rowref->{'status'} eq 'imported') {
 603             $num_ignored++;
 604             next;
 605         }
 606
 607         my $marc_type;
 608         if ($marcflavour eq 'UNIMARC' && $record_type eq 'auth') {
 609             $marc_type = 'UNIMARCAUTH';
 610         } elsif ($marcflavour eq 'UNIMARC') {
 611             $marc_type = 'UNIMARC';
 612         } else {
 613             $marc_type = 'USMARC';
 614         }
 615         my $marc_record = MARC::Record->new_from_usmarc($rowref->{'marc'});
 616
 617         if ($record_type eq 'biblio') {
 618             # remove any item tags - rely on BatchCommitItems
 619             ($item_tag,$item_subfield) = &GetMarcFromKohaField( "items.itemnumber" );
 620             foreach my $item_field ($marc_record->field($item_tag)) {
 621                 $marc_record->delete_field($item_field);
 622             }
 623         }
 624
 625         my ($record_result, $item_result, $record_match) =
 626             _get_commit_action($overlay_action, $nomatch_action, $item_action,
 627                                $rowref->{'overlay_status'}, $rowref->{'import_record_id'}, $record_type);
 628
 629         my $recordid;
 630         my $query;
 631         if ($record_result eq 'create_new') {
 632             $num_added++;
 633             if ($record_type eq 'biblio') {
 634                 my $biblioitemnumber;
 635                 ($recordid, $biblioitemnumber) = AddBiblio($marc_record, $framework);
 636                 $query = "UPDATE import_biblios SET matched_biblionumber = ? WHERE import_record_id = ?";
 637                 if ($item_result eq 'create_new' || $item_result eq 'replace') {
 638                     my ($bib_items_added, $bib_items_replaced, $bib_items_errored) = BatchCommitItems($rowref->{'import_record_id'}, $recordid, $item_result);
 639                     $num_items_added += $bib_items_added;
 640                     $num_items_replaced += $bib_items_replaced;
 641                     $num_items_errored += $bib_items_errored;
 642                 }
 643             } else {
 644                 $recordid = AddAuthority($marc_record, undef, GuessAuthTypeCode($marc_record));
 645                 $query = "UPDATE import_auths SET matched_authid = ? WHERE import_record_id = ?";
 646             }
 647             my $sth = $dbh->prepare_cached($query);
 648             $sth->execute($recordid, $rowref->{'import_record_id'});
 649             $sth->finish();
 650             SetImportRecordStatus($rowref->{'import_record_id'}, 'imported');
 651         } elsif ($record_result eq 'replace') {
 652             $num_updated++;
 653             $recordid = $record_match;
 654             my $oldxml;
 655             if ($record_type eq 'biblio') {
 656                 my $oldbiblio = Koha::Biblios->find( $recordid );
 657                 $oldxml = GetXmlBiblio($recordid);
 658
 659                 # remove item fields so that they don't get
 660                 # added again if record is reverted
 661                 # 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.
 662                 my $old_marc = MARC::Record->new_from_xml(StripNonXmlChars($oldxml), 'UTF-8', $rowref->{'encoding'}, $marc_type);
 663                 foreach my $item_field ($old_marc->field($item_tag)) {
 664                     $old_marc->delete_field($item_field);
 665                 }
 666                 $oldxml = $old_marc->as_xml($marc_type);
 667
 668                 ModBiblio($marc_record, $recordid, $oldbiblio->frameworkcode);
 669                 $query = "UPDATE import_biblios SET matched_biblionumber = ? WHERE import_record_id = ?";
 670
 671                 if ($item_result eq 'create_new' || $item_result eq 'replace') {
 672                     my ($bib_items_added, $bib_items_replaced, $bib_items_errored) = BatchCommitItems($rowref->{'import_record_id'}, $recordid, $item_result);
 673                     $num_items_added += $bib_items_added;
 674                     $num_items_replaced += $bib_items_replaced;
 675                     $num_items_errored += $bib_items_errored;
 676                 }
 677             } else {
 678                 $oldxml = GetAuthorityXML($recordid);
 679
 680                 ModAuthority($recordid, $marc_record, GuessAuthTypeCode($marc_record));
 681                 $query = "UPDATE import_auths SET matched_authid = ? WHERE import_record_id = ?";
 682             }
 683             my $sth = $dbh->prepare_cached("UPDATE import_records SET marcxml_old = ? WHERE import_record_id = ?");
 684             $sth->execute($oldxml, $rowref->{'import_record_id'});
 685             $sth->finish();
 686             my $sth2 = $dbh->prepare_cached($query);
 687             $sth2->execute($recordid, $rowref->{'import_record_id'});
 688             $sth2->finish();
 689             SetImportRecordOverlayStatus($rowref->{'import_record_id'}, 'match_applied');
 690             SetImportRecordStatus($rowref->{'import_record_id'}, 'imported');
 691         } elsif ($record_result eq 'ignore') {
 692             $recordid = $record_match;
 693             $num_ignored++;
 694             $recordid = $record_match;
 695             if ($record_type eq 'biblio' and defined $recordid and ( $item_result eq 'create_new' || $item_result eq 'replace' ) ) {
 696                 my ($bib_items_added, $bib_items_replaced, $bib_items_errored) = BatchCommitItems($rowref->{'import_record_id'}, $recordid, $item_result);
 697                 $num_items_added += $bib_items_added;
 698          $num_items_replaced += $bib_items_replaced;
 699                 $num_items_errored += $bib_items_errored;
 700                 # still need to record the matched biblionumber so that the
 701                 # items can be reverted
 702                 my $sth2 = $dbh->prepare_cached("UPDATE import_biblios SET matched_biblionumber = ? WHERE import_record_id = ?");
 703                 $sth2->execute($recordid, $rowref->{'import_record_id'});
 704                 SetImportRecordOverlayStatus($rowref->{'import_record_id'}, 'match_applied');
 705             }
 706             SetImportRecordStatus($rowref->{'import_record_id'}, 'ignored');
 707         }
 708     }
 709     $sth->finish();
 710     SetImportBatchStatus($batch_id, 'imported');
 711     return ($num_added, $num_updated, $num_items_added, $num_items_replaced, $num_items_errored, $num_ignored);
 712 }
 713
 714 =head2 BatchCommitItems
 715
 716   ($num_items_added, $num_items_errored) =
 717          BatchCommitItems($import_record_id, $biblionumber);
 718
 719 =cut
 720
 721 sub BatchCommitItems {
 722     my ( $import_record_id, $biblionumber, $action ) = @_;
 723
 724     my $dbh = C4::Context->dbh;
 725
 726     my $num_items_added = 0;
 727     my $num_items_errored = 0;
 728     my $num_items_replaced = 0;
 729
 730     my $sth = $dbh->prepare( "
 731         SELECT import_items_id, import_items.marcxml, encoding
 732         FROM import_items
 733         JOIN import_records USING (import_record_id)
 734         WHERE import_record_id = ?
 735         ORDER BY import_items_id
 736     " );
 737     $sth->bind_param( 1, $import_record_id );
 738     $sth->execute();
 739
 740     while ( my $row = $sth->fetchrow_hashref() ) {
 741         my $item_marc = MARC::Record->new_from_xml( StripNonXmlChars( $row->{'marcxml'} ), 'UTF-8', $row->{'encoding'} );
 742
 743         # Delete date_due subfield as to not accidentally delete item checkout due dates
 744         my ( $MARCfield, $MARCsubfield ) = GetMarcFromKohaField( 'items.onloan' );
 745         $item_marc->field($MARCfield)->delete_subfield( code => $MARCsubfield );
 746
 747         my $item = TransformMarcToKoha( $item_marc );
 748
 749         my $duplicate_barcode = exists( $item->{'barcode'} ) && Koha::Items->find({ barcode => $item->{'barcode'} });
 750         my $duplicate_itemnumber = exists( $item->{'itemnumber'} );
 751
 752         my $updsth = $dbh->prepare("UPDATE import_items SET status = ?, itemnumber = ? WHERE import_items_id = ?");
 753         if ( $action eq "replace" && $duplicate_itemnumber ) {
 754             # Duplicate itemnumbers have precedence, that way we can update barcodes by overlaying
 755             ModItemFromMarc( $item_marc, $biblionumber, $item->{itemnumber} );
 756             $updsth->bind_param( 1, 'imported' );
 757             $updsth->bind_param( 2, $item->{itemnumber} );
 758             $updsth->bind_param( 3, $row->{'import_items_id'} );
 759             $updsth->execute();
 760             $updsth->finish();
 761             $num_items_replaced++;
 762         } elsif ( $action eq "replace" && $duplicate_barcode ) {
 763             my $itemnumber = $duplicate_barcode->itemnumber;
 764             ModItemFromMarc( $item_marc, $biblionumber, $itemnumber );
 765             $updsth->bind_param( 1, 'imported' );
 766             $updsth->bind_param( 2, $item->{itemnumber} );
 767             $updsth->bind_param( 3, $row->{'import_items_id'} );
 768             $updsth->execute();
 769             $updsth->finish();
 770             $num_items_replaced++;
 771         } elsif ($duplicate_barcode) {
 772             $updsth->bind_param( 1, 'error' );
 773             $updsth->bind_param( 2, 'duplicate item barcode' );
 774             $updsth->bind_param( 3, $row->{'import_items_id'} );
 775             $updsth->execute();
 776             $num_items_errored++;
 777         } else {
 778             my ( $item_biblionumber, $biblioitemnumber, $itemnumber ) = AddItemFromMarc( $item_marc, $biblionumber );
 779             if( $itemnumber ) {
 780                 $updsth->bind_param( 1, 'imported' );
 781                 $updsth->bind_param( 2, $itemnumber );
 782                 $updsth->bind_param( 3, $row->{'import_items_id'} );
 783                 $updsth->execute();
 784                 $updsth->finish();
 785                 $num_items_added++;
 786             }
 787         }
 788     }
 789
 790     return ( $num_items_added, $num_items_replaced, $num_items_errored );
 791 }
 792
 793 =head2 BatchRevertRecords
 794
 795   my ($num_deleted, $num_errors, $num_reverted, $num_items_deleted,
 796       $num_ignored) = BatchRevertRecords($batch_id);
 797
 798 =cut
 799
 800 sub BatchRevertRecords {
 801     my $batch_id = shift;
 802
 803     $logger->trace("C4::ImportBatch::BatchRevertRecords( $batch_id )");
 804
 805     my $record_type;
 806     my $num_deleted = 0;
 807     my $num_errors = 0;
 808     my $num_reverted = 0;
 809     my $num_ignored = 0;
 810     my $num_items_deleted = 0;
 811     # commit (i.e., save, all records in the batch)
 812     SetImportBatchStatus('reverting');
 813     my $overlay_action = GetImportBatchOverlayAction($batch_id);
 814     my $nomatch_action = GetImportBatchNoMatchAction($batch_id);
 815     my $dbh = C4::Context->dbh;
 816     my $sth = $dbh->prepare("SELECT import_records.import_record_id, record_type, status, overlay_status, marcxml_old, encoding, matched_biblionumber, matched_authid
 817                              FROM import_records
 818                              LEFT JOIN import_auths ON (import_records.import_record_id=import_auths.import_record_id)
 819                              LEFT JOIN import_biblios ON (import_records.import_record_id=import_biblios.import_record_id)
 820                              WHERE import_batch_id = ?");
 821     $sth->execute($batch_id);
 822     my $marc_type;
 823     my $marcflavour = C4::Context->preference('marcflavour');
 824     while (my $rowref = $sth->fetchrow_hashref) {
 825         $record_type = $rowref->{'record_type'};
 826         if ($rowref->{'status'} eq 'error' or $rowref->{'status'} eq 'reverted') {
 827             $num_ignored++;
 828             next;
 829         }
 830         if ($marcflavour eq 'UNIMARC' && $record_type eq 'auth') {
 831             $marc_type = 'UNIMARCAUTH';
 832         } elsif ($marcflavour eq 'UNIMARC') {
 833             $marc_type = 'UNIMARC';
 834         } else {
 835             $marc_type = 'USMARC';
 836         }
 837
 838         my $record_result = _get_revert_action($overlay_action, $rowref->{'overlay_status'}, $rowref->{'status'});
 839
 840         if ($record_result eq 'delete') {
 841             my $error = undef;
 842             if  ($record_type eq 'biblio') {
 843                 $num_items_deleted += BatchRevertItems($rowref->{'import_record_id'}, $rowref->{'matched_biblionumber'});
 844                 $error = DelBiblio($rowref->{'matched_biblionumber'});
 845             } else {
 846                 DelAuthority({ authid => $rowref->{'matched_authid'} });
 847             }
 848             if (defined $error) {
 849                 $num_errors++;
 850             } else {
 851                 $num_deleted++;
 852                 SetImportRecordStatus($rowref->{'import_record_id'}, 'reverted');
 853             }
 854         } elsif ($record_result eq 'restore') {
 855             $num_reverted++;
 856             my $old_record = MARC::Record->new_from_xml(StripNonXmlChars($rowref->{'marcxml_old'}), 'UTF-8', $rowref->{'encoding'}, $marc_type);
 857             if ($record_type eq 'biblio') {
 858                 my $biblionumber = $rowref->{'matched_biblionumber'};
 859                 my $oldbiblio = Koha::Biblios->find( $biblionumber );
 860
 861                 $logger->info("C4::ImportBatch::BatchRevertRecords: Biblio record $biblionumber does not exist, restoration of this record was skipped") unless $oldbiblio;
 862                 next unless $oldbiblio; # Record has since been deleted. Deleted records should stay deleted.
 863
 864                 $num_items_deleted += BatchRevertItems($rowref->{'import_record_id'}, $rowref->{'matched_biblionumber'});
 865                 ModBiblio($old_record, $biblionumber, $oldbiblio->frameworkcode);
 866             } else {
 867                 my $authid = $rowref->{'matched_authid'};
 868                 ModAuthority($authid, $old_record, GuessAuthTypeCode($old_record));
 869             }
 870             SetImportRecordStatus($rowref->{'import_record_id'}, 'reverted');
 871         } elsif ($record_result eq 'ignore') {
 872             if ($record_type eq 'biblio') {
 873                 $num_items_deleted += BatchRevertItems($rowref->{'import_record_id'}, $rowref->{'matched_biblionumber'});
 874             }
 875             SetImportRecordStatus($rowref->{'import_record_id'}, 'reverted');
 876         }
 877         my $query;
 878         if ($record_type eq 'biblio') {
 879             # remove matched_biblionumber only if there is no 'imported' item left
 880             $query = "UPDATE import_biblios SET matched_biblionumber = NULL WHERE import_record_id = ?";
 881             $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')";
 882         } else {
 883             $query = "UPDATE import_auths SET matched_authid = NULL WHERE import_record_id = ?";
 884         }
 885         my $sth2 = $dbh->prepare_cached($query);
 886         $sth2->execute($rowref->{'import_record_id'});
 887     }
 888
 889     $sth->finish();
 890     SetImportBatchStatus($batch_id, 'reverted');
 891     return ($num_deleted, $num_errors, $num_reverted, $num_items_deleted, $num_ignored);
 892 }
 893
 894 =head2 BatchRevertItems
 895
 896   my $num_items_deleted = BatchRevertItems($import_record_id, $biblionumber);
 897
 898 =cut
 899
 900 sub BatchRevertItems {
 901     my ($import_record_id, $biblionumber) = @_;
 902
 903     my $dbh = C4::Context->dbh;
 904     my $num_items_deleted = 0;
 905
 906     my $sth = $dbh->prepare_cached("SELECT import_items_id, itemnumber
 907                                    FROM import_items
 908                                    JOIN items USING (itemnumber)
 909                                    WHERE import_record_id = ?");
 910     $sth->bind_param(1, $import_record_id);
 911     $sth->execute();
 912     while (my $row = $sth->fetchrow_hashref()) {
 913         my $item = Koha::Items->find($row->{itemnumber});
 914         my $error = $item->safe_delete;
 915         if ($error eq '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( { $_ eq $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, $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, marcxml_old,
1599                                                          record_type, encoding)
1600                                     VALUES (?, ?, ?, ?, ?, ?, ?)");
1601     $sth->execute($batch_id, $record_sequence, $marc_record->as_usmarc(), $marc_record->as_xml($marc_type), '',
1602                   $record_type, $encoding);
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