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