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