Koha/SearchEngine/Elasticsearch.pm

   1 package Koha::SearchEngine::Elasticsearch;
   2
   3 # Copyright 2015 Catalyst IT
   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 base qw(Class::Accessor);
  21
  22 use C4::Context;
  23
  24 use Koha::Database;
  25 use Koha::Exceptions::Config;
  26 use Koha::Exceptions::Elasticsearch;
  27 use Koha::Filter::MARC::EmbedSeeFromHeadings;
  28 use Koha::SearchFields;
  29 use Koha::SearchMarcMaps;
  30 use Koha::Caches;
  31 use C4::Heading;
  32 use C4::AuthoritiesMarc qw( GuessAuthTypeCode );
  33 use C4::Biblio;
  34
  35 use Carp qw( carp croak );
  36 use Clone qw( clone );
  37 use Modern::Perl;
  38 use Readonly qw( Readonly );
  39 use Search::Elasticsearch;
  40 use Try::Tiny qw( catch try );
  41 use YAML::XS;
  42
  43 use List::Util qw( sum0 );
  44 use MARC::File::XML;
  45 use MIME::Base64 qw( encode_base64 );
  46 use Encode qw( encode );
  47 use Business::ISBN;
  48 use Scalar::Util qw( looks_like_number );
  49
  50 __PACKAGE__->mk_ro_accessors(qw( index index_name ));
  51 __PACKAGE__->mk_accessors(qw( sort_fields ));
  52
  53 # Constants to refer to the standard index names
  54 Readonly our $BIBLIOS_INDEX     => 'biblios';
  55 Readonly our $AUTHORITIES_INDEX => 'authorities';
  56
  57 =head1 NAME
  58
  59 Koha::SearchEngine::Elasticsearch - Base module for things using elasticsearch
  60
  61 =head1 ACCESSORS
  62
  63 =over 4
  64
  65 =item index
  66
  67 The name of the index to use, generally 'biblios' or 'authorities'.
  68
  69 =item index_name
  70
  71 The Elasticsearch index name with Koha instance prefix.
  72
  73 =back
  74
  75
  76 =head1 FUNCTIONS
  77
  78 =cut
  79
  80 sub new {
  81     my $class = shift @_;
  82     my ($params) = @_;
  83
  84     # Check for a valid index
  85     Koha::Exceptions::MissingParameter->throw('No index name provided') unless $params->{index};
  86     my $config = _read_configuration();
  87     $params->{index_name} = $config->{index_name} . '_' . $params->{index};
  88
  89     my $self = $class->SUPER::new(@_);
  90     return $self;
  91 }
  92
  93 =head2 get_elasticsearch
  94
  95     my $elasticsearch_client = $self->get_elasticsearch();
  96
  97 Returns a C<Search::Elasticsearch> client. The client is cached on a C<Koha::SearchEngine::ElasticSearch>
  98 instance level and will be reused if method is called multiple times.
  99
 100 =cut
 101
 102 sub get_elasticsearch {
 103     my $self = shift @_;
 104     unless (defined $self->{elasticsearch}) {
 105         $self->{elasticsearch} = Search::Elasticsearch->new(
 106             $self->get_elasticsearch_params()
 107         );
 108     }
 109     return $self->{elasticsearch};
 110 }
 111
 112 =head2 get_elasticsearch_params
 113
 114     my $params = $self->get_elasticsearch_params();
 115
 116 This provides a hashref that contains the parameters for connecting to the
 117 ElasicSearch servers, in the form:
 118
 119     {
 120         'nodes' => ['127.0.0.1:9200', 'anotherserver:9200'],
 121         'index_name' => 'koha_instance_index',
 122     }
 123
 124 This is configured by the following in the C<config> block in koha-conf.xml:
 125
 126     <elasticsearch>
 127         <server>127.0.0.1:9200</server>
 128         <server>anotherserver:9200</server>
 129         <index_name>koha_instance</index_name>
 130     </elasticsearch>
 131
 132 =cut
 133
 134 sub get_elasticsearch_params {
 135     my ($self) = @_;
 136
 137     my $conf;
 138     try {
 139         $conf = _read_configuration();
 140     } catch {
 141         if ( ref($_) eq 'Koha::Exceptions::Config::MissingEntry' ) {
 142             croak($_->message);
 143         }
 144     };
 145
 146     return $conf
 147 }
 148
 149 =head2 get_elasticsearch_settings
 150
 151     my $settings = $self->get_elasticsearch_settings();
 152
 153 This provides the settings provided to Elasticsearch when an index is created.
 154 These can do things like define tokenization methods.
 155
 156 A hashref containing the settings is returned.
 157
 158 =cut
 159
 160 sub get_elasticsearch_settings {
 161     my ($self) = @_;
 162
 163     # Use state to speed up repeated calls
 164     state $settings = undef;
 165     if (!defined $settings) {
 166         my $config_file = C4::Context->config('elasticsearch_index_config');
 167         $config_file ||= C4::Context->config('intranetdir') . '/admin/searchengine/elasticsearch/index_config.yaml';
 168         $settings = YAML::XS::LoadFile( $config_file );
 169     }
 170
 171     return $settings;
 172 }
 173
 174 =head2 get_elasticsearch_mappings
 175
 176     my $mappings = $self->get_elasticsearch_mappings();
 177
 178 This provides the mappings that get passed to Elasticsearch when an index is
 179 created.
 180
 181 =cut
 182
 183 sub get_elasticsearch_mappings {
 184     my ($self) = @_;
 185
 186     # Use state to speed up repeated calls
 187     state %all_mappings;
 188     state %sort_fields;
 189
 190     if (!defined $all_mappings{$self->index}) {
 191         $sort_fields{$self->index} = {};
 192         # Clone the general mapping to break ties with the original hash
 193         my $mappings = clone(_get_elasticsearch_field_config('general', ''));
 194         my $marcflavour = lc C4::Context->preference('marcflavour');
 195         $self->_foreach_mapping(
 196             sub {
 197                 my ( $name, $type, $facet, $suggestible, $sort, $search, $marc_type ) = @_;
 198                 return if $marc_type ne $marcflavour;
 199                 # TODO if this gets any sort of complexity to it, it should
 200                 # be broken out into its own function.
 201
 202                 # TODO be aware of date formats, but this requires pre-parsing
 203                 # as ES will simply reject anything with an invalid date.
 204                 my $es_type = 'text';
 205                 if ($type eq 'boolean') {
 206                     $es_type = 'boolean';
 207                 } elsif ($type eq 'number' || $type eq 'sum') {
 208                     $es_type = 'integer';
 209                 } elsif ($type eq 'isbn' || $type eq 'stdno') {
 210                     $es_type = 'stdno';
 211                 } elsif ($type eq 'year') {
 212                     $es_type = 'year';
 213                 } elsif ($type eq 'callnumber') {
 214                     $es_type = 'cn_sort';
 215                 }
 216
 217                 if ($search) {
 218                     $mappings->{properties}{$name} = _get_elasticsearch_field_config('search', $es_type);
 219                 }
 220
 221                 if ($facet) {
 222                     $mappings->{properties}{ $name . '__facet' } = _get_elasticsearch_field_config('facet', $es_type);
 223                 }
 224                 if ($suggestible) {
 225                     $mappings->{properties}{ $name . '__suggestion' } = _get_elasticsearch_field_config('suggestible', $es_type);
 226                 }
 227                 # Sort is a bit special as it can be true, false, undef.
 228                 # We care about "true" or "undef",
 229                 # "undef" means to do the default thing, which is make it sortable.
 230                 if (!defined $sort || $sort) {
 231                     $mappings->{properties}{ $name . '__sort' } = _get_elasticsearch_field_config('sort', $es_type);
 232                     $sort_fields{$self->index}{$name} = 1;
 233                 }
 234             }
 235         );
 236         $mappings->{properties}{ 'match-heading' } = _get_elasticsearch_field_config('search', 'text') if $self->index eq 'authorities';
 237         $all_mappings{$self->index} = $mappings;
 238     }
 239     $self->sort_fields(\%{$sort_fields{$self->index}});
 240     return $all_mappings{$self->index};
 241 }
 242
 243 =head2 raw_elasticsearch_mappings
 244
 245 Return elasticsearch mapping as it is in database.
 246 marc_type: marc21|unimarc
 247
 248 $raw_mappings = raw_elasticsearch_mappings( $marc_type )
 249
 250 =cut
 251
 252 sub raw_elasticsearch_mappings {
 253     my ( $marc_type ) = @_;
 254
 255     my $schema = Koha::Database->new()->schema();
 256
 257     my $search_fields = Koha::SearchFields->search({}, { order_by => { -asc => 'name' } });
 258
 259     my $mappings = {};
 260     while ( my $search_field = $search_fields->next ) {
 261
 262         my $marc_to_fields = $schema->resultset('SearchMarcToField')->search(
 263             { search_field_id => $search_field->id },
 264             {
 265                 join     => 'search_marc_map',
 266                 order_by => { -asc => ['search_marc_map.marc_type','search_marc_map.marc_field'] }
 267             }
 268         );
 269
 270         while ( my $marc_to_field = $marc_to_fields->next ) {
 271
 272             my $marc_map = $marc_to_field->search_marc_map;
 273
 274             next if $marc_type && $marc_map->marc_type ne $marc_type;
 275
 276             $mappings->{ $marc_map->index_name }{ $search_field->name }{label} = $search_field->label;
 277             $mappings->{ $marc_map->index_name }{ $search_field->name }{type} = $search_field->type;
 278             $mappings->{ $marc_map->index_name }{ $search_field->name }{mandatory} = $search_field->mandatory;
 279             $mappings->{ $marc_map->index_name }{ $search_field->name }{facet_order} = $search_field->facet_order if defined $search_field->facet_order;
 280             $mappings->{ $marc_map->index_name }{ $search_field->name }{weight} = $search_field->weight if defined $search_field->weight;
 281             $mappings->{ $marc_map->index_name }{ $search_field->name }{opac} = $search_field->opac if defined $search_field->opac;
 282             $mappings->{ $marc_map->index_name }{ $search_field->name }{staff_client} = $search_field->staff_client if defined $search_field->staff_client;
 283
 284             push (@{ $mappings->{ $marc_map->index_name }{ $search_field->name }{mappings} },
 285                 {
 286                     facet   => $marc_to_field->facet || '',
 287                     marc_type => $marc_map->marc_type,
 288                     marc_field => $marc_map->marc_field,
 289                     sort        => $marc_to_field->sort,
 290                     suggestible => $marc_to_field->suggestible || ''
 291                 });
 292
 293         }
 294     }
 295
 296     return $mappings;
 297 }
 298
 299 =head2 _get_elasticsearch_field_config
 300
 301 Get the Elasticsearch field config for the given purpose and data type.
 302
 303 $mapping = _get_elasticsearch_field_config('search', 'text');
 304
 305 =cut
 306
 307 sub _get_elasticsearch_field_config {
 308
 309     my ( $purpose, $type ) = @_;
 310
 311     # Use state to speed up repeated calls
 312     state $settings = undef;
 313     if (!defined $settings) {
 314         my $config_file = C4::Context->config('elasticsearch_field_config');
 315         $config_file ||= C4::Context->config('intranetdir') . '/admin/searchengine/elasticsearch/field_config.yaml';
 316         local $YAML::XS::Boolean = 'JSON::PP';
 317         $settings = YAML::XS::LoadFile( $config_file );
 318     }
 319
 320     if (!defined $settings->{$purpose}) {
 321         die "Field purpose $purpose not defined in field config";
 322     }
 323     if ($type eq '') {
 324         return $settings->{$purpose};
 325     }
 326     if (defined $settings->{$purpose}{$type}) {
 327         return $settings->{$purpose}{$type};
 328     }
 329     if (defined $settings->{$purpose}{'default'}) {
 330         return $settings->{$purpose}{'default'};
 331     }
 332     return;
 333 }
 334
 335 =head2 _load_elasticsearch_mappings
 336
 337 Load Elasticsearch mappings in the format of mappings.yaml.
 338
 339 $indexes = _load_elasticsearch_mappings();
 340
 341 =cut
 342
 343 sub _load_elasticsearch_mappings {
 344     my $mappings_yaml = C4::Context->config('elasticsearch_index_mappings');
 345     $mappings_yaml ||= C4::Context->config('intranetdir') . '/admin/searchengine/elasticsearch/mappings.yaml';
 346     return YAML::XS::LoadFile( $mappings_yaml );
 347 }
 348
 349 sub reset_elasticsearch_mappings {
 350     my ( $self ) = @_;
 351     my $indexes = $self->_load_elasticsearch_mappings();
 352
 353     Koha::SearchMarcMaps->delete;
 354     Koha::SearchFields->delete;
 355
 356     while ( my ( $index_name, $fields ) = each %$indexes ) {
 357         while ( my ( $field_name, $data ) = each %$fields ) {
 358
 359             my %sf_params = map { $_ => $data->{$_} } grep { exists $data->{$_} } qw/ type label weight staff_client opac facet_order mandatory/;
 360
 361             # Set default values
 362             $sf_params{staff_client} //= 1;
 363             $sf_params{opac} //= 1;
 364
 365             $sf_params{name} = $field_name;
 366
 367             my $search_field = Koha::SearchFields->find_or_create( \%sf_params, { key => 'name' } );
 368
 369             my $mappings = $data->{mappings};
 370             for my $mapping ( @$mappings ) {
 371                 my $marc_field = Koha::SearchMarcMaps->find_or_create({
 372                     index_name => $index_name,
 373                     marc_type => $mapping->{marc_type},
 374                     marc_field => $mapping->{marc_field}
 375                 });
 376                 $search_field->add_to_search_marc_maps($marc_field, {
 377                     facet => $mapping->{facet} || 0,
 378                     suggestible => $mapping->{suggestible} || 0,
 379                     sort => $mapping->{sort} // 1,
 380                     search => $mapping->{search} // 1
 381                 });
 382             }
 383         }
 384     }
 385
 386     $self->clear_search_fields_cache();
 387
 388     # FIXME return the mappings?
 389 }
 390
 391 # This overrides the accessor provided by Class::Accessor so that if
 392 # sort_fields isn't set, then it'll generate it.
 393 sub sort_fields {
 394     my $self = shift;
 395     if (@_) {
 396         $self->_sort_fields_accessor(@_);
 397         return;
 398     }
 399     my $val = $self->_sort_fields_accessor();
 400     return $val if $val;
 401
 402     # This will populate the accessor as a side effect
 403     $self->get_elasticsearch_mappings();
 404     return $self->_sort_fields_accessor();
 405 }
 406
 407 =head2 _process_mappings($mappings, $data, $record_document, $meta)
 408
 409     $self->_process_mappings($mappings, $marc_field_data, $record_document, 0)
 410
 411 Process all C<$mappings> targets operating on a specific MARC field C<$data>.
 412 Since we group all mappings by MARC field targets C<$mappings> will contain
 413 all targets for C<$data> and thus we need to fetch the MARC field only once.
 414 C<$mappings> will be applied to C<$record_document> and new field values added.
 415 The method has no return value.
 416
 417 =over 4
 418
 419 =item C<$mappings>
 420
 421 Arrayref of mappings containing arrayrefs in the format
 422 [C<$target>, C<$options>] where C<$target> is the name of the target field and
 423 C<$options> is a hashref containing processing directives for this particular
 424 mapping.
 425
 426 =item C<$data>
 427
 428 The source data from a MARC record field.
 429
 430 =item C<$record_document>
 431
 432 Hashref representing the Elasticsearch document on which mappings should be
 433 applied.
 434
 435 =item C<$meta>
 436
 437 A hashref containing metadata useful for enforcing per mapping rules. For
 438 example for providing extra context for mapping options, or treating mapping
 439 targets differently depending on type (sort, search, facet etc). Combining
 440 this metadata with the mapping options and metadata allows us to mutate the
 441 data per mapping, or even replace it with other data retrieved from the
 442 metadata context.
 443
 444 Current properties are:
 445
 446 C<altscript>: A boolean value indicating whether an alternate script presentation is being
 447 processed.
 448
 449 C<data_source>: The source of the $<data> argument. Possible values are: 'leader', 'control_field',
 450 'subfield' or 'subfields_group'.
 451
 452 C<code>: The code of the subfield C<$data> was retrieved, if C<data_source> is 'subfield'.
 453
 454 C<codes>: Subfield codes of the subfields group from which C<$data> was retrieved, if C<data_source>
 455 is 'subfields_group'.
 456
 457 C<field>: The original C<MARC::Record> object.
 458
 459 =back
 460
 461 =cut
 462
 463 sub _process_mappings {
 464     my ($_self, $mappings, $data, $record_document, $meta) = @_;
 465     foreach my $mapping (@{$mappings}) {
 466         my ($target, $options) = @{$mapping};
 467
 468         # Don't process sort fields for alternate scripts
 469         my $sort = $target =~ /__sort$/;
 470         if ($sort && $meta->{altscript}) {
 471             next;
 472         }
 473
 474         # Copy (scalar) data since can have multiple targets
 475         # with differing options for (possibly) mutating data
 476         # so need a different copy for each
 477         my $data_copy = $data;
 478         if (defined $options->{substr}) {
 479             my ($start, $length) = @{$options->{substr}};
 480             $data_copy = length($data) > $start ? substr $data_copy, $start, $length : '';
 481         }
 482
 483         # Add data to values array for callbacks processing
 484         my $values = [$data_copy];
 485
 486         # Value callbacks takes subfield data (or values from previous
 487         # callbacks) as argument, and returns a possibly different list of values.
 488         # Note that the returned list may also be empty.
 489         if (defined $options->{value_callbacks}) {
 490             foreach my $callback (@{$options->{value_callbacks}}) {
 491                 # Pass each value to current callback which returns a list
 492                 # (scalar is fine too) resulting either in a list or
 493                 # a list of lists that will be flattened by perl.
 494                 # The next callback will receive the possibly expanded list of values.
 495                 $values = [ map { $callback->($_) } @{$values} ];
 496             }
 497         }
 498
 499         # Skip mapping if all values has been removed
 500         next unless @{$values};
 501
 502         if (defined $options->{property}) {
 503             $values = [ map { { $options->{property} => $_ } if $_} @{$values} ];
 504         }
 505         if (defined $options->{nonfiling_characters_indicator}) {
 506             my $nonfiling_chars = $meta->{field}->indicator($options->{nonfiling_characters_indicator});
 507             $nonfiling_chars = looks_like_number($nonfiling_chars) ? int($nonfiling_chars) : 0;
 508             # Nonfiling chars does not make sense for multiple values
 509             # Only apply on first element
 510             $values->[0] = substr $values->[0], $nonfiling_chars;
 511         }
 512
 513         $values = [ grep(!/^$/, @{$values}) ];
 514
 515         $record_document->{$target} //= [];
 516         push @{$record_document->{$target}}, @{$values};
 517     }
 518 }
 519
 520 =head2 marc_records_to_documents($marc_records)
 521
 522     my $record_documents = $self->marc_records_to_documents($marc_records);
 523
 524 Using mappings stored in database convert C<$marc_records> to Elasticsearch documents.
 525
 526 Returns array of hash references, representing Elasticsearch documents,
 527 acceptable as body payload in C<Search::Elasticsearch> requests.
 528
 529 =over 4
 530
 531 =item C<$marc_documents>
 532
 533 Reference to array of C<MARC::Record> objects to be converted to Elasticsearch documents.
 534
 535 =back
 536
 537 =cut
 538
 539 sub marc_records_to_documents {
 540     my ($self, $records) = @_;
 541     my $rules = $self->_get_marc_mapping_rules();
 542     my $control_fields_rules = $rules->{control_fields};
 543     my $data_fields_rules = $rules->{data_fields};
 544     my $marcflavour = lc C4::Context->preference('marcflavour');
 545     my $use_array = C4::Context->preference('ElasticsearchMARCFormat') eq 'ARRAY';
 546
 547     my @record_documents;
 548
 549     my %auth_match_headings;
 550     if( $self->index eq 'authorities' ){
 551         my @auth_types = Koha::Authority::Types->search->as_list;
 552         %auth_match_headings = map { $_->authtypecode => $_->auth_tag_to_report } @auth_types;
 553     }
 554
 555     foreach my $record (@{$records}) {
 556         my $record_document = {};
 557
 558         if ( $self->index eq 'authorities' ){
 559             my $authtypecode = GuessAuthTypeCode( $record );
 560             if( $authtypecode ){
 561                 if( $authtypecode !~ m/_SUBD/ ){ #Subdivision records will not be used for linking and so don't require match-heading to be built
 562                     my $field = $record->field( $auth_match_headings{ $authtypecode } );
 563                     my $heading = C4::Heading->new_from_field( $field, undef, 1 ); #new auth heading
 564                     push @{$record_document->{'match-heading'}}, $heading->search_form if $heading;
 565                 }
 566             } else {
 567                 warn "Cannot determine authority type for record: " . $record->field('001')->as_string;
 568             }
 569         }
 570
 571         my $mappings = $rules->{leader};
 572         if ($mappings) {
 573             $self->_process_mappings($mappings, $record->leader(), $record_document, {
 574                     altscript => 0,
 575                     data_source => 'leader'
 576                 }
 577             );
 578         }
 579         foreach my $field ($record->fields()) {
 580             if ($field->is_control_field()) {
 581                 my $mappings = $control_fields_rules->{$field->tag()};
 582                 if ($mappings) {
 583                     $self->_process_mappings($mappings, $field->data(), $record_document, {
 584                             altscript => 0,
 585                             data_source => 'control_field',
 586                             field => $field
 587                         }
 588                     );
 589                 }
 590             }
 591             else {
 592                 my $tag = $field->tag();
 593                 # Handle alternate scripts in MARC 21
 594                 my $altscript = 0;
 595                 if ($marcflavour eq 'marc21' && $tag eq '880') {
 596                     my $sub6 = $field->subfield('6');
 597                     if ($sub6 =~ /^(...)-\d+/) {
 598                         $tag = $1;
 599                         $altscript = 1;
 600                     }
 601                 }
 602
 603                 my $data_field_rules = $data_fields_rules->{$tag};
 604                 if ($data_field_rules) {
 605                     my $subfields_mappings = $data_field_rules->{subfields};
 606                     my $wildcard_mappings = $subfields_mappings->{'*'};
 607                     foreach my $subfield ($field->subfields()) {
 608                         my ($code, $data) = @{$subfield};
 609                         my $mappings = $subfields_mappings->{$code} // [];
 610                         if ($wildcard_mappings) {
 611                             $mappings = [@{$mappings}, @{$wildcard_mappings}];
 612                         }
 613                         if (@{$mappings}) {
 614                             $self->_process_mappings($mappings, $data, $record_document, {
 615                                     altscript => $altscript,
 616                                     data_source => 'subfield',
 617                                     code => $code,
 618                                     field => $field
 619                                 }
 620                             );
 621                         }
 622                     }
 623
 624                     my $subfields_join_mappings = $data_field_rules->{subfields_join};
 625                     if ($subfields_join_mappings) {
 626                         foreach my $subfields_group (keys %{$subfields_join_mappings}) {
 627                             my $data_field = $field->clone; #copy field to preserve for alt scripts
 628                             $data_field->delete_subfield(match => qr/^$/); #remove empty subfields, otherwise they are printed as a space
 629                             my $data = $data_field->as_string( $subfields_group ); #get values for subfields as a combined string, preserving record order
 630                             if ($data) {
 631                                 $self->_process_mappings($subfields_join_mappings->{$subfields_group}, $data, $record_document, {
 632                                         altscript => $altscript,
 633                                         data_source => 'subfields_group',
 634                                         codes => $subfields_group,
 635                                         field => $field
 636                                     }
 637                                 );
 638                             }
 639                         }
 640                     }
 641                 }
 642             }
 643         }
 644
 645         if (C4::Context->preference('IncludeSeeFromInSearches') and $self->index eq 'biblios') {
 646             foreach my $field (Koha::Filter::MARC::EmbedSeeFromHeadings->new->fields($record)) {
 647                 my $data_field_rules = $data_fields_rules->{$field->tag()};
 648                 if ($data_field_rules) {
 649                     my $subfields_mappings = $data_field_rules->{subfields};
 650                     my $wildcard_mappings = $subfields_mappings->{'*'};
 651                     foreach my $subfield ($field->subfields()) {
 652                         my ($code, $data) = @{$subfield};
 653                         my @mappings;
 654                         push @mappings, @{ $subfields_mappings->{$code} } if $subfields_mappings->{$code};
 655                         push @mappings, @$wildcard_mappings if $wildcard_mappings;
 656                         # Do not include "see from" into these kind of fields
 657                         @mappings = grep { $_->[0] !~ /__(sort|facet|suggestion)$/ } @mappings;
 658                         if (@mappings) {
 659                             $self->_process_mappings(\@mappings, $data, $record_document, {
 660                                     data_source => 'subfield',
 661                                     code => $code,
 662                                     field => $field
 663                                 }
 664                             );
 665                         }
 666                     }
 667
 668                     my $subfields_join_mappings = $data_field_rules->{subfields_join};
 669                     if ($subfields_join_mappings) {
 670                         foreach my $subfields_group (keys %{$subfields_join_mappings}) {
 671                             my $data_field = $field->clone;
 672                             # remove empty subfields, otherwise they are printed as a space
 673                             $data_field->delete_subfield(match => qr/^$/);
 674                             my $data = $data_field->as_string( $subfields_group );
 675                             if ($data) {
 676                                 my @mappings = @{ $subfields_join_mappings->{$subfields_group} };
 677                                 # Do not include "see from" into these kind of fields
 678                                 @mappings = grep { $_->[0] !~ /__(sort|facet|suggestion)$/ } @mappings;
 679                                 $self->_process_mappings(\@mappings, $data, $record_document, {
 680                                         data_source => 'subfields_group',
 681                                         codes => $subfields_group,
 682                                         field => $field
 683                                     }
 684                                 );
 685                             }
 686                         }
 687                     }
 688                 }
 689             }
 690         }
 691
 692         foreach my $field (keys %{$rules->{defaults}}) {
 693             unless (defined $record_document->{$field}) {
 694                 $record_document->{$field} = $rules->{defaults}->{$field};
 695             }
 696         }
 697         foreach my $field (@{$rules->{sum}}) {
 698             if (defined $record_document->{$field}) {
 699                 # TODO: validate numeric? filter?
 700                 # TODO: Or should only accept fields without nested values?
 701                 # TODO: Quick and dirty, improve if needed
 702                 $record_document->{$field} = sum0(grep { !ref($_) && m/\d+(\.\d+)?/} @{$record_document->{$field}});
 703             }
 704         }
 705         # Index all applicable ISBN forms (ISBN-10 and ISBN-13 with and without dashes)
 706         foreach my $field (@{$rules->{isbn}}) {
 707             if (defined $record_document->{$field}) {
 708                 my @isbns = ();
 709                 foreach my $input_isbn (@{$record_document->{$field}}) {
 710                     my $isbn = Business::ISBN->new($input_isbn);
 711                     if (defined $isbn && $isbn->is_valid) {
 712                         my $isbn13 = $isbn->as_isbn13->as_string;
 713                         push @isbns, $isbn13;
 714                         $isbn13 =~ s/\-//g;
 715                         push @isbns, $isbn13;
 716
 717                         my $isbn10 = $isbn->as_isbn10;
 718                         if ($isbn10) {
 719                             $isbn10 = $isbn10->as_string;
 720                             push @isbns, $isbn10;
 721                             $isbn10 =~ s/\-//g;
 722                             push @isbns, $isbn10;
 723                         }
 724                     } else {
 725                         push @isbns, $input_isbn;
 726                     }
 727                 }
 728                 $record_document->{$field} = \@isbns;
 729             }
 730         }
 731
 732         # Remove duplicate values and collapse sort fields
 733         foreach my $field (keys %{$record_document}) {
 734             if (ref($record_document->{$field}) eq 'ARRAY') {
 735                 @{$record_document->{$field}} = do {
 736                     my %seen;
 737                     grep { !$seen{ref($_) eq 'HASH' && defined $_->{input} ? $_->{input} : $_}++ } @{$record_document->{$field}};
 738                 };
 739                 if ($field =~ /__sort$/) {
 740                     # Make sure to keep the sort field length sensible. 255 was chosen as a nice round value.
 741                     $record_document->{$field} = [substr(join(' ', @{$record_document->{$field}}), 0, 255)];
 742                 }
 743             }
 744         }
 745
 746         # TODO: Perhaps should check if $records_document non empty, but really should never be the case
 747         $record->encoding('UTF-8');
 748         if ($use_array) {
 749             $record_document->{'marc_data_array'} = $self->_marc_to_array($record);
 750             $record_document->{'marc_format'} = 'ARRAY';
 751         } else {
 752             my @warnings;
 753             {
 754                 # Temporarily intercept all warn signals (MARC::Record carps when record length > 99999)
 755                 local $SIG{__WARN__} = sub {
 756                     push @warnings, $_[0];
 757                 };
 758                 $record_document->{'marc_data'} = encode_base64(encode('UTF-8', $record->as_usmarc()));
 759             }
 760             if (@warnings) {
 761                 # Suppress warnings if record length exceeded
 762                 unless (substr($record->leader(), 0, 5) eq '99999') {
 763                     foreach my $warning (@warnings) {
 764                         carp $warning;
 765                     }
 766                 }
 767                 $record_document->{'marc_data'} = $record->as_xml_record($marcflavour);
 768                 $record_document->{'marc_format'} = 'MARCXML';
 769             }
 770             else {
 771                 $record_document->{'marc_format'} = 'base64ISO2709';
 772             }
 773         }
 774
 775         # Check if there is at least one available item
 776         if ($self->index eq $BIBLIOS_INDEX) {
 777             my ($tag, $code) = C4::Biblio::GetMarcFromKohaField('biblio.biblionumber');
 778             my $field = $record->field($tag);
 779             if ($field) {
 780                 my $biblionumber = $field->is_control_field ? $field->data : $field->subfield($code);
 781                 my $avail_items = Koha::Items->search({
 782                     biblionumber => $biblionumber,
 783                     onloan       => undef,
 784                     itemlost     => 0,
 785                 })->count;
 786
 787                 $record_document->{available} = $avail_items ? \1 : \0;
 788             }
 789         }
 790
 791         push @record_documents, $record_document;
 792     }
 793     return \@record_documents;
 794 }
 795
 796 =head2 _marc_to_array($record)
 797
 798     my @fields = _marc_to_array($record)
 799
 800 Convert a MARC::Record to an array modeled after MARC-in-JSON
 801 (see https://github.com/marc4j/marc4j/wiki/MARC-in-JSON-Description)
 802
 803 =over 4
 804
 805 =item C<$record>
 806
 807 A MARC::Record object
 808
 809 =back
 810
 811 =cut
 812
 813 sub _marc_to_array {
 814     my ($self, $record) = @_;
 815
 816     my $data = {
 817         leader => $record->leader(),
 818         fields => []
 819     };
 820     for my $field ($record->fields()) {
 821         my $tag = $field->tag();
 822         if ($field->is_control_field()) {
 823             push @{$data->{fields}}, {$tag => $field->data()};
 824         } else {
 825             my $subfields = ();
 826             foreach my $subfield ($field->subfields()) {
 827                 my ($code, $contents) = @{$subfield};
 828                 push @{$subfields}, {$code => $contents};
 829             }
 830             push @{$data->{fields}}, {
 831                 $tag => {
 832                     ind1 => $field->indicator(1),
 833                     ind2 => $field->indicator(2),
 834                     subfields => $subfields
 835                 }
 836             };
 837         }
 838     }
 839     return $data;
 840 }
 841
 842 =head2 _array_to_marc($data)
 843
 844     my $record = _array_to_marc($data)
 845
 846 Convert an array modeled after MARC-in-JSON to a MARC::Record
 847
 848 =over 4
 849
 850 =item C<$data>
 851
 852 An array modeled after MARC-in-JSON
 853 (see https://github.com/marc4j/marc4j/wiki/MARC-in-JSON-Description)
 854
 855 =back
 856
 857 =cut
 858
 859 sub _array_to_marc {
 860     my ($self, $data) = @_;
 861
 862     my $record = MARC::Record->new();
 863
 864     $record->leader($data->{leader});
 865     for my $field (@{$data->{fields}}) {
 866         my $tag = (keys %{$field})[0];
 867         $field = $field->{$tag};
 868         my $marc_field;
 869         if (ref($field) eq 'HASH') {
 870             my @subfields;
 871             foreach my $subfield (@{$field->{subfields}}) {
 872                 my $code = (keys %{$subfield})[0];
 873                 push @subfields, $code;
 874                 push @subfields, $subfield->{$code};
 875             }
 876             $marc_field = MARC::Field->new($tag, $field->{ind1}, $field->{ind2}, @subfields);
 877         } else {
 878             $marc_field = MARC::Field->new($tag, $field)
 879         }
 880         $record->append_fields($marc_field);
 881     }
 882 ;
 883     return $record;
 884 }
 885
 886 =head2 _field_mappings($facet, $suggestible, $sort, $search, $target_name, $target_type, $range)
 887
 888     my @mappings = _field_mappings($facet, $suggestible, $sort, $search, $target_name, $target_type, $range)
 889
 890 Get mappings, an internal data structure later used by
 891 L<_process_mappings($mappings, $data, $record_document, $meta)> to process MARC target
 892 data for a MARC mapping.
 893
 894 The returned C<$mappings> is not to to be confused with mappings provided by
 895 C<_foreach_mapping>, rather this sub accepts properties from a mapping as
 896 provided by C<_foreach_mapping> and expands it to this internal data structure.
 897 In the caller context (C<_get_marc_mapping_rules>) the returned C<@mappings>
 898 is then applied to each MARC target (leader, control field data, subfield or
 899 joined subfields) and integrated into the mapping rules data structure used in
 900 C<marc_records_to_documents> to transform MARC records into Elasticsearch
 901 documents.
 902
 903 =over 4
 904
 905 =item C<$facet>
 906
 907 Boolean indicating whether to create a facet field for this mapping.
 908
 909 =item C<$suggestible>
 910
 911 Boolean indicating whether to create a suggestion field for this mapping.
 912
 913 =item C<$sort>
 914
 915 Boolean indicating whether to create a sort field for this mapping.
 916
 917 =item C<$search>
 918
 919 Boolean indicating whether to create a search field for this mapping.
 920
 921 =item C<$target_name>
 922
 923 Elasticsearch document target field name.
 924
 925 =item C<$target_type>
 926
 927 Elasticsearch document target field type.
 928
 929 =item C<$range>
 930
 931 An optional range as a string in the format "<START>-<END>" or "<START>",
 932 where "<START>" and "<END>" are integers specifying a range that will be used
 933 for extracting a substring from MARC data as Elasticsearch field target value.
 934
 935 The first character position is "0", and the range is inclusive,
 936 so "0-2" means the first three characters of MARC data.
 937
 938 If only "<START>" is provided only one character at position "<START>" will
 939 be extracted.
 940
 941 =back
 942
 943 =cut
 944
 945 sub _field_mappings {
 946     my ($_self, $facet, $suggestible, $sort, $search, $target_name, $target_type, $range) = @_;
 947     my %mapping_defaults = ();
 948     my @mappings;
 949
 950     my $substr_args = undef;
 951     if (defined $range) {
 952         # TODO: use value_callback instead?
 953         my ($start, $end) = map(int, split /-/, $range, 2);
 954         $substr_args = [$start];
 955         push @{$substr_args}, (defined $end ? $end - $start + 1 : 1);
 956     }
 957     my $default_options = {};
 958     if ($substr_args) {
 959         $default_options->{substr} = $substr_args;
 960     }
 961
 962     # TODO: Should probably have per type value callback/hook
 963     # but hard code for now
 964     if ($target_type eq 'boolean') {
 965         $default_options->{value_callbacks} //= [];
 966         push @{$default_options->{value_callbacks}}, sub {
 967             my ($value) = @_;
 968             # Trim whitespace at both ends
 969             $value =~ s/^\s+|\s+$//g;
 970             return $value ? 'true' : 'false';
 971         };
 972     }
 973     elsif ($target_type eq 'year') {
 974         $default_options->{value_callbacks} //= [];
 975         # Only accept years containing digits and "u"
 976         push @{$default_options->{value_callbacks}}, sub {
 977             my ($value) = @_;
 978             # Replace "u" with "0" for sorting
 979             return map { s/[u\s]/0/gr } ( $value =~ /[0-9u\s]{4}/g );
 980         };
 981     }
 982
 983     if ($search) {
 984         my $mapping = [$target_name, $default_options];
 985         push @mappings, $mapping;
 986     }
 987
 988     my @suffixes = ();
 989     push @suffixes, 'facet' if $facet;
 990     push @suffixes, 'suggestion' if $suggestible;
 991     push @suffixes, 'sort' if !defined $sort || $sort;
 992
 993     foreach my $suffix (@suffixes) {
 994         my $mapping = ["${target_name}__$suffix"];
 995         # TODO: Hack, fix later in less hideous manner
 996         if ($suffix eq 'suggestion') {
 997             push @{$mapping}, {%{$default_options}, property => 'input'};
 998         }
 999         else {
1000             # Important! Make shallow clone, or we end up with the same hashref
1001             # shared by all mappings
1002             push @{$mapping}, {%{$default_options}};
1003         }
1004         push @mappings, $mapping;
1005     }
1006     return @mappings;
1007 };
1008
1009 =head2 _get_marc_mapping_rules
1010
1011     my $mapping_rules = $self->_get_marc_mapping_rules()
1012
1013 Generates rules from mappings stored in database for MARC records to Elasticsearch JSON document conversion.
1014
1015 Since field retrieval is slow in C<MARC::Records> (all fields are itereted through for
1016 each call to C<MARC::Record>->field) we create an optimized structure of mapping
1017 rules keyed by MARC field tags holding all the mapping rules for that particular tag.
1018
1019 We can then iterate through all MARC fields for each record and apply all relevant
1020 rules once per fields instead of retreiving fields multiple times for each mapping rule
1021 which is terribly slow.
1022
1023 =cut
1024
1025 # TODO: This structure can be used for processing multiple MARC::Records so is currently
1026 # rebuilt for each batch. Since it is cacheable it could also be stored in an in
1027 # memory cache which it is currently not. The performance gain of caching
1028 # would probably be marginal, but to do this could be a further improvement.
1029
1030 sub _get_marc_mapping_rules {
1031     my ($self) = @_;
1032     my $marcflavour = lc C4::Context->preference('marcflavour');
1033     my $field_spec_regexp = qr/^([0-9]{3})([()0-9a-zA-Z]+)?(?:_\/(\d+(?:-\d+)?))?$/;
1034     my $leader_regexp = qr/^leader(?:_\/(\d+(?:-\d+)?))?$/;
1035     my $rules = {
1036         'leader' => [],
1037         'control_fields' => {},
1038         'data_fields' => {},
1039         'sum' => [],
1040         'isbn' => [],
1041         'defaults' => {}
1042     };
1043
1044     $self->_foreach_mapping(sub {
1045         my ($name, $type, $facet, $suggestible, $sort, $search, $marc_type, $marc_field) = @_;
1046         return if $marc_type ne $marcflavour;
1047
1048         if ($type eq 'sum') {
1049             push @{$rules->{sum}}, $name;
1050             push @{$rules->{sum}}, $name."__sort" if $sort;
1051         }
1052         elsif ($type eq 'isbn') {
1053             push @{$rules->{isbn}}, $name;
1054         }
1055         elsif ($type eq 'boolean') {
1056             # boolean gets special handling, if value doesn't exist for a field,
1057             # it is set to false
1058             $rules->{defaults}->{$name} = 'false';
1059         }
1060
1061         if ($marc_field =~ $field_spec_regexp) {
1062             my $field_tag = $1;
1063
1064             my @subfields;
1065             my @subfield_groups;
1066             # Parse and separate subfields form subfield groups
1067             if (defined $2) {
1068                 my $subfield_group = '';
1069                 my $open_group = 0;
1070
1071                 foreach my $token (split //, $2) {
1072                     if ($token eq "(") {
1073                         if ($open_group) {
1074                             Koha::Exceptions::Elasticsearch::MARCFieldExprParseError->throw(
1075                                 "Unmatched opening parenthesis for $marc_field"
1076                             );
1077                         }
1078                         else {
1079                             $open_group = 1;
1080                         }
1081                     }
1082                     elsif ($token eq ")") {
1083                         if ($open_group) {
1084                             if ($subfield_group) {
1085                                 push @subfield_groups, $subfield_group;
1086                                 $subfield_group = '';
1087                             }
1088                             $open_group = 0;
1089                         }
1090                         else {
1091                             Koha::Exceptions::Elasticsearch::MARCFieldExprParseError->throw(
1092                                 "Unmatched closing parenthesis for $marc_field"
1093                             );
1094                         }
1095                     }
1096                     elsif ($open_group) {
1097                         $subfield_group .= $token;
1098                     }
1099                     else {
1100                         push @subfields, $token;
1101                     }
1102                 }
1103             }
1104             else {
1105                 push @subfields, '*';
1106             }
1107
1108             my $range = defined $3 ? $3 : undef;
1109             my @mappings = $self->_field_mappings($facet, $suggestible, $sort, $search, $name, $type, $range);
1110             if ($field_tag < 10) {
1111                 $rules->{control_fields}->{$field_tag} //= [];
1112                 push @{$rules->{control_fields}->{$field_tag}}, @mappings;
1113             }
1114             else {
1115                 $rules->{data_fields}->{$field_tag} //= {};
1116                 foreach my $subfield (@subfields) {
1117                     $rules->{data_fields}->{$field_tag}->{subfields}->{$subfield} //= [];
1118                     push @{$rules->{data_fields}->{$field_tag}->{subfields}->{$subfield}}, @mappings;
1119                 }
1120                 foreach my $subfield_group (@subfield_groups) {
1121                     $rules->{data_fields}->{$field_tag}->{subfields_join}->{$subfield_group} //= [];
1122                     push @{$rules->{data_fields}->{$field_tag}->{subfields_join}->{$subfield_group}}, @mappings;
1123                 }
1124             }
1125         }
1126         elsif ($marc_field =~ $leader_regexp) {
1127             my $range = defined $1 ? $1 : undef;
1128             my @mappings = $self->_field_mappings($facet, $suggestible, $sort, $search, $name, $type, $range);
1129             push @{$rules->{leader}}, @mappings;
1130         }
1131         else {
1132             Koha::Exceptions::Elasticsearch::MARCFieldExprParseError->throw(
1133                 "Invalid MARC field expression: $marc_field"
1134             );
1135         }
1136     });
1137
1138     # Marc-flavour specific rule tweaks, could/should also provide hook for this
1139     if ($marcflavour eq 'marc21') {
1140         # Nonfiling characters processing for sort fields
1141         my %title_fields;
1142         if ($self->index eq $Koha::SearchEngine::BIBLIOS_INDEX) {
1143             # Format is: nonfiling characters indicator => field names list
1144             %title_fields = (
1145                 1 => [130, 630, 730, 740],
1146                 2 => [222, 240, 242, 243, 245, 440, 830]
1147             );
1148         }
1149         elsif ($self->index eq $Koha::SearchEngine::AUTHORITIES_INDEX) {
1150             %title_fields = (
1151                 1 => [730],
1152                 2 => [130, 430, 530]
1153             );
1154         }
1155         foreach my $indicator (keys %title_fields) {
1156             foreach my $field_tag (@{$title_fields{$indicator}}) {
1157                 my $mappings = $rules->{data_fields}->{$field_tag}->{subfields}->{a} // [];
1158                 foreach my $mapping (@{$mappings}) {
1159                     if ($mapping->[0] =~ /__sort$/) {
1160                         # Mark this as to be processed for nonfiling characters indicator
1161                         # later on in _process_mappings
1162                         $mapping->[1]->{nonfiling_characters_indicator} = $indicator;
1163                     }
1164                 }
1165             }
1166         }
1167     }
1168
1169     return $rules;
1170 }
1171
1172 =head2 _foreach_mapping
1173
1174     $self->_foreach_mapping(
1175         sub {
1176             my ( $name, $type, $facet, $suggestible, $sort, $marc_type,
1177                 $marc_field )
1178               = @_;
1179             return unless $marc_type eq 'marc21';
1180             print "Data comes from: " . $marc_field . "\n";
1181         }
1182     );
1183
1184 This allows you to apply a function to each entry in the elasticsearch mappings
1185 table, in order to build the mappings for whatever is needed.
1186
1187 In the provided function, the files are:
1188
1189 =over 4
1190
1191 =item C<$name>
1192
1193 The field name for elasticsearch (corresponds to the 'mapping' column in the
1194 database.
1195
1196 =item C<$type>
1197
1198 The type for this value, e.g. 'string'.
1199
1200 =item C<$facet>
1201
1202 True if this value should be facetised. This only really makes sense if the
1203 field is understood by the facet processing code anyway.
1204
1205 =item C<$sort>
1206
1207 True if this is a field that a) needs special sort handling, and b) if it
1208 should be sorted on. False if a) but not b). Undef if not a). This allows,
1209 for example, author to be sorted on but not everything marked with "author"
1210 to be included in that sort.
1211
1212 =item C<$marc_type>
1213
1214 A string that indicates the MARC type that this mapping is for, e.g. 'marc21',
1215 'unimarc'.
1216
1217 =item C<$marc_field>
1218
1219 A string that describes the MARC field that contains the data to extract.
1220
1221 =back
1222
1223 =cut
1224
1225 sub _foreach_mapping {
1226     my ( $self, $sub ) = @_;
1227
1228     # TODO use a caching framework here
1229     my $search_fields = Koha::Database->schema->resultset('SearchField')->search(
1230         {
1231             'search_marc_map.index_name' => $self->index,
1232         },
1233         {   join => { search_marc_to_fields => 'search_marc_map' },
1234             '+select' => [
1235                 'search_marc_to_fields.facet',
1236                 'search_marc_to_fields.suggestible',
1237                 'search_marc_to_fields.sort',
1238                 'search_marc_to_fields.search',
1239                 'search_marc_map.marc_type',
1240                 'search_marc_map.marc_field',
1241             ],
1242             '+as'     => [
1243                 'facet',
1244                 'suggestible',
1245                 'sort',
1246                 'search',
1247                 'marc_type',
1248                 'marc_field',
1249             ],
1250         }
1251     );
1252
1253     while ( my $search_field = $search_fields->next ) {
1254         $sub->(
1255             # Force lower case on indexed field names for case insensitive
1256             # field name searches
1257             lc($search_field->name),
1258             $search_field->type,
1259             $search_field->get_column('facet'),
1260             $search_field->get_column('suggestible'),
1261             $search_field->get_column('sort'),
1262             $search_field->get_column('search'),
1263             $search_field->get_column('marc_type'),
1264             $search_field->get_column('marc_field'),
1265         );
1266     }
1267 }
1268
1269 =head2 process_error
1270
1271     die process_error($@);
1272
1273 This parses an Elasticsearch error message and produces a human-readable
1274 result from it. This result is probably missing all the useful information
1275 that you might want in diagnosing an issue, so the warning is also logged.
1276
1277 Note that currently the resulting message is not internationalised. This
1278 will happen eventually by some method or other.
1279
1280 =cut
1281
1282 sub process_error {
1283     my ($self, $msg) = @_;
1284
1285     warn $msg; # simple logging
1286
1287     # This is super-primitive
1288     return "Unable to understand your search query, please rephrase and try again.\n" if $msg =~ /ParseException|parse_exception/;
1289
1290     return "Unable to perform your search. Please try again.\n";
1291 }
1292
1293 =head2 _read_configuration
1294
1295     my $conf = _read_configuration();
1296
1297 Reads the I<configuration file> and returns a hash structure with the
1298 configuration information. It raises an exception if mandatory entries
1299 are missing.
1300
1301 The hashref structure has the following form:
1302
1303     {
1304         'nodes' => ['127.0.0.1:9200', 'anotherserver:9200'],
1305         'index_name' => 'koha_instance',
1306     }
1307
1308 This is configured by the following in the C<config> block in koha-conf.xml:
1309
1310     <elasticsearch>
1311         <server>127.0.0.1:9200</server>
1312         <server>anotherserver:9200</server>
1313         <index_name>koha_instance</index_name>
1314     </elasticsearch>
1315
1316 =cut
1317
1318 sub _read_configuration {
1319
1320     my $configuration;
1321
1322     my $conf = C4::Context->config('elasticsearch');
1323     unless ( defined $conf ) {
1324         Koha::Exceptions::Config::MissingEntry->throw(
1325             "Missing <elasticsearch> entry in koha-conf.xml"
1326         );
1327     }
1328
1329     unless ( exists $conf->{server} ) {
1330         Koha::Exceptions::Config::MissingEntry->throw(
1331             "Missing <elasticsearch>/<server> entry in koha-conf.xml"
1332         );
1333     }
1334
1335     unless ( exists $conf->{index_name} ) {
1336         Koha::Exceptions::Config::MissingEntry->throw(
1337             "Missing <elasticsearch>/<index_name> entry in koha-conf.xml",
1338         );
1339     }
1340
1341     while ( my ( $var, $val ) = each %$conf ) {
1342         if ( $var eq 'server' ) {
1343             if ( ref($val) eq 'ARRAY' ) {
1344                 $configuration->{nodes} = $val;
1345             }
1346             else {
1347                 $configuration->{nodes} = [$val];
1348             }
1349         } else {
1350             $configuration->{$var} = $val;
1351         }
1352     }
1353
1354     $configuration->{cxn_pool} //= 'Static';
1355
1356     return $configuration;
1357 }
1358
1359 =head2 get_facetable_fields
1360
1361 my @facetable_fields = Koha::SearchEngine::Elasticsearch->get_facetable_fields();
1362
1363 Returns the list of Koha::SearchFields marked to be faceted in the ES configuration
1364
1365 =cut
1366
1367 sub get_facetable_fields {
1368     my ($self) = @_;
1369
1370     # These should correspond to the ES field names, as opposed to the CCL
1371     # things that zebra uses.
1372     my @search_field_names = qw( author itype location su-geo title-series subject ccode holdingbranch homebranch ln );
1373     my @faceted_fields = Koha::SearchFields->search(
1374         { name => { -in => \@search_field_names }, facet_order => { '!=' => undef } }, { order_by => ['facet_order'] }
1375     )->as_list;
1376     my @not_faceted_fields = Koha::SearchFields->search(
1377         { name => { -in => \@search_field_names }, facet_order => undef }, { order_by => ['facet_order'] }
1378     )->as_list;
1379     # This could certainly be improved
1380     return ( @faceted_fields, @not_faceted_fields );
1381 }
1382
1383 =head2 clear_search_fields_cache
1384
1385 Koha::SearchEngine::Elasticsearch->clear_search_fields_cache();
1386
1387 Clear cached values for ES search fields
1388
1389 =cut
1390
1391 sub clear_search_fields_cache {
1392
1393     my $cache = Koha::Caches->get_instance();
1394     $cache->clear_from_cache('elasticsearch_search_fields_staff_client_biblios');
1395     $cache->clear_from_cache('elasticsearch_search_fields_opac_biblios');
1396     $cache->clear_from_cache('elasticsearch_search_fields_staff_client_authorities');
1397     $cache->clear_from_cache('elasticsearch_search_fields_opac_authorities');
1398
1399 }
1400
1401 1;
1402
1403 __END__
1404
1405 =head1 AUTHOR
1406
1407 =over 4
1408
1409 =item Chris Cormack C<< <chrisc@catalyst.net.nz> >>
1410
1411 =item Robin Sheat C<< <robin@catalyst.net.nz> >>
1412
1413 =item Jonathan Druart C<< <jonathan.druart@bugs.koha-community.org> >>
1414
1415 =back
1416
1417 =cut