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             if ( $nonfiling_chars > 0 ) {
 515                 if ($sort) {
 516                     $values->[0] = substr $values->[0], $nonfiling_chars;
 517                 } else {
 518                     push @{$values}, substr $values->[0], $nonfiling_chars;
 519                 }
 520             }
 521         }
 522
 523         $values = [ grep(!/^$/, @{$values}) ];
 524
 525         $record_document->{$target} //= [];
 526         push @{$record_document->{$target}}, @{$values};
 527     }
 528 }
 529
 530 =head2 marc_records_to_documents($marc_records)
 531
 532     my $record_documents = $self->marc_records_to_documents($marc_records);
 533
 534 Using mappings stored in database convert C<$marc_records> to Elasticsearch documents.
 535
 536 Returns array of hash references, representing Elasticsearch documents,
 537 acceptable as body payload in C<Search::Elasticsearch> requests.
 538
 539 =over 4
 540
 541 =item C<$marc_documents>
 542
 543 Reference to array of C<MARC::Record> objects to be converted to Elasticsearch documents.
 544
 545 =back
 546
 547 =cut
 548
 549 sub marc_records_to_documents {
 550     my ($self, $records) = @_;
 551     my $rules = $self->_get_marc_mapping_rules();
 552     my $control_fields_rules = $rules->{control_fields};
 553     my $data_fields_rules = $rules->{data_fields};
 554     my $marcflavour = lc C4::Context->preference('marcflavour');
 555     my $use_array = C4::Context->preference('ElasticsearchMARCFormat') eq 'ARRAY';
 556
 557     my @record_documents;
 558
 559     my %auth_match_headings;
 560     if( $self->index eq 'authorities' ){
 561         my @auth_types = Koha::Authority::Types->search->as_list;
 562         %auth_match_headings = map { $_->authtypecode => $_->auth_tag_to_report } @auth_types;
 563     }
 564
 565     foreach my $record (@{$records}) {
 566         my $record_document = {};
 567
 568         if ( $self->index eq 'authorities' ){
 569             my $authtypecode = GuessAuthTypeCode( $record );
 570             if( $authtypecode ){
 571                 if( $authtypecode !~ m/_SUBD/ ){ #Subdivision records will not be used for linking and so don't require match-heading to be built
 572                     my $field = $record->field( $auth_match_headings{ $authtypecode } );
 573                     my $heading = C4::Heading->new_from_field( $field, undef, 1 ); #new auth heading
 574                     push @{$record_document->{'match-heading'}}, $heading->search_form if $heading;
 575                 }
 576             } else {
 577                 warn "Cannot determine authority type for record: " . $record->field('001')->as_string;
 578             }
 579         }
 580
 581         my $mappings = $rules->{leader};
 582         if ($mappings) {
 583             $self->_process_mappings($mappings, $record->leader(), $record_document, {
 584                     altscript => 0,
 585                     data_source => 'leader'
 586                 }
 587             );
 588         }
 589         foreach my $field ($record->fields()) {
 590             if ($field->is_control_field()) {
 591                 my $mappings = $control_fields_rules->{$field->tag()};
 592                 if ($mappings) {
 593                     $self->_process_mappings($mappings, $field->data(), $record_document, {
 594                             altscript => 0,
 595                             data_source => 'control_field',
 596                             field => $field
 597                         }
 598                     );
 599                 }
 600             }
 601             else {
 602                 my $tag = $field->tag();
 603                 # Handle alternate scripts in MARC 21
 604                 my $altscript = 0;
 605                 if ($marcflavour eq 'marc21' && $tag eq '880') {
 606                     my $sub6 = $field->subfield('6');
 607                     if ($sub6 =~ /^(...)-\d+/) {
 608                         $tag = $1;
 609                         $altscript = 1;
 610                     }
 611                 }
 612
 613                 my $data_field_rules = $data_fields_rules->{$tag};
 614                 if ($data_field_rules) {
 615                     my $subfields_mappings = $data_field_rules->{subfields};
 616                     my $wildcard_mappings = $subfields_mappings->{'*'};
 617                     foreach my $subfield ($field->subfields()) {
 618                         my ($code, $data) = @{$subfield};
 619                         my $mappings = $subfields_mappings->{$code} // [];
 620                         if ($wildcard_mappings) {
 621                             $mappings = [@{$mappings}, @{$wildcard_mappings}];
 622                         }
 623                         if (@{$mappings}) {
 624                             $self->_process_mappings($mappings, $data, $record_document, {
 625                                     altscript => $altscript,
 626                                     data_source => 'subfield',
 627                                     code => $code,
 628                                     field => $field
 629                                 }
 630                             );
 631                         }
 632                     }
 633
 634                     my $subfields_join_mappings = $data_field_rules->{subfields_join};
 635                     if ($subfields_join_mappings) {
 636                         foreach my $subfields_group (keys %{$subfields_join_mappings}) {
 637                             my $data_field = $field->clone; #copy field to preserve for alt scripts
 638                             $data_field->delete_subfield(match => qr/^$/); #remove empty subfields, otherwise they are printed as a space
 639                             my $data = $data_field->as_string( $subfields_group ); #get values for subfields as a combined string, preserving record order
 640                             if ($data) {
 641                                 $self->_process_mappings($subfields_join_mappings->{$subfields_group}, $data, $record_document, {
 642                                         altscript => $altscript,
 643                                         data_source => 'subfields_group',
 644                                         codes => $subfields_group,
 645                                         field => $field
 646                                     }
 647                                 );
 648                             }
 649                         }
 650                     }
 651                 }
 652             }
 653         }
 654
 655         if (C4::Context->preference('IncludeSeeFromInSearches') and $self->index eq 'biblios') {
 656             foreach my $field (Koha::Filter::MARC::EmbedSeeFromHeadings->new->fields($record)) {
 657                 my $data_field_rules = $data_fields_rules->{$field->tag()};
 658                 if ($data_field_rules) {
 659                     my $subfields_mappings = $data_field_rules->{subfields};
 660                     my $wildcard_mappings = $subfields_mappings->{'*'};
 661                     foreach my $subfield ($field->subfields()) {
 662                         my ($code, $data) = @{$subfield};
 663                         my @mappings;
 664                         push @mappings, @{ $subfields_mappings->{$code} } if $subfields_mappings->{$code};
 665                         push @mappings, @$wildcard_mappings if $wildcard_mappings;
 666                         # Do not include "see from" into these kind of fields
 667                         @mappings = grep { $_->[0] !~ /__(sort|facet|suggestion)$/ } @mappings;
 668                         if (@mappings) {
 669                             $self->_process_mappings(\@mappings, $data, $record_document, {
 670                                     data_source => 'subfield',
 671                                     code => $code,
 672                                     field => $field
 673                                 }
 674                             );
 675                         }
 676                     }
 677
 678                     my $subfields_join_mappings = $data_field_rules->{subfields_join};
 679                     if ($subfields_join_mappings) {
 680                         foreach my $subfields_group (keys %{$subfields_join_mappings}) {
 681                             my $data_field = $field->clone;
 682                             # remove empty subfields, otherwise they are printed as a space
 683                             $data_field->delete_subfield(match => qr/^$/);
 684                             my $data = $data_field->as_string( $subfields_group );
 685                             if ($data) {
 686                                 my @mappings = @{ $subfields_join_mappings->{$subfields_group} };
 687                                 # Do not include "see from" into these kind of fields
 688                                 @mappings = grep { $_->[0] !~ /__(sort|facet|suggestion)$/ } @mappings;
 689                                 $self->_process_mappings(\@mappings, $data, $record_document, {
 690                                         data_source => 'subfields_group',
 691                                         codes => $subfields_group,
 692                                         field => $field
 693                                     }
 694                                 );
 695                             }
 696                         }
 697                     }
 698                 }
 699             }
 700         }
 701
 702         foreach my $field (keys %{$rules->{defaults}}) {
 703             unless (defined $record_document->{$field}) {
 704                 $record_document->{$field} = $rules->{defaults}->{$field};
 705             }
 706         }
 707         foreach my $field (@{$rules->{sum}}) {
 708             if (defined $record_document->{$field}) {
 709                 # TODO: validate numeric? filter?
 710                 # TODO: Or should only accept fields without nested values?
 711                 # TODO: Quick and dirty, improve if needed
 712                 $record_document->{$field} = sum0(grep { !ref($_) && m/\d+(\.\d+)?/} @{$record_document->{$field}});
 713             }
 714         }
 715         # Index all applicable ISBN forms (ISBN-10 and ISBN-13 with and without dashes)
 716         foreach my $field (@{$rules->{isbn}}) {
 717             if (defined $record_document->{$field}) {
 718                 my @isbns = ();
 719                 foreach my $input_isbn (@{$record_document->{$field}}) {
 720                     my $isbn = Business::ISBN->new($input_isbn);
 721                     if (defined $isbn && $isbn->is_valid) {
 722                         my $isbn13 = $isbn->as_isbn13->as_string;
 723                         push @isbns, $isbn13;
 724                         $isbn13 =~ s/\-//g;
 725                         push @isbns, $isbn13;
 726
 727                         my $isbn10 = $isbn->as_isbn10;
 728                         if ($isbn10) {
 729                             $isbn10 = $isbn10->as_string;
 730                             push @isbns, $isbn10;
 731                             $isbn10 =~ s/\-//g;
 732                             push @isbns, $isbn10;
 733                         }
 734                     } else {
 735                         push @isbns, $input_isbn;
 736                     }
 737                 }
 738                 $record_document->{$field} = \@isbns;
 739             }
 740         }
 741
 742         # Remove duplicate values and collapse sort fields
 743         foreach my $field (keys %{$record_document}) {
 744             if (ref($record_document->{$field}) eq 'ARRAY') {
 745                 @{$record_document->{$field}} = do {
 746                     my %seen;
 747                     grep { !$seen{ref($_) eq 'HASH' && defined $_->{input} ? $_->{input} : $_}++ } @{$record_document->{$field}};
 748                 };
 749                 if ($field =~ /__sort$/) {
 750                     # Make sure to keep the sort field length sensible. 255 was chosen as a nice round value.
 751                     $record_document->{$field} = [substr(join(' ', @{$record_document->{$field}}), 0, 255)];
 752                 }
 753             }
 754         }
 755
 756         # TODO: Perhaps should check if $records_document non empty, but really should never be the case
 757         $record->encoding('UTF-8');
 758         if ($use_array) {
 759             $record_document->{'marc_data_array'} = $self->_marc_to_array($record);
 760             $record_document->{'marc_format'} = 'ARRAY';
 761         } else {
 762             my @warnings;
 763             {
 764                 # Temporarily intercept all warn signals (MARC::Record carps when record length > 99999)
 765                 local $SIG{__WARN__} = sub {
 766                     push @warnings, $_[0];
 767                 };
 768                 $record_document->{'marc_data'} = encode_base64(encode('UTF-8', $record->as_usmarc()));
 769             }
 770             if (@warnings) {
 771                 # Suppress warnings if record length exceeded
 772                 unless (substr($record->leader(), 0, 5) eq '99999') {
 773                     foreach my $warning (@warnings) {
 774                         carp $warning;
 775                     }
 776                 }
 777                 $record_document->{'marc_data'} = $record->as_xml_record($marcflavour);
 778                 $record_document->{'marc_format'} = 'MARCXML';
 779             }
 780             else {
 781                 $record_document->{'marc_format'} = 'base64ISO2709';
 782             }
 783         }
 784
 785         # Check if there is at least one available item
 786         if ($self->index eq $BIBLIOS_INDEX) {
 787             my ($tag, $code) = C4::Biblio::GetMarcFromKohaField('biblio.biblionumber');
 788             my $field = $record->field($tag);
 789             if ($field) {
 790                 my $biblionumber = $field->is_control_field ? $field->data : $field->subfield($code);
 791                 my $avail_items = Koha::Items->search({
 792                     biblionumber => $biblionumber,
 793                     onloan       => undef,
 794                     itemlost     => 0,
 795                 })->count;
 796
 797                 $record_document->{available} = $avail_items ? \1 : \0;
 798             }
 799         }
 800
 801         push @record_documents, $record_document;
 802     }
 803     return \@record_documents;
 804 }
 805
 806 =head2 _marc_to_array($record)
 807
 808     my @fields = _marc_to_array($record)
 809
 810 Convert a MARC::Record to an array modeled after MARC-in-JSON
 811 (see https://github.com/marc4j/marc4j/wiki/MARC-in-JSON-Description)
 812
 813 =over 4
 814
 815 =item C<$record>
 816
 817 A MARC::Record object
 818
 819 =back
 820
 821 =cut
 822
 823 sub _marc_to_array {
 824     my ($self, $record) = @_;
 825
 826     my $data = {
 827         leader => $record->leader(),
 828         fields => []
 829     };
 830     for my $field ($record->fields()) {
 831         my $tag = $field->tag();
 832         if ($field->is_control_field()) {
 833             push @{$data->{fields}}, {$tag => $field->data()};
 834         } else {
 835             my $subfields = ();
 836             foreach my $subfield ($field->subfields()) {
 837                 my ($code, $contents) = @{$subfield};
 838                 push @{$subfields}, {$code => $contents};
 839             }
 840             push @{$data->{fields}}, {
 841                 $tag => {
 842                     ind1 => $field->indicator(1),
 843                     ind2 => $field->indicator(2),
 844                     subfields => $subfields
 845                 }
 846             };
 847         }
 848     }
 849     return $data;
 850 }
 851
 852 =head2 _array_to_marc($data)
 853
 854     my $record = _array_to_marc($data)
 855
 856 Convert an array modeled after MARC-in-JSON to a MARC::Record
 857
 858 =over 4
 859
 860 =item C<$data>
 861
 862 An array modeled after MARC-in-JSON
 863 (see https://github.com/marc4j/marc4j/wiki/MARC-in-JSON-Description)
 864
 865 =back
 866
 867 =cut
 868
 869 sub _array_to_marc {
 870     my ($self, $data) = @_;
 871
 872     my $record = MARC::Record->new();
 873
 874     $record->leader($data->{leader});
 875     for my $field (@{$data->{fields}}) {
 876         my $tag = (keys %{$field})[0];
 877         $field = $field->{$tag};
 878         my $marc_field;
 879         if (ref($field) eq 'HASH') {
 880             my @subfields;
 881             foreach my $subfield (@{$field->{subfields}}) {
 882                 my $code = (keys %{$subfield})[0];
 883                 push @subfields, $code;
 884                 push @subfields, $subfield->{$code};
 885             }
 886             $marc_field = MARC::Field->new($tag, $field->{ind1}, $field->{ind2}, @subfields);
 887         } else {
 888             $marc_field = MARC::Field->new($tag, $field)
 889         }
 890         $record->append_fields($marc_field);
 891     }
 892 ;
 893     return $record;
 894 }
 895
 896 =head2 _field_mappings($facet, $suggestible, $sort, $search, $target_name, $target_type, $range)
 897
 898     my @mappings = _field_mappings($facet, $suggestible, $sort, $search, $target_name, $target_type, $range)
 899
 900 Get mappings, an internal data structure later used by
 901 L<_process_mappings($mappings, $data, $record_document, $meta)> to process MARC target
 902 data for a MARC mapping.
 903
 904 The returned C<$mappings> is not to to be confused with mappings provided by
 905 C<_foreach_mapping>, rather this sub accepts properties from a mapping as
 906 provided by C<_foreach_mapping> and expands it to this internal data structure.
 907 In the caller context (C<_get_marc_mapping_rules>) the returned C<@mappings>
 908 is then applied to each MARC target (leader, control field data, subfield or
 909 joined subfields) and integrated into the mapping rules data structure used in
 910 C<marc_records_to_documents> to transform MARC records into Elasticsearch
 911 documents.
 912
 913 =over 4
 914
 915 =item C<$facet>
 916
 917 Boolean indicating whether to create a facet field for this mapping.
 918
 919 =item C<$suggestible>
 920
 921 Boolean indicating whether to create a suggestion field for this mapping.
 922
 923 =item C<$sort>
 924
 925 Boolean indicating whether to create a sort field for this mapping.
 926
 927 =item C<$search>
 928
 929 Boolean indicating whether to create a search field for this mapping.
 930
 931 =item C<$target_name>
 932
 933 Elasticsearch document target field name.
 934
 935 =item C<$target_type>
 936
 937 Elasticsearch document target field type.
 938
 939 =item C<$range>
 940
 941 An optional range as a string in the format "<START>-<END>" or "<START>",
 942 where "<START>" and "<END>" are integers specifying a range that will be used
 943 for extracting a substring from MARC data as Elasticsearch field target value.
 944
 945 The first character position is "0", and the range is inclusive,
 946 so "0-2" means the first three characters of MARC data.
 947
 948 If only "<START>" is provided only one character at position "<START>" will
 949 be extracted.
 950
 951 =back
 952
 953 =cut
 954
 955 sub _field_mappings {
 956     my ($_self, $facet, $suggestible, $sort, $search, $target_name, $target_type, $range) = @_;
 957     my %mapping_defaults = ();
 958     my @mappings;
 959
 960     my $substr_args = undef;
 961     if (defined $range) {
 962         # TODO: use value_callback instead?
 963         my ($start, $end) = map(int, split /-/, $range, 2);
 964         $substr_args = [$start];
 965         push @{$substr_args}, (defined $end ? $end - $start + 1 : 1);
 966     }
 967     my $default_options = {};
 968     if ($substr_args) {
 969         $default_options->{substr} = $substr_args;
 970     }
 971
 972     # TODO: Should probably have per type value callback/hook
 973     # but hard code for now
 974     if ($target_type eq 'boolean') {
 975         $default_options->{value_callbacks} //= [];
 976         push @{$default_options->{value_callbacks}}, sub {
 977             my ($value) = @_;
 978             # Trim whitespace at both ends
 979             $value =~ s/^\s+|\s+$//g;
 980             return $value ? 'true' : 'false';
 981         };
 982     }
 983     elsif ($target_type eq 'year') {
 984         $default_options->{value_callbacks} //= [];
 985         # Only accept years containing digits and "u"
 986         push @{$default_options->{value_callbacks}}, sub {
 987             my ($value) = @_;
 988             # Replace "u" with "0" for sorting
 989             return map { s/[u\s]/0/gr } ( $value =~ /[0-9u\s]{4}/g );
 990         };
 991     }
 992
 993     if ($search) {
 994         my $mapping = [$target_name, $default_options];
 995         push @mappings, $mapping;
 996     }
 997
 998     my @suffixes = ();
 999     push @suffixes, 'facet' if $facet;
1000     push @suffixes, 'suggestion' if $suggestible;
1001     push @suffixes, 'sort' if !defined $sort || $sort;
1002
1003     foreach my $suffix (@suffixes) {
1004         my $mapping = ["${target_name}__$suffix"];
1005         # TODO: Hack, fix later in less hideous manner
1006         if ($suffix eq 'suggestion') {
1007             push @{$mapping}, {%{$default_options}, property => 'input'};
1008         }
1009         else {
1010             # Important! Make shallow clone, or we end up with the same hashref
1011             # shared by all mappings
1012             push @{$mapping}, {%{$default_options}};
1013         }
1014         push @mappings, $mapping;
1015     }
1016     return @mappings;
1017 };
1018
1019 =head2 _get_marc_mapping_rules
1020
1021     my $mapping_rules = $self->_get_marc_mapping_rules()
1022
1023 Generates rules from mappings stored in database for MARC records to Elasticsearch JSON document conversion.
1024
1025 Since field retrieval is slow in C<MARC::Records> (all fields are itereted through for
1026 each call to C<MARC::Record>->field) we create an optimized structure of mapping
1027 rules keyed by MARC field tags holding all the mapping rules for that particular tag.
1028
1029 We can then iterate through all MARC fields for each record and apply all relevant
1030 rules once per fields instead of retreiving fields multiple times for each mapping rule
1031 which is terribly slow.
1032
1033 =cut
1034
1035 # TODO: This structure can be used for processing multiple MARC::Records so is currently
1036 # rebuilt for each batch. Since it is cacheable it could also be stored in an in
1037 # memory cache which it is currently not. The performance gain of caching
1038 # would probably be marginal, but to do this could be a further improvement.
1039
1040 sub _get_marc_mapping_rules {
1041     my ($self) = @_;
1042     my $marcflavour = lc C4::Context->preference('marcflavour');
1043     my $field_spec_regexp = qr/^([0-9]{3})([()0-9a-zA-Z]+)?(?:_\/(\d+(?:-\d+)?))?$/;
1044     my $leader_regexp = qr/^leader(?:_\/(\d+(?:-\d+)?))?$/;
1045     my $rules = {
1046         'leader' => [],
1047         'control_fields' => {},
1048         'data_fields' => {},
1049         'sum' => [],
1050         'isbn' => [],
1051         'defaults' => {}
1052     };
1053
1054     $self->_foreach_mapping(sub {
1055         my ($name, $type, $facet, $suggestible, $sort, $search, $marc_type, $marc_field) = @_;
1056         return if $marc_type ne $marcflavour;
1057
1058         if ($type eq 'sum') {
1059             push @{$rules->{sum}}, $name;
1060             push @{$rules->{sum}}, $name."__sort" if $sort;
1061         }
1062         elsif ($type eq 'isbn') {
1063             push @{$rules->{isbn}}, $name;
1064         }
1065         elsif ($type eq 'boolean') {
1066             # boolean gets special handling, if value doesn't exist for a field,
1067             # it is set to false
1068             $rules->{defaults}->{$name} = 'false';
1069         }
1070
1071         if ($marc_field =~ $field_spec_regexp) {
1072             my $field_tag = $1;
1073
1074             my @subfields;
1075             my @subfield_groups;
1076             # Parse and separate subfields form subfield groups
1077             if (defined $2) {
1078                 my $subfield_group = '';
1079                 my $open_group = 0;
1080
1081                 foreach my $token (split //, $2) {
1082                     if ($token eq "(") {
1083                         if ($open_group) {
1084                             Koha::Exceptions::Elasticsearch::MARCFieldExprParseError->throw(
1085                                 "Unmatched opening parenthesis for $marc_field"
1086                             );
1087                         }
1088                         else {
1089                             $open_group = 1;
1090                         }
1091                     }
1092                     elsif ($token eq ")") {
1093                         if ($open_group) {
1094                             if ($subfield_group) {
1095                                 push @subfield_groups, $subfield_group;
1096                                 $subfield_group = '';
1097                             }
1098                             $open_group = 0;
1099                         }
1100                         else {
1101                             Koha::Exceptions::Elasticsearch::MARCFieldExprParseError->throw(
1102                                 "Unmatched closing parenthesis for $marc_field"
1103                             );
1104                         }
1105                     }
1106                     elsif ($open_group) {
1107                         $subfield_group .= $token;
1108                     }
1109                     else {
1110                         push @subfields, $token;
1111                     }
1112                 }
1113             }
1114             else {
1115                 push @subfields, '*';
1116             }
1117
1118             my $range = defined $3 ? $3 : undef;
1119             my @mappings = $self->_field_mappings($facet, $suggestible, $sort, $search, $name, $type, $range);
1120             if ($field_tag < 10) {
1121                 $rules->{control_fields}->{$field_tag} //= [];
1122                 push @{$rules->{control_fields}->{$field_tag}}, @{clone(\@mappings)};
1123             }
1124             else {
1125                 $rules->{data_fields}->{$field_tag} //= {};
1126                 foreach my $subfield (@subfields) {
1127                     $rules->{data_fields}->{$field_tag}->{subfields}->{$subfield} //= [];
1128                     push @{$rules->{data_fields}->{$field_tag}->{subfields}->{$subfield}}, @{clone(\@mappings)};
1129                 }
1130                 foreach my $subfield_group (@subfield_groups) {
1131                     $rules->{data_fields}->{$field_tag}->{subfields_join}->{$subfield_group} //= [];
1132                     push @{$rules->{data_fields}->{$field_tag}->{subfields_join}->{$subfield_group}}, @{clone(\@mappings)};
1133                 }
1134             }
1135         }
1136         elsif ($marc_field =~ $leader_regexp) {
1137             my $range = defined $1 ? $1 : undef;
1138             my @mappings = $self->_field_mappings($facet, $suggestible, $sort, $search, $name, $type, $range);
1139             push @{$rules->{leader}}, @{clone(\@mappings)};
1140         }
1141         else {
1142             Koha::Exceptions::Elasticsearch::MARCFieldExprParseError->throw(
1143                 "Invalid MARC field expression: $marc_field"
1144             );
1145         }
1146     });
1147
1148     # Marc-flavour specific rule tweaks, could/should also provide hook for this
1149     if ($marcflavour eq 'marc21') {
1150         # Nonfiling characters processing for sort fields
1151         my %title_fields;
1152         if ($self->index eq $Koha::SearchEngine::BIBLIOS_INDEX) {
1153             # Format is: nonfiling characters indicator => field names list
1154             %title_fields = (
1155                 1 => [130, 630, 730, 740],
1156                 2 => [222, 240, 242, 243, 245, 440, 830]
1157             );
1158         }
1159         elsif ($self->index eq $Koha::SearchEngine::AUTHORITIES_INDEX) {
1160             %title_fields = (
1161                 1 => [730],
1162                 2 => [130, 430, 530]
1163             );
1164         }
1165         foreach my $indicator (keys %title_fields) {
1166             foreach my $field_tag (@{$title_fields{$indicator}}) {
1167                 my $mappings = $rules->{data_fields}->{$field_tag}->{subfields}->{a} // [];
1168                 foreach my $mapping ( @{$mappings} ) {
1169                     # Mark this as to be processed for nonfiling characters indicator
1170                     # later on in _process_mappings
1171                     $mapping->[1]->{nonfiling_characters_indicator} = $indicator;
1172                 }
1173             }
1174         }
1175     }
1176
1177     if( $self->index eq 'authorities' ){
1178         push @{$rules->{control_fields}->{'008'}}, ['subject-heading-thesaurus', { 'substr' => [ 11, 1 ] } ];
1179         push @{$rules->{data_fields}->{'040'}->{subfields}->{f}}, ['subject-heading-thesaurus', { } ];
1180     }
1181
1182     return $rules;
1183 }
1184
1185 =head2 _foreach_mapping
1186
1187     $self->_foreach_mapping(
1188         sub {
1189             my ( $name, $type, $facet, $suggestible, $sort, $marc_type,
1190                 $marc_field )
1191               = @_;
1192             return unless $marc_type eq 'marc21';
1193             print "Data comes from: " . $marc_field . "\n";
1194         }
1195     );
1196
1197 This allows you to apply a function to each entry in the elasticsearch mappings
1198 table, in order to build the mappings for whatever is needed.
1199
1200 In the provided function, the files are:
1201
1202 =over 4
1203
1204 =item C<$name>
1205
1206 The field name for elasticsearch (corresponds to the 'mapping' column in the
1207 database.
1208
1209 =item C<$type>
1210
1211 The type for this value, e.g. 'string'.
1212
1213 =item C<$facet>
1214
1215 True if this value should be facetised. This only really makes sense if the
1216 field is understood by the facet processing code anyway.
1217
1218 =item C<$sort>
1219
1220 True if this is a field that a) needs special sort handling, and b) if it
1221 should be sorted on. False if a) but not b). Undef if not a). This allows,
1222 for example, author to be sorted on but not everything marked with "author"
1223 to be included in that sort.
1224
1225 =item C<$marc_type>
1226
1227 A string that indicates the MARC type that this mapping is for, e.g. 'marc21',
1228 'unimarc'.
1229
1230 =item C<$marc_field>
1231
1232 A string that describes the MARC field that contains the data to extract.
1233
1234 =back
1235
1236 =cut
1237
1238 sub _foreach_mapping {
1239     my ( $self, $sub ) = @_;
1240
1241     # TODO use a caching framework here
1242     my $search_fields = Koha::Database->schema->resultset('SearchField')->search(
1243         {
1244             'search_marc_map.index_name' => $self->index,
1245         },
1246         {   join => { search_marc_to_fields => 'search_marc_map' },
1247             '+select' => [
1248                 'search_marc_to_fields.facet',
1249                 'search_marc_to_fields.suggestible',
1250                 'search_marc_to_fields.sort',
1251                 'search_marc_to_fields.search',
1252                 'search_marc_map.marc_type',
1253                 'search_marc_map.marc_field',
1254             ],
1255             '+as'     => [
1256                 'facet',
1257                 'suggestible',
1258                 'sort',
1259                 'search',
1260                 'marc_type',
1261                 'marc_field',
1262             ],
1263         }
1264     );
1265
1266     while ( my $search_field = $search_fields->next ) {
1267         $sub->(
1268             # Force lower case on indexed field names for case insensitive
1269             # field name searches
1270             lc($search_field->name),
1271             $search_field->type,
1272             $search_field->get_column('facet'),
1273             $search_field->get_column('suggestible'),
1274             $search_field->get_column('sort'),
1275             $search_field->get_column('search'),
1276             $search_field->get_column('marc_type'),
1277             $search_field->get_column('marc_field'),
1278         );
1279     }
1280 }
1281
1282 =head2 process_error
1283
1284     die process_error($@);
1285
1286 This parses an Elasticsearch error message and produces a human-readable
1287 result from it. This result is probably missing all the useful information
1288 that you might want in diagnosing an issue, so the warning is also logged.
1289
1290 Note that currently the resulting message is not internationalised. This
1291 will happen eventually by some method or other.
1292
1293 =cut
1294
1295 sub process_error {
1296     my ($self, $msg) = @_;
1297
1298     warn $msg; # simple logging
1299
1300     # This is super-primitive
1301     return "Unable to understand your search query, please rephrase and try again.\n" if $msg =~ /ParseException|parse_exception/;
1302
1303     return "Unable to perform your search. Please try again.\n";
1304 }
1305
1306 =head2 _read_configuration
1307
1308     my $conf = _read_configuration();
1309
1310 Reads the I<configuration file> and returns a hash structure with the
1311 configuration information. It raises an exception if mandatory entries
1312 are missing.
1313
1314 The hashref structure has the following form:
1315
1316     {
1317         'nodes' => ['127.0.0.1:9200', 'anotherserver:9200'],
1318         'index_name' => 'koha_instance',
1319     }
1320
1321 This is configured by the following in the C<config> block in koha-conf.xml:
1322
1323     <elasticsearch>
1324         <server>127.0.0.1:9200</server>
1325         <server>anotherserver:9200</server>
1326         <index_name>koha_instance</index_name>
1327     </elasticsearch>
1328
1329 =cut
1330
1331 sub _read_configuration {
1332
1333     my $configuration;
1334
1335     my $conf = C4::Context->config('elasticsearch');
1336     unless ( defined $conf ) {
1337         Koha::Exceptions::Config::MissingEntry->throw(
1338             "Missing <elasticsearch> entry in koha-conf.xml"
1339         );
1340     }
1341
1342     unless ( exists $conf->{server} ) {
1343         Koha::Exceptions::Config::MissingEntry->throw(
1344             "Missing <elasticsearch>/<server> entry in koha-conf.xml"
1345         );
1346     }
1347
1348     unless ( exists $conf->{index_name} ) {
1349         Koha::Exceptions::Config::MissingEntry->throw(
1350             "Missing <elasticsearch>/<index_name> entry in koha-conf.xml",
1351         );
1352     }
1353
1354     while ( my ( $var, $val ) = each %$conf ) {
1355         if ( $var eq 'server' ) {
1356             if ( ref($val) eq 'ARRAY' ) {
1357                 $configuration->{nodes} = $val;
1358             }
1359             else {
1360                 $configuration->{nodes} = [$val];
1361             }
1362         } else {
1363             $configuration->{$var} = $val;
1364         }
1365     }
1366
1367     $configuration->{cxn_pool} //= 'Static';
1368
1369     return $configuration;
1370 }
1371
1372 =head2 get_facetable_fields
1373
1374 my @facetable_fields = Koha::SearchEngine::Elasticsearch->get_facetable_fields();
1375
1376 Returns the list of Koha::SearchFields marked to be faceted in the ES configuration
1377
1378 =cut
1379
1380 sub get_facetable_fields {
1381     my ($self) = @_;
1382
1383     # These should correspond to the ES field names, as opposed to the CCL
1384     # things that zebra uses.
1385     my @search_field_names = qw( author itype location su-geo title-series subject ccode holdingbranch homebranch ln );
1386     my @faceted_fields = Koha::SearchFields->search(
1387         { name => { -in => \@search_field_names }, facet_order => { '!=' => undef } }, { order_by => ['facet_order'] }
1388     )->as_list;
1389     my @not_faceted_fields = Koha::SearchFields->search(
1390         { name => { -in => \@search_field_names }, facet_order => undef }, { order_by => ['facet_order'] }
1391     )->as_list;
1392     # This could certainly be improved
1393     return ( @faceted_fields, @not_faceted_fields );
1394 }
1395
1396 =head2 clear_search_fields_cache
1397
1398 Koha::SearchEngine::Elasticsearch->clear_search_fields_cache();
1399
1400 Clear cached values for ES search fields
1401
1402 =cut
1403
1404 sub clear_search_fields_cache {
1405
1406     my $cache = Koha::Caches->get_instance();
1407     $cache->clear_from_cache('elasticsearch_search_fields_staff_client_biblios');
1408     $cache->clear_from_cache('elasticsearch_search_fields_opac_biblios');
1409     $cache->clear_from_cache('elasticsearch_search_fields_staff_client_authorities');
1410     $cache->clear_from_cache('elasticsearch_search_fields_opac_authorities');
1411
1412 }
1413
1414 1;
1415
1416 __END__
1417
1418 =head1 AUTHOR
1419
1420 =over 4
1421
1422 =item Chris Cormack C<< <chrisc@catalyst.net.nz> >>
1423
1424 =item Robin Sheat C<< <robin@catalyst.net.nz> >>
1425
1426 =item Jonathan Druart C<< <jonathan.druart@bugs.koha-community.org> >>
1427
1428 =back
1429
1430 =cut