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