C4/ImportBatch.pm

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