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