1 package Koha::SearchEngine::Elasticsearch;
3 # Copyright 2015 Catalyst IT
5 # This file is part of Koha.
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.
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.
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>.
20 use base qw(Class::Accessor);
25 use Koha::Exceptions::Config;
26 use Koha::Exceptions::Elasticsearch;
27 use Koha::Filter::MARC::EmbedSeeFromHeadings;
28 use Koha::SearchFields;
29 use Koha::SearchMarcMaps;
32 use C4::AuthoritiesMarc qw( GuessAuthTypeCode );
35 use Carp qw( carp croak );
36 use Clone qw( clone );
38 use Readonly qw( Readonly );
39 use Search::Elasticsearch;
40 use Try::Tiny qw( catch try );
43 use List::Util qw( sum0 );
45 use MIME::Base64 qw( encode_base64 );
46 use Encode qw( encode );
48 use Scalar::Util qw( looks_like_number );
50 __PACKAGE__->mk_ro_accessors(qw( index index_name ));
51 __PACKAGE__->mk_accessors(qw( sort_fields ));
53 # Constants to refer to the standard index names
54 Readonly our $BIBLIOS_INDEX => 'biblios';
55 Readonly our $AUTHORITIES_INDEX => 'authorities';
59 Koha::SearchEngine::Elasticsearch - Base module for things using elasticsearch
67 The name of the index to use, generally 'biblios' or 'authorities'.
71 The Elasticsearch index name with Koha instance prefix.
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};
89 my $self = $class->SUPER::new(@_);
93 =head2 get_elasticsearch
95 my $elasticsearch_client = $self->get_elasticsearch();
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.
102 sub get_elasticsearch {
104 unless (defined $self->{elasticsearch}) {
105 $self->{elasticsearch} = Search::Elasticsearch->new(
106 $self->get_elasticsearch_params()
109 return $self->{elasticsearch};
112 =head2 get_elasticsearch_params
114 my $params = $self->get_elasticsearch_params();
116 This provides a hashref that contains the parameters for connecting to the
117 ElasicSearch servers, in the form:
120 'nodes' => ['127.0.0.1:9200', 'anotherserver:9200'],
121 'index_name' => 'koha_instance_index',
124 This is configured by the following in the C<config> block in koha-conf.xml:
127 <server>127.0.0.1:9200</server>
128 <server>anotherserver:9200</server>
129 <index_name>koha_instance</index_name>
134 sub get_elasticsearch_params {
139 $conf = _read_configuration();
141 if ( ref($_) eq 'Koha::Exceptions::Config::MissingEntry' ) {
149 =head2 get_elasticsearch_settings
151 my $settings = $self->get_elasticsearch_settings();
153 This provides the settings provided to Elasticsearch when an index is created.
154 These can do things like define tokenization methods.
156 A hashref containing the settings is returned.
160 sub get_elasticsearch_settings {
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 );
174 =head2 get_elasticsearch_mappings
176 my $mappings = $self->get_elasticsearch_mappings();
178 This provides the mappings that get passed to Elasticsearch when an index is
183 sub get_elasticsearch_mappings {
186 # Use state to speed up repeated calls
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(
197 my ( $name, $type, $facet, $suggestible, $sort, $search, $filter, $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.
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') {
211 } elsif ($type eq 'year') {
213 } elsif ($type eq 'callnumber') {
214 $es_type = 'cn_sort';
218 $mappings->{properties}{$name} = _get_elasticsearch_field_config('search', $es_type);
222 $mappings->{properties}{ $name . '__facet' } = _get_elasticsearch_field_config('facet', $es_type);
225 $mappings->{properties}{ $name . '__suggestion' } = _get_elasticsearch_field_config('suggestible', $es_type);
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;
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');
241 $all_mappings{$self->index} = $mappings;
243 $self->sort_fields(\%{$sort_fields{$self->index}});
244 return $all_mappings{$self->index};
247 =head2 raw_elasticsearch_mappings
249 Return elasticsearch mapping as it is in database.
250 marc_type: marc21|unimarc
252 $raw_mappings = raw_elasticsearch_mappings( $marc_type )
256 sub raw_elasticsearch_mappings {
257 my ( $marc_type ) = @_;
259 my $schema = Koha::Database->new()->schema();
261 my $search_fields = Koha::SearchFields->search({}, { order_by => { -asc => 'name' } });
264 while ( my $search_field = $search_fields->next ) {
266 my $marc_to_fields = $schema->resultset('SearchMarcToField')->search(
267 { search_field_id => $search_field->id },
269 join => 'search_marc_map',
270 order_by => { -asc => ['search_marc_map.marc_type','search_marc_map.marc_field'] }
274 while ( my $marc_to_field = $marc_to_fields->next ) {
276 my $marc_map = $marc_to_field->search_marc_map;
278 next if $marc_type && $marc_map->marc_type ne $marc_type;
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;
288 push (@{ $mappings->{ $marc_map->index_name }{ $search_field->name }{mappings} },
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 filter => $marc_to_field->filter || ''
304 =head2 _get_elasticsearch_field_config
306 Get the Elasticsearch field config for the given purpose and data type.
308 $mapping = _get_elasticsearch_field_config('search', 'text');
312 sub _get_elasticsearch_field_config {
314 my ( $purpose, $type ) = @_;
316 # Use state to speed up repeated calls
317 state $settings = undef;
318 if (!defined $settings) {
319 my $config_file = C4::Context->config('elasticsearch_field_config');
320 $config_file ||= C4::Context->config('intranetdir') . '/admin/searchengine/elasticsearch/field_config.yaml';
321 local $YAML::XS::Boolean = 'JSON::PP';
322 $settings = YAML::XS::LoadFile( $config_file );
325 if (!defined $settings->{$purpose}) {
326 die "Field purpose $purpose not defined in field config";
329 return $settings->{$purpose};
331 if (defined $settings->{$purpose}{$type}) {
332 return $settings->{$purpose}{$type};
334 if (defined $settings->{$purpose}{'default'}) {
335 return $settings->{$purpose}{'default'};
340 =head2 _load_elasticsearch_mappings
342 Load Elasticsearch mappings in the format of mappings.yaml.
344 $indexes = _load_elasticsearch_mappings();
348 sub _load_elasticsearch_mappings {
349 my $mappings_yaml = C4::Context->config('elasticsearch_index_mappings');
350 $mappings_yaml ||= C4::Context->config('intranetdir') . '/admin/searchengine/elasticsearch/mappings.yaml';
351 return YAML::XS::LoadFile( $mappings_yaml );
354 sub reset_elasticsearch_mappings {
356 my $indexes = $self->_load_elasticsearch_mappings();
358 Koha::SearchMarcMaps->delete;
359 Koha::SearchFields->delete;
361 while ( my ( $index_name, $fields ) = each %$indexes ) {
362 while ( my ( $field_name, $data ) = each %$fields ) {
364 my %sf_params = map { $_ => $data->{$_} } grep { exists $data->{$_} } qw/ type label weight staff_client opac facet_order mandatory/;
367 $sf_params{staff_client} //= 1;
368 $sf_params{opac} //= 1;
370 $sf_params{name} = $field_name;
372 my $search_field = Koha::SearchFields->find_or_create( \%sf_params, { key => 'name' } );
374 my $mappings = $data->{mappings};
375 for my $mapping ( @$mappings ) {
376 my $marc_field = Koha::SearchMarcMaps->find_or_create({
377 index_name => $index_name,
378 marc_type => $mapping->{marc_type},
379 marc_field => $mapping->{marc_field}
381 $search_field->add_to_search_marc_maps($marc_field, {
382 facet => $mapping->{facet} || 0,
383 suggestible => $mapping->{suggestible} || 0,
384 sort => $mapping->{sort} // 1,
385 search => $mapping->{search} // 1,
386 filter => $mapping->{filter} // ''
392 $self->clear_search_fields_cache();
394 # FIXME return the mappings?
397 # This overrides the accessor provided by Class::Accessor so that if
398 # sort_fields isn't set, then it'll generate it.
402 $self->_sort_fields_accessor(@_);
405 my $val = $self->_sort_fields_accessor();
408 # This will populate the accessor as a side effect
409 $self->get_elasticsearch_mappings();
410 return $self->_sort_fields_accessor();
413 =head2 _process_mappings($mappings, $data, $record_document, $meta)
415 $self->_process_mappings($mappings, $marc_field_data, $record_document, 0)
417 Process all C<$mappings> targets operating on a specific MARC field C<$data>.
418 Since we group all mappings by MARC field targets C<$mappings> will contain
419 all targets for C<$data> and thus we need to fetch the MARC field only once.
420 C<$mappings> will be applied to C<$record_document> and new field values added.
421 The method has no return value.
427 Arrayref of mappings containing arrayrefs in the format
428 [C<$target>, C<$options>] where C<$target> is the name of the target field and
429 C<$options> is a hashref containing processing directives for this particular
434 The source data from a MARC record field.
436 =item C<$record_document>
438 Hashref representing the Elasticsearch document on which mappings should be
443 A hashref containing metadata useful for enforcing per mapping rules. For
444 example for providing extra context for mapping options, or treating mapping
445 targets differently depending on type (sort, search, facet etc). Combining
446 this metadata with the mapping options and metadata allows us to mutate the
447 data per mapping, or even replace it with other data retrieved from the
450 Current properties are:
452 C<altscript>: A boolean value indicating whether an alternate script presentation is being
455 C<data_source>: The source of the $<data> argument. Possible values are: 'leader', 'control_field',
456 'subfield' or 'subfields_group'.
458 C<code>: The code of the subfield C<$data> was retrieved, if C<data_source> is 'subfield'.
460 C<codes>: Subfield codes of the subfields group from which C<$data> was retrieved, if C<data_source>
461 is 'subfields_group'.
463 C<field>: The original C<MARC::Record> object.
469 sub _process_mappings {
470 my ($_self, $mappings, $data, $record_document, $meta) = @_;
471 foreach my $mapping (@{$mappings}) {
472 my ($target, $options) = @{$mapping};
474 # Don't process sort fields for alternate scripts
475 my $sort = $target =~ /__sort$/;
476 if ($sort && $meta->{altscript}) {
480 # Copy (scalar) data since can have multiple targets
481 # with differing options for (possibly) mutating data
482 # so need a different copy for each
483 my $data_copy = $data;
484 if (defined $options->{substr}) {
485 my ($start, $length) = @{$options->{substr}};
486 $data_copy = length($data) > $start ? substr $data_copy, $start, $length : '';
489 # Add data to values array for callbacks processing
490 my $values = [$data_copy];
492 # Value callbacks takes subfield data (or values from previous
493 # callbacks) as argument, and returns a possibly different list of values.
494 # Note that the returned list may also be empty.
495 if (defined $options->{value_callbacks}) {
496 foreach my $callback (@{$options->{value_callbacks}}) {
497 # Pass each value to current callback which returns a list
498 # (scalar is fine too) resulting either in a list or
499 # a list of lists that will be flattened by perl.
500 # The next callback will receive the possibly expanded list of values.
501 $values = [ map { $callback->($_) } @{$values} ];
505 # Skip mapping if all values has been removed
506 next unless @{$values};
508 if (defined $options->{property}) {
509 $values = [ map { { $options->{property} => $_ } if $_} @{$values} ];
511 if (defined $options->{nonfiling_characters_indicator}) {
512 my $nonfiling_chars = $meta->{field}->indicator($options->{nonfiling_characters_indicator});
513 $nonfiling_chars = looks_like_number($nonfiling_chars) ? int($nonfiling_chars) : 0;
514 # Nonfiling chars does not make sense for multiple values
515 # Only apply on first element
516 if ( $nonfiling_chars > 0 ) {
518 $values->[0] = substr $values->[0], $nonfiling_chars;
520 push @{$values}, substr $values->[0], $nonfiling_chars;
525 $values = [ grep(!/^$/, @{$values}) ];
527 $record_document->{$target} //= [];
528 push @{$record_document->{$target}}, @{$values};
532 =head2 marc_records_to_documents($marc_records)
534 my $record_documents = $self->marc_records_to_documents($marc_records);
536 Using mappings stored in database convert C<$marc_records> to Elasticsearch documents.
538 Returns array of hash references, representing Elasticsearch documents,
539 acceptable as body payload in C<Search::Elasticsearch> requests.
543 =item C<$marc_documents>
545 Reference to array of C<MARC::Record> objects to be converted to Elasticsearch documents.
551 sub marc_records_to_documents {
552 my ($self, $records) = @_;
553 my $rules = $self->_get_marc_mapping_rules();
554 my $control_fields_rules = $rules->{control_fields};
555 my $data_fields_rules = $rules->{data_fields};
556 my $marcflavour = lc C4::Context->preference('marcflavour');
557 my $use_array = C4::Context->preference('ElasticsearchMARCFormat') eq 'ARRAY';
559 my @record_documents;
561 my %auth_match_headings;
562 if( $self->index eq 'authorities' ){
563 my @auth_types = Koha::Authority::Types->search->as_list;
564 %auth_match_headings = map { $_->authtypecode => $_->auth_tag_to_report } @auth_types;
567 foreach my $record (@{$records}) {
568 my $record_document = {};
570 if ( $self->index eq 'authorities' ){
571 my $authtypecode = GuessAuthTypeCode( $record );
573 if( $authtypecode !~ m/_SUBD/ ){ #Subdivision records will not be used for linking and so don't require match-heading to be built
574 my $field = $record->field( $auth_match_headings{ $authtypecode } );
575 my $heading = C4::Heading->new_from_field( $field, undef, 1 ); #new auth heading
576 push @{$record_document->{'match-heading'}}, $heading->search_form if $heading;
579 warn "Cannot determine authority type for record: " . $record->field('001')->as_string;
583 my $mappings = $rules->{leader};
585 $self->_process_mappings($mappings, $record->leader(), $record_document, {
587 data_source => 'leader'
591 foreach my $field ($record->fields()) {
592 if ($field->is_control_field()) {
593 my $mappings = $control_fields_rules->{$field->tag()};
595 $self->_process_mappings($mappings, $field->data(), $record_document, {
597 data_source => 'control_field',
604 my $tag = $field->tag();
605 # Handle alternate scripts in MARC 21
607 if ($marcflavour eq 'marc21' && $tag eq '880') {
608 my $sub6 = $field->subfield('6');
609 if ($sub6 =~ /^(...)-\d+/) {
615 my $data_field_rules = $data_fields_rules->{$tag};
616 if ($data_field_rules) {
617 my $subfields_mappings = $data_field_rules->{subfields};
618 my $wildcard_mappings = $subfields_mappings->{'*'};
619 foreach my $subfield ($field->subfields()) {
620 my ($code, $data) = @{$subfield};
621 my $mappings = $subfields_mappings->{$code} // [];
622 if ($wildcard_mappings) {
623 $mappings = [@{$mappings}, @{$wildcard_mappings}];
626 $self->_process_mappings($mappings, $data, $record_document, {
627 altscript => $altscript,
628 data_source => 'subfield',
636 my $subfields_join_mappings = $data_field_rules->{subfields_join};
637 if ($subfields_join_mappings) {
638 foreach my $subfields_group (keys %{$subfields_join_mappings}) {
639 my $data_field = $field->clone; #copy field to preserve for alt scripts
640 $data_field->delete_subfield(match => qr/^$/); #remove empty subfields, otherwise they are printed as a space
641 my $data = $data_field->as_string( $subfields_group ); #get values for subfields as a combined string, preserving record order
643 $self->_process_mappings($subfields_join_mappings->{$subfields_group}, $data, $record_document, {
644 altscript => $altscript,
645 data_source => 'subfields_group',
646 codes => $subfields_group,
657 if (C4::Context->preference('IncludeSeeFromInSearches') and $self->index eq 'biblios') {
658 foreach my $field (Koha::Filter::MARC::EmbedSeeFromHeadings->new->fields($record)) {
659 my $data_field_rules = $data_fields_rules->{$field->tag()};
660 if ($data_field_rules) {
661 my $subfields_mappings = $data_field_rules->{subfields};
662 my $wildcard_mappings = $subfields_mappings->{'*'};
663 foreach my $subfield ($field->subfields()) {
664 my ($code, $data) = @{$subfield};
666 push @mappings, @{ $subfields_mappings->{$code} } if $subfields_mappings->{$code};
667 push @mappings, @$wildcard_mappings if $wildcard_mappings;
668 # Do not include "see from" into these kind of fields
669 @mappings = grep { $_->[0] !~ /__(sort|facet|suggestion)$/ } @mappings;
671 $self->_process_mappings(\@mappings, $data, $record_document, {
672 data_source => 'subfield',
680 my $subfields_join_mappings = $data_field_rules->{subfields_join};
681 if ($subfields_join_mappings) {
682 foreach my $subfields_group (keys %{$subfields_join_mappings}) {
683 my $data_field = $field->clone;
684 # remove empty subfields, otherwise they are printed as a space
685 $data_field->delete_subfield(match => qr/^$/);
686 my $data = $data_field->as_string( $subfields_group );
688 my @mappings = @{ $subfields_join_mappings->{$subfields_group} };
689 # Do not include "see from" into these kind of fields
690 @mappings = grep { $_->[0] !~ /__(sort|facet|suggestion)$/ } @mappings;
691 $self->_process_mappings(\@mappings, $data, $record_document, {
692 data_source => 'subfields_group',
693 codes => $subfields_group,
704 foreach my $field (keys %{$rules->{defaults}}) {
705 unless (defined $record_document->{$field}) {
706 $record_document->{$field} = $rules->{defaults}->{$field};
709 foreach my $field (@{$rules->{sum}}) {
710 if (defined $record_document->{$field}) {
711 # TODO: validate numeric? filter?
712 # TODO: Or should only accept fields without nested values?
713 # TODO: Quick and dirty, improve if needed
714 $record_document->{$field} = sum0(grep { !ref($_) && m/\d+(\.\d+)?/} @{$record_document->{$field}});
717 # Index all applicable ISBN forms (ISBN-10 and ISBN-13 with and without dashes)
718 foreach my $field (@{$rules->{isbn}}) {
719 if (defined $record_document->{$field}) {
721 foreach my $input_isbn (@{$record_document->{$field}}) {
722 my $isbn = Business::ISBN->new($input_isbn);
723 if (defined $isbn && $isbn->is_valid) {
724 my $isbn13 = $isbn->as_isbn13->as_string;
725 push @isbns, $isbn13;
727 push @isbns, $isbn13;
729 my $isbn10 = $isbn->as_isbn10;
731 $isbn10 = $isbn10->as_string;
732 push @isbns, $isbn10;
734 push @isbns, $isbn10;
737 push @isbns, $input_isbn;
740 $record_document->{$field} = \@isbns;
744 # Remove duplicate values and collapse sort fields
745 foreach my $field (keys %{$record_document}) {
746 if (ref($record_document->{$field}) eq 'ARRAY') {
747 @{$record_document->{$field}} = do {
749 grep { !$seen{ref($_) eq 'HASH' && defined $_->{input} ? $_->{input} : $_}++ } @{$record_document->{$field}};
751 if ($field =~ /__sort$/) {
752 # Make sure to keep the sort field length sensible. 255 was chosen as a nice round value.
753 $record_document->{$field} = [substr(join(' ', @{$record_document->{$field}}), 0, 255)];
758 # TODO: Perhaps should check if $records_document non empty, but really should never be the case
759 $record->encoding('UTF-8');
761 $record_document->{'marc_data_array'} = $self->_marc_to_array($record);
762 $record_document->{'marc_format'} = 'ARRAY';
766 # Temporarily intercept all warn signals (MARC::Record carps when record length > 99999)
767 local $SIG{__WARN__} = sub {
768 push @warnings, $_[0];
770 $record_document->{'marc_data'} = encode_base64(encode('UTF-8', $record->as_usmarc()));
773 # Suppress warnings if record length exceeded
774 unless (substr($record->leader(), 0, 5) eq '99999') {
775 foreach my $warning (@warnings) {
779 $record_document->{'marc_data'} = $record->as_xml_record($marcflavour);
780 $record_document->{'marc_format'} = 'MARCXML';
783 $record_document->{'marc_format'} = 'base64ISO2709';
787 # Check if there is at least one available item
788 if ($self->index eq $BIBLIOS_INDEX) {
789 my ($tag, $code) = C4::Biblio::GetMarcFromKohaField('biblio.biblionumber');
790 my $field = $record->field($tag);
792 my $biblionumber = $field->is_control_field ? $field->data : $field->subfield($code);
793 my $avail_items = Koha::Items->search({
794 biblionumber => $biblionumber,
799 $record_document->{available} = $avail_items ? \1 : \0;
803 push @record_documents, $record_document;
805 return \@record_documents;
808 =head2 _marc_to_array($record)
810 my @fields = _marc_to_array($record)
812 Convert a MARC::Record to an array modeled after MARC-in-JSON
813 (see https://github.com/marc4j/marc4j/wiki/MARC-in-JSON-Description)
819 A MARC::Record object
826 my ($self, $record) = @_;
829 leader => $record->leader(),
832 for my $field ($record->fields()) {
833 my $tag = $field->tag();
834 if ($field->is_control_field()) {
835 push @{$data->{fields}}, {$tag => $field->data()};
838 foreach my $subfield ($field->subfields()) {
839 my ($code, $contents) = @{$subfield};
840 push @{$subfields}, {$code => $contents};
842 push @{$data->{fields}}, {
844 ind1 => $field->indicator(1),
845 ind2 => $field->indicator(2),
846 subfields => $subfields
854 =head2 _array_to_marc($data)
856 my $record = _array_to_marc($data)
858 Convert an array modeled after MARC-in-JSON to a MARC::Record
864 An array modeled after MARC-in-JSON
865 (see https://github.com/marc4j/marc4j/wiki/MARC-in-JSON-Description)
872 my ($self, $data) = @_;
874 my $record = MARC::Record->new();
876 $record->leader($data->{leader});
877 for my $field (@{$data->{fields}}) {
878 my $tag = (keys %{$field})[0];
879 $field = $field->{$tag};
881 if (ref($field) eq 'HASH') {
883 foreach my $subfield (@{$field->{subfields}}) {
884 my $code = (keys %{$subfield})[0];
885 push @subfields, $code;
886 push @subfields, $subfield->{$code};
888 $marc_field = MARC::Field->new($tag, $field->{ind1}, $field->{ind2}, @subfields);
890 $marc_field = MARC::Field->new($tag, $field)
892 $record->append_fields($marc_field);
898 =head2 _field_mappings($facet, $suggestible, $sort, $search, $filter, $target_name, $target_type, $range)
900 my @mappings = _field_mappings( $facet, $suggestible, $sort, $search, $filter, $target_name, $target_type, $range )
902 Get mappings, an internal data structure later used by
903 L<_process_mappings($mappings, $data, $record_document, $meta)> to process MARC target
904 data for a MARC mapping.
906 The returned C<$mappings> is not to to be confused with mappings provided by
907 C<_foreach_mapping>, rather this sub accepts properties from a mapping as
908 provided by C<_foreach_mapping> and expands it to this internal data structure.
909 In the caller context (C<_get_marc_mapping_rules>) the returned C<@mappings>
910 is then applied to each MARC target (leader, control field data, subfield or
911 joined subfields) and integrated into the mapping rules data structure used in
912 C<marc_records_to_documents> to transform MARC records into Elasticsearch
919 Boolean indicating whether to create a facet field for this mapping.
921 =item C<$suggestible>
923 Boolean indicating whether to create a suggestion field for this mapping.
927 Boolean indicating whether to create a sort field for this mapping.
931 Boolean indicating whether to create a search field for this mapping.
933 =item C<$target_name>
935 Elasticsearch document target field name.
937 =item C<$target_type>
939 Elasticsearch document target field type.
943 An optional range as a string in the format "<START>-<END>" or "<START>",
944 where "<START>" and "<END>" are integers specifying a range that will be used
945 for extracting a substring from MARC data as Elasticsearch field target value.
947 The first character position is "0", and the range is inclusive,
948 so "0-2" means the first three characters of MARC data.
950 If only "<START>" is provided only one character at position "<START>" will
957 sub _field_mappings {
958 my ( $_self, $facet, $suggestible, $sort, $search, $filter, $target_name, $target_type, $range ) = @_;
959 my %mapping_defaults = ();
962 my $substr_args = undef;
963 if (defined $range) {
964 # TODO: use value_callback instead?
965 my ($start, $end) = map(int, split /-/, $range, 2);
966 $substr_args = [$start];
967 push @{$substr_args}, (defined $end ? $end - $start + 1 : 1);
969 my $default_options = {};
971 $default_options->{substr} = $substr_args;
974 # TODO: Should probably have per type value callback/hook
975 # but hard code for now
976 if ($target_type eq 'boolean') {
977 $default_options->{value_callbacks} //= [];
978 push @{$default_options->{value_callbacks}}, sub {
980 # Trim whitespace at both ends
981 $value =~ s/^\s+|\s+$//g;
982 return $value ? 'true' : 'false';
985 elsif ($target_type eq 'year') {
986 $default_options->{value_callbacks} //= [];
987 # Only accept years containing digits and "u"
988 push @{$default_options->{value_callbacks}}, sub {
990 # Replace "u" with "0" for sorting
991 return map { s/[u\s]/0/gr } ( $value =~ /[0-9u\s]{4}/g );
995 if ( defined $filter && $filter eq 'punctuation' ) {
996 $default_options->{value_callbacks} //= [];
997 push @{ $default_options->{value_callbacks} }, sub {
1000 # Trim punctuation marks from field
1002 s/[\x00-\x1F,\x21-\x2F,\x3A-\x40,\x5B-\x60,\x7B-\x89,\x8B,\x8D,\x8F,\x90-\x99,\x9B,\x9D,\xA0-\xBF,\xD7,\xF7]//g;
1008 my $mapping = [$target_name, $default_options];
1009 push @mappings, $mapping;
1013 push @suffixes, 'facet' if $facet;
1014 push @suffixes, 'suggestion' if $suggestible;
1015 push @suffixes, 'sort' if !defined $sort || $sort;
1017 foreach my $suffix (@suffixes) {
1018 my $mapping = ["${target_name}__$suffix"];
1019 # TODO: Hack, fix later in less hideous manner
1020 if ($suffix eq 'suggestion') {
1021 push @{$mapping}, {%{$default_options}, property => 'input'};
1024 # Important! Make shallow clone, or we end up with the same hashref
1025 # shared by all mappings
1026 push @{$mapping}, {%{$default_options}};
1028 push @mappings, $mapping;
1033 =head2 _get_marc_mapping_rules
1035 my $mapping_rules = $self->_get_marc_mapping_rules()
1037 Generates rules from mappings stored in database for MARC records to Elasticsearch JSON document conversion.
1039 Since field retrieval is slow in C<MARC::Records> (all fields are itereted through for
1040 each call to C<MARC::Record>->field) we create an optimized structure of mapping
1041 rules keyed by MARC field tags holding all the mapping rules for that particular tag.
1043 We can then iterate through all MARC fields for each record and apply all relevant
1044 rules once per fields instead of retreiving fields multiple times for each mapping rule
1045 which is terribly slow.
1049 # TODO: This structure can be used for processing multiple MARC::Records so is currently
1050 # rebuilt for each batch. Since it is cacheable it could also be stored in an in
1051 # memory cache which it is currently not. The performance gain of caching
1052 # would probably be marginal, but to do this could be a further improvement.
1054 sub _get_marc_mapping_rules {
1056 my $marcflavour = lc C4::Context->preference('marcflavour');
1057 my $field_spec_regexp = qr/^([0-9]{3})([()0-9a-zA-Z]+)?(?:_\/(\d+(?:-\d+)?))?$/;
1058 my $leader_regexp = qr/^leader(?:_\/(\d+(?:-\d+)?))?$/;
1061 'control_fields' => {},
1062 'data_fields' => {},
1068 $self->_foreach_mapping(sub {
1069 my ($name, $type, $facet, $suggestible, $sort, $search, $filter, $marc_type, $marc_field) = @_;
1070 return if $marc_type ne $marcflavour;
1072 if ($type eq 'sum') {
1073 push @{$rules->{sum}}, $name;
1074 push @{$rules->{sum}}, $name."__sort" if $sort;
1076 elsif ($type eq 'isbn') {
1077 push @{$rules->{isbn}}, $name;
1079 elsif ($type eq 'boolean') {
1080 # boolean gets special handling, if value doesn't exist for a field,
1081 # it is set to false
1082 $rules->{defaults}->{$name} = 'false';
1085 if ($marc_field =~ $field_spec_regexp) {
1089 my @subfield_groups;
1090 # Parse and separate subfields form subfield groups
1092 my $subfield_group = '';
1095 foreach my $token (split //, $2) {
1096 if ($token eq "(") {
1098 Koha::Exceptions::Elasticsearch::MARCFieldExprParseError->throw(
1099 "Unmatched opening parenthesis for $marc_field"
1106 elsif ($token eq ")") {
1108 if ($subfield_group) {
1109 push @subfield_groups, $subfield_group;
1110 $subfield_group = '';
1115 Koha::Exceptions::Elasticsearch::MARCFieldExprParseError->throw(
1116 "Unmatched closing parenthesis for $marc_field"
1120 elsif ($open_group) {
1121 $subfield_group .= $token;
1124 push @subfields, $token;
1129 push @subfields, '*';
1132 my $range = defined $3 ? $3 : undef;
1133 my @mappings = $self->_field_mappings($facet, $suggestible, $sort, $search, $filter, $name, $type, $range);
1134 if ($field_tag < 10) {
1135 $rules->{control_fields}->{$field_tag} //= [];
1136 push @{$rules->{control_fields}->{$field_tag}}, @{clone(\@mappings)};
1139 $rules->{data_fields}->{$field_tag} //= {};
1140 foreach my $subfield (@subfields) {
1141 $rules->{data_fields}->{$field_tag}->{subfields}->{$subfield} //= [];
1142 push @{$rules->{data_fields}->{$field_tag}->{subfields}->{$subfield}}, @{clone(\@mappings)};
1144 foreach my $subfield_group (@subfield_groups) {
1145 $rules->{data_fields}->{$field_tag}->{subfields_join}->{$subfield_group} //= [];
1146 push @{$rules->{data_fields}->{$field_tag}->{subfields_join}->{$subfield_group}}, @{clone(\@mappings)};
1150 elsif ($marc_field =~ $leader_regexp) {
1151 my $range = defined $1 ? $1 : undef;
1152 my @mappings = $self->_field_mappings($facet, $suggestible, $sort, $search, $filter, $name, $type, $range);
1153 push @{$rules->{leader}}, @{clone(\@mappings)};
1156 Koha::Exceptions::Elasticsearch::MARCFieldExprParseError->throw(
1157 "Invalid MARC field expression: $marc_field"
1162 # Marc-flavour specific rule tweaks, could/should also provide hook for this
1163 if ($marcflavour eq 'marc21') {
1164 # Nonfiling characters processing for sort fields
1166 if ($self->index eq $Koha::SearchEngine::BIBLIOS_INDEX) {
1167 # Format is: nonfiling characters indicator => field names list
1169 1 => [130, 630, 730, 740],
1170 2 => [222, 240, 242, 243, 245, 440, 830]
1173 elsif ($self->index eq $Koha::SearchEngine::AUTHORITIES_INDEX) {
1176 2 => [130, 430, 530]
1179 foreach my $indicator (keys %title_fields) {
1180 foreach my $field_tag (@{$title_fields{$indicator}}) {
1181 my $mappings = $rules->{data_fields}->{$field_tag}->{subfields}->{a} // [];
1182 foreach my $mapping ( @{$mappings} ) {
1183 # Mark this as to be processed for nonfiling characters indicator
1184 # later on in _process_mappings
1185 $mapping->[1]->{nonfiling_characters_indicator} = $indicator;
1191 if( $self->index eq 'authorities' ){
1192 push @{$rules->{control_fields}->{'008'}}, ['subject-heading-thesaurus', { 'substr' => [ 11, 1 ] } ];
1193 push @{$rules->{data_fields}->{'040'}->{subfields}->{f}}, ['subject-heading-thesaurus', { } ];
1199 =head2 _foreach_mapping
1201 $self->_foreach_mapping(
1203 my ( $name, $type, $facet, $suggestible, $sort, $search, $filter, $marc_type,
1206 return unless $marc_type eq 'marc21';
1207 print "Data comes from: " . $marc_field . "\n";
1211 This allows you to apply a function to each entry in the elasticsearch mappings
1212 table, in order to build the mappings for whatever is needed.
1214 In the provided function, the files are:
1220 The field name for elasticsearch (corresponds to the 'mapping' column in the
1225 The type for this value, e.g. 'string'.
1229 True if this value should be facetised. This only really makes sense if the
1230 field is understood by the facet processing code anyway.
1234 True if this is a field that a) needs special sort handling, and b) if it
1235 should be sorted on. False if a) but not b). Undef if not a). This allows,
1236 for example, author to be sorted on but not everything marked with "author"
1237 to be included in that sort.
1241 True if this value should be searchable.
1245 Contains a string that represents a filter defined in the indexing code. Currently supports
1246 the option 'punctuation'
1250 A string that indicates the MARC type that this mapping is for, e.g. 'marc21',
1253 =item C<$marc_field>
1255 A string that describes the MARC field that contains the data to extract.
1261 sub _foreach_mapping {
1262 my ( $self, $sub ) = @_;
1264 # TODO use a caching framework here
1265 my $search_fields = Koha::Database->schema->resultset('SearchField')->search(
1267 'search_marc_map.index_name' => $self->index,
1269 { join => { search_marc_to_fields => 'search_marc_map' },
1271 'search_marc_to_fields.facet',
1272 'search_marc_to_fields.suggestible',
1273 'search_marc_to_fields.sort',
1274 'search_marc_to_fields.search',
1275 'search_marc_to_fields.filter',
1276 'search_marc_map.marc_type',
1277 'search_marc_map.marc_field',
1291 while ( my $search_field = $search_fields->next ) {
1293 # Force lower case on indexed field names for case insensitive
1294 # field name searches
1295 lc($search_field->name),
1296 $search_field->type,
1297 $search_field->get_column('facet'),
1298 $search_field->get_column('suggestible'),
1299 $search_field->get_column('sort'),
1300 $search_field->get_column('search'),
1301 $search_field->get_column('filter'),
1302 $search_field->get_column('marc_type'),
1303 $search_field->get_column('marc_field'),
1308 =head2 process_error
1310 die process_error($@);
1312 This parses an Elasticsearch error message and produces a human-readable
1313 result from it. This result is probably missing all the useful information
1314 that you might want in diagnosing an issue, so the warning is also logged.
1316 Note that currently the resulting message is not internationalised. This
1317 will happen eventually by some method or other.
1322 my ($self, $msg) = @_;
1324 warn $msg; # simple logging
1326 # This is super-primitive
1327 return "Unable to understand your search query, please rephrase and try again.\n" if $msg =~ /ParseException|parse_exception/;
1329 return "Unable to perform your search. Please try again.\n";
1332 =head2 _read_configuration
1334 my $conf = _read_configuration();
1336 Reads the I<configuration file> and returns a hash structure with the
1337 configuration information. It raises an exception if mandatory entries
1340 The hashref structure has the following form:
1343 'nodes' => ['127.0.0.1:9200', 'anotherserver:9200'],
1344 'index_name' => 'koha_instance',
1347 This is configured by the following in the C<config> block in koha-conf.xml:
1350 <server>127.0.0.1:9200</server>
1351 <server>anotherserver:9200</server>
1352 <index_name>koha_instance</index_name>
1357 sub _read_configuration {
1361 my $conf = C4::Context->config('elasticsearch');
1362 unless ( defined $conf ) {
1363 Koha::Exceptions::Config::MissingEntry->throw(
1364 "Missing <elasticsearch> entry in koha-conf.xml"
1368 unless ( exists $conf->{server} ) {
1369 Koha::Exceptions::Config::MissingEntry->throw(
1370 "Missing <elasticsearch>/<server> entry in koha-conf.xml"
1374 unless ( exists $conf->{index_name} ) {
1375 Koha::Exceptions::Config::MissingEntry->throw(
1376 "Missing <elasticsearch>/<index_name> entry in koha-conf.xml",
1380 while ( my ( $var, $val ) = each %$conf ) {
1381 if ( $var eq 'server' ) {
1382 if ( ref($val) eq 'ARRAY' ) {
1383 $configuration->{nodes} = $val;
1386 $configuration->{nodes} = [$val];
1389 $configuration->{$var} = $val;
1393 $configuration->{cxn_pool} //= 'Static';
1395 return $configuration;
1398 =head2 get_facetable_fields
1400 my @facetable_fields = Koha::SearchEngine::Elasticsearch->get_facetable_fields();
1402 Returns the list of Koha::SearchFields marked to be faceted in the ES configuration
1406 sub get_facetable_fields {
1409 # These should correspond to the ES field names, as opposed to the CCL
1410 # things that zebra uses.
1411 my @search_field_names = qw( author itype location su-geo title-series subject ccode holdingbranch homebranch ln );
1412 my @faceted_fields = Koha::SearchFields->search(
1413 { name => { -in => \@search_field_names }, facet_order => { '!=' => undef } }, { order_by => ['facet_order'] }
1415 my @not_faceted_fields = Koha::SearchFields->search(
1416 { name => { -in => \@search_field_names }, facet_order => undef }, { order_by => ['facet_order'] }
1418 # This could certainly be improved
1419 return ( @faceted_fields, @not_faceted_fields );
1422 =head2 clear_search_fields_cache
1424 Koha::SearchEngine::Elasticsearch->clear_search_fields_cache();
1426 Clear cached values for ES search fields
1430 sub clear_search_fields_cache {
1432 my $cache = Koha::Caches->get_instance();
1433 $cache->clear_from_cache('elasticsearch_search_fields_staff_client_biblios');
1434 $cache->clear_from_cache('elasticsearch_search_fields_opac_biblios');
1435 $cache->clear_from_cache('elasticsearch_search_fields_staff_client_authorities');
1436 $cache->clear_from_cache('elasticsearch_search_fields_opac_authorities');
1448 =item Chris Cormack C<< <chrisc@catalyst.net.nz> >>
1450 =item Robin Sheat C<< <robin@catalyst.net.nz> >>
1452 =item Jonathan Druart C<< <jonathan.druart@bugs.koha-community.org> >>