Bug 18235: Add missing POD to get_facetable_fields
[koha.git] / 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 under the
8 # terms of the GNU General Public License as published by the Free Software
9 # Foundation; either version 3 of the License, or (at your option) any later
10 # version.
11 #
12 # Koha is distributed in the hope that it will be useful, but WITHOUT ANY
13 # WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
14 # A PARTICULAR PURPOSE.  See the GNU General Public License for more details.
15 #
16 # You should have received a copy of the GNU General Public License along
17 # with Koha; if not, write to the Free Software Foundation, Inc.,
18 # 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
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::SearchFields;
27 use Koha::SearchMarcMaps;
28
29 use Carp;
30 use JSON;
31 use Modern::Perl;
32 use Readonly;
33 use Search::Elasticsearch;
34 use Try::Tiny;
35 use YAML::Syck;
36
37 use List::Util qw( sum0 reduce );
38 use MARC::File::XML;
39 use MIME::Base64;
40 use Encode qw(encode);
41 use Business::ISBN;
42
43 __PACKAGE__->mk_ro_accessors(qw( index ));
44 __PACKAGE__->mk_accessors(qw( sort_fields ));
45
46 # Constants to refer to the standard index names
47 Readonly our $BIBLIOS_INDEX     => 'biblios';
48 Readonly our $AUTHORITIES_INDEX => 'authorities';
49
50 =head1 NAME
51
52 Koha::SearchEngine::Elasticsearch - Base module for things using elasticsearch
53
54 =head1 ACCESSORS
55
56 =over 4
57
58 =item index
59
60 The name of the index to use, generally 'biblios' or 'authorities'.
61
62 =back
63
64 =head1 FUNCTIONS
65
66 =cut
67
68 sub new {
69     my $class = shift @_;
70     my $self = $class->SUPER::new(@_);
71     # Check for a valid index
72     Koha::Exceptions::MissingParameter->throw('No index name provided') unless $self->index;
73     return $self;
74 }
75
76 =head2 get_elasticsearch
77
78     my $elasticsearch_client = $self->get_elasticsearch();
79
80 Returns a C<Search::Elasticsearch> client. The client is cached on a C<Koha::SearchEngine::ElasticSearch>
81 instance level and will be reused if method is called multiple times.
82
83 =cut
84
85 sub get_elasticsearch {
86     my $self = shift @_;
87     unless (defined $self->{elasticsearch}) {
88         my $conf = $self->get_elasticsearch_params();
89         $self->{elasticsearch} = Search::Elasticsearch->new($conf);
90     }
91     return $self->{elasticsearch};
92 }
93
94 =head2 get_elasticsearch_params
95
96     my $params = $self->get_elasticsearch_params();
97
98 This provides a hashref that contains the parameters for connecting to the
99 ElasicSearch servers, in the form:
100
101     {
102         'nodes' => ['127.0.0.1:9200', 'anotherserver:9200'],
103         'index_name' => 'koha_instance_index',
104     }
105
106 This is configured by the following in the C<config> block in koha-conf.xml:
107
108     <elasticsearch>
109         <server>127.0.0.1:9200</server>
110         <server>anotherserver:9200</server>
111         <index_name>koha_instance</index_name>
112     </elasticsearch>
113
114 =cut
115
116 sub get_elasticsearch_params {
117     my ($self) = @_;
118
119     # Copy the hash so that we're not modifying the original
120     my $conf = C4::Context->config('elasticsearch');
121     die "No 'elasticsearch' block is defined in koha-conf.xml.\n" if ( !$conf );
122     my $es = { %{ $conf } };
123
124     # Helpfully, the multiple server lines end up in an array for us anyway
125     # if there are multiple ones, but not if there's only one.
126     my $server = $es->{server};
127     delete $es->{server};
128     if ( ref($server) eq 'ARRAY' ) {
129
130         # store it called 'nodes' (which is used by newer Search::Elasticsearch)
131         $es->{nodes} = $server;
132     }
133     elsif ($server) {
134         $es->{nodes} = [$server];
135     }
136     else {
137         die "No elasticsearch servers were specified in koha-conf.xml.\n";
138     }
139     die "No elasticsearch index_name was specified in koha-conf.xml.\n"
140       if ( !$es->{index_name} );
141     # Append the name of this particular index to our namespace
142     $es->{index_name} .= '_' . $self->index;
143
144     $es->{key_prefix} = 'es_';
145     $es->{client} //= '5_0::Direct';
146     $es->{cxn_pool} //= 'Sniff';
147     $es->{request_timeout} //= 60;
148
149     return $es;
150 }
151
152 =head2 get_elasticsearch_settings
153
154     my $settings = $self->get_elasticsearch_settings();
155
156 This provides the settings provided to Elasticsearch when an index is created.
157 These can do things like define tokenization methods.
158
159 A hashref containing the settings is returned.
160
161 =cut
162
163 sub get_elasticsearch_settings {
164     my ($self) = @_;
165
166     # Use state to speed up repeated calls
167     state $settings = undef;
168     if (!defined $settings) {
169         my $config_file = C4::Context->config('elasticsearch_index_config');
170         $config_file ||= C4::Context->config('intranetdir') . '/admin/searchengine/elasticsearch/index_config.yaml';
171         $settings = LoadFile( $config_file );
172     }
173
174     return $settings;
175 }
176
177 =head2 get_elasticsearch_mappings
178
179     my $mappings = $self->get_elasticsearch_mappings();
180
181 This provides the mappings that get passed to Elasticsearch when an index is
182 created.
183
184 =cut
185
186 sub get_elasticsearch_mappings {
187     my ($self) = @_;
188
189     # Use state to speed up repeated calls
190     state %all_mappings;
191     state %sort_fields;
192
193     if (!defined $all_mappings{$self->index}) {
194         $sort_fields{$self->index} = {};
195         my $mappings = {
196             data => scalar _get_elasticsearch_mapping('general', '')
197         };
198         my $marcflavour = lc C4::Context->preference('marcflavour');
199         $self->_foreach_mapping(
200             sub {
201                 my ( $name, $type, $facet, $suggestible, $sort, $marc_type ) = @_;
202                 return if $marc_type ne $marcflavour;
203                 # TODO if this gets any sort of complexity to it, it should
204                 # be broken out into its own function.
205
206                 # TODO be aware of date formats, but this requires pre-parsing
207                 # as ES will simply reject anything with an invalid date.
208                 my $es_type = 'text';
209                 if ($type eq 'boolean') {
210                     $es_type = 'boolean';
211                 } elsif ($type eq 'number' || $type eq 'sum') {
212                     $es_type = 'integer';
213                 } elsif ($type eq 'isbn' || $type eq 'stdno') {
214                     $es_type = 'stdno';
215                 }
216
217                 $mappings->{data}{properties}{$name} = _get_elasticsearch_mapping('search', $es_type);
218
219                 if ($facet) {
220                     $mappings->{data}{properties}{ $name . '__facet' } = _get_elasticsearch_mapping('facet', $es_type);
221                 }
222                 if ($suggestible) {
223                     $mappings->{data}{properties}{ $name . '__suggestion' } = _get_elasticsearch_mapping('suggestible', $es_type);
224                 }
225                 # Sort is a bit special as it can be true, false, undef.
226                 # We care about "true" or "undef",
227                 # "undef" means to do the default thing, which is make it sortable.
228                 if (!defined $sort || $sort) {
229                     $mappings->{data}{properties}{ $name . '__sort' } = _get_elasticsearch_mapping('sort', $es_type);
230                     $sort_fields{$self->index}{$name} = 1;
231                 }
232             }
233         );
234         $all_mappings{$self->index} = $mappings;
235     }
236     $self->sort_fields(\%{$sort_fields{$self->index}});
237
238     return $all_mappings{$self->index};
239 }
240
241 =head2 _get_elasticsearch_mapping
242
243 Get the Elasticsearch mappings for the given purpose and data type.
244
245 $mapping = _get_elasticsearch_mapping('search', 'text');
246
247 =cut
248
249 sub _get_elasticsearch_mapping {
250
251     my ( $purpose, $type ) = @_;
252
253     # Use state to speed up repeated calls
254     state $settings = undef;
255     if (!defined $settings) {
256         my $config_file = C4::Context->config('elasticsearch_field_config');
257         $config_file ||= C4::Context->config('intranetdir') . '/admin/searchengine/elasticsearch/field_config.yaml';
258         $settings = LoadFile( $config_file );
259     }
260
261     if (!defined $settings->{$purpose}) {
262         die "Field purpose $purpose not defined in field config";
263     }
264     if ($type eq '') {
265         return $settings->{$purpose};
266     }
267     if (defined $settings->{$purpose}{$type}) {
268         return $settings->{$purpose}{$type};
269     }
270     if (defined $settings->{$purpose}{'default'}) {
271         return $settings->{$purpose}{'default'};
272     }
273     return;
274 }
275
276 sub reset_elasticsearch_mappings {
277     my ( $reset_fields ) = @_;
278     my $mappings_yaml = C4::Context->config('elasticsearch_index_mappings');
279     $mappings_yaml ||= C4::Context->config('intranetdir') . '/admin/searchengine/elasticsearch/mappings.yaml';
280     my $indexes = LoadFile( $mappings_yaml );
281
282     while ( my ( $index_name, $fields ) = each %$indexes ) {
283         while ( my ( $field_name, $data ) = each %$fields ) {
284             my $field_type = $data->{type};
285             my $field_label = $data->{label};
286             my $mappings = $data->{mappings};
287             my $facet_order = $data->{facet_order};
288             my $search_field = Koha::SearchFields->find_or_create({ name => $field_name });
289             $search_field->update(
290                 {
291                     label       => $field_label,
292                     type        => $field_type,
293                     facet_order => $facet_order
294                 }
295             );
296             for my $mapping ( @$mappings ) {
297                 my $marc_field = Koha::SearchMarcMaps->find_or_create({ index_name => $index_name, marc_type => $mapping->{marc_type}, marc_field => $mapping->{marc_field} });
298                 $search_field->add_to_search_marc_maps($marc_field, { facet => $mapping->{facet} || 0, suggestible => $mapping->{suggestible} || 0, sort => $mapping->{sort} } );
299             }
300         }
301     }
302 }
303
304 # This overrides the accessor provided by Class::Accessor so that if
305 # sort_fields isn't set, then it'll generate it.
306 sub sort_fields {
307     my $self = shift;
308     if (@_) {
309         $self->_sort_fields_accessor(@_);
310         return;
311     }
312     my $val = $self->_sort_fields_accessor();
313     return $val if $val;
314
315     # This will populate the accessor as a side effect
316     $self->get_elasticsearch_mappings();
317     return $self->_sort_fields_accessor();
318 }
319
320 =head2 _process_mappings($mappings, $data, $record_document, $altscript)
321
322     $self->_process_mappings($mappings, $marc_field_data, $record_document, 0)
323
324 Process all C<$mappings> targets operating on a specific MARC field C<$data>.
325 Since we group all mappings by MARC field targets C<$mappings> will contain
326 all targets for C<$data> and thus we need to fetch the MARC field only once.
327 C<$mappings> will be applied to C<$record_document> and new field values added.
328 The method has no return value.
329
330 =over 4
331
332 =item C<$mappings>
333
334 Arrayref of mappings containing arrayrefs in the format
335 [C<$target>, C<$options>] where C<$target> is the name of the target field and
336 C<$options> is a hashref containing processing directives for this particular
337 mapping.
338
339 =item C<$data>
340
341 The source data from a MARC record field.
342
343 =item C<$record_document>
344
345 Hashref representing the Elasticsearch document on which mappings should be
346 applied.
347
348 =item C<$altscript>
349
350 A boolean value indicating whether an alternate script presentation is being
351 processed.
352
353 =back
354
355 =cut
356
357 sub _process_mappings {
358     my ($_self, $mappings, $data, $record_document, $altscript) = @_;
359     foreach my $mapping (@{$mappings}) {
360         my ($target, $options) = @{$mapping};
361
362         # Don't process sort fields for alternate scripts
363         my $sort = $target =~ /__sort$/;
364         if ($sort && $altscript) {
365             next;
366         }
367
368         # Copy (scalar) data since can have multiple targets
369         # with differing options for (possibly) mutating data
370         # so need a different copy for each
371         my $_data = $data;
372         $record_document->{$target} //= [];
373         if (defined $options->{substr}) {
374             my ($start, $length) = @{$options->{substr}};
375             $_data = length($data) > $start ? substr $data, $start, $length : '';
376         }
377         if (defined $options->{value_callbacks}) {
378             $_data = reduce { $b->($a) } ($_data, @{$options->{value_callbacks}});
379         }
380         if (defined $options->{property}) {
381             $_data = {
382                 $options->{property} => $_data
383             }
384         }
385         push @{$record_document->{$target}}, $_data;
386     }
387 }
388
389 =head2 marc_records_to_documents($marc_records)
390
391     my @record_documents = $self->marc_records_to_documents($marc_records);
392
393 Using mappings stored in database convert C<$marc_records> to Elasticsearch documents.
394
395 Returns array of hash references, representing Elasticsearch documents,
396 acceptable as body payload in C<Search::Elasticsearch> requests.
397
398 =over 4
399
400 =item C<$marc_documents>
401
402 Reference to array of C<MARC::Record> objects to be converted to Elasticsearch documents.
403
404 =back
405
406 =cut
407
408 sub marc_records_to_documents {
409     my ($self, $records) = @_;
410     my $rules = $self->_get_marc_mapping_rules();
411     my $control_fields_rules = $rules->{control_fields};
412     my $data_fields_rules = $rules->{data_fields};
413     my $marcflavour = lc C4::Context->preference('marcflavour');
414
415     my @record_documents;
416
417     foreach my $record (@{$records}) {
418         my $record_document = {};
419         my $mappings = $rules->{leader};
420         if ($mappings) {
421             $self->_process_mappings($mappings, $record->leader(), $record_document, 0);
422         }
423         foreach my $field ($record->fields()) {
424             if ($field->is_control_field()) {
425                 my $mappings = $control_fields_rules->{$field->tag()};
426                 if ($mappings) {
427                     $self->_process_mappings($mappings, $field->data(), $record_document, 0);
428                 }
429             }
430             else {
431                 my $tag = $field->tag();
432                 # Handle alternate scripts in MARC 21
433                 my $altscript = 0;
434                 if ($marcflavour eq 'marc21' && $tag eq '880') {
435                     my $sub6 = $field->subfield('6');
436                     if ($sub6 =~ /^(...)-\d+/) {
437                         $tag = $1;
438                         $altscript = 1;
439                     }
440                 }
441
442                 my $data_field_rules = $data_fields_rules->{$tag};
443
444                 if ($data_field_rules) {
445                     my $subfields_mappings = $data_field_rules->{subfields};
446                     my $wildcard_mappings = $subfields_mappings->{'*'};
447                     foreach my $subfield ($field->subfields()) {
448                         my ($code, $data) = @{$subfield};
449                         my $mappings = $subfields_mappings->{$code} // [];
450                         if ($wildcard_mappings) {
451                             $mappings = [@{$mappings}, @{$wildcard_mappings}];
452                         }
453                         if (@{$mappings}) {
454                             $self->_process_mappings($mappings, $data, $record_document, $altscript);
455                         }
456                     }
457
458                     my $subfields_join_mappings = $data_field_rules->{subfields_join};
459                     if ($subfields_join_mappings) {
460                         foreach my $subfields_group (keys %{$subfields_join_mappings}) {
461                             # Map each subfield to values, remove empty values, join with space
462                             my $data = join(
463                                 ' ',
464                                 grep(
465                                     $_,
466                                     map { join(' ', $field->subfield($_)) } split(//, $subfields_group)
467                                 )
468                             );
469                             if ($data) {
470                                 $self->_process_mappings($subfields_join_mappings->{$subfields_group}, $data, $record_document, $altscript);
471                             }
472                         }
473                     }
474                 }
475             }
476         }
477         foreach my $field (keys %{$rules->{defaults}}) {
478             unless (defined $record_document->{$field}) {
479                 $record_document->{$field} = $rules->{defaults}->{$field};
480             }
481         }
482         foreach my $field (@{$rules->{sum}}) {
483             if (defined $record_document->{$field}) {
484                 # TODO: validate numeric? filter?
485                 # TODO: Or should only accept fields without nested values?
486                 # TODO: Quick and dirty, improve if needed
487                 $record_document->{$field} = sum0(grep { !ref($_) && m/\d+(\.\d+)?/} @{$record_document->{$field}});
488             }
489         }
490         # Index all applicable ISBN forms (ISBN-10 and ISBN-13 with and without dashes)
491         foreach my $field (@{$rules->{isbn}}) {
492             if (defined $record_document->{$field}) {
493                 my @isbns = ();
494                 foreach my $input_isbn (@{$record_document->{$field}}) {
495                     my $isbn = Business::ISBN->new($input_isbn);
496                     if (defined $isbn && $isbn->is_valid) {
497                         my $isbn13 = $isbn->as_isbn13->as_string;
498                         push @isbns, $isbn13;
499                         $isbn13 =~ s/\-//g;
500                         push @isbns, $isbn13;
501
502                         my $isbn10 = $isbn->as_isbn10;
503                         if ($isbn10) {
504                             $isbn10 = $isbn10->as_string;
505                             push @isbns, $isbn10;
506                             $isbn10 =~ s/\-//g;
507                             push @isbns, $isbn10;
508                         }
509                     } else {
510                         push @isbns, $input_isbn;
511                     }
512                 }
513                 $record_document->{$field} = \@isbns;
514             }
515         }
516
517         # Remove duplicate values and collapse sort fields
518         foreach my $field (keys %{$record_document}) {
519             if (ref($record_document->{$field}) eq 'ARRAY') {
520                 @{$record_document->{$field}} = do {
521                     my %seen;
522                     grep { !$seen{ref($_) eq 'HASH' && defined $_->{input} ? $_->{input} : $_}++ } @{$record_document->{$field}};
523                 };
524                 if ($field =~ /__sort$/) {
525                     # Make sure to keep the sort field length sensible. 255 was chosen as a nice round value.
526                     $record_document->{$field} = [substr(join(' ', @{$record_document->{$field}}), 0, 255)];
527                 }
528             }
529         }
530
531         # TODO: Perhaps should check if $records_document non empty, but really should never be the case
532         $record->encoding('UTF-8');
533         my @warnings;
534         {
535             # Temporarily intercept all warn signals (MARC::Record carps when record length > 99999)
536             local $SIG{__WARN__} = sub {
537                 push @warnings, $_[0];
538             };
539             $record_document->{'marc_data'} = encode_base64(encode('UTF-8', $record->as_usmarc()));
540         }
541         if (@warnings) {
542             # Suppress warnings if record length exceeded
543             unless (substr($record->leader(), 0, 5) eq '99999') {
544                 foreach my $warning (@warnings) {
545                     carp $warning;
546                 }
547             }
548             $record_document->{'marc_data'} = $record->as_xml_record($marcflavour);
549             $record_document->{'marc_format'} = 'MARCXML';
550         }
551         else {
552             $record_document->{'marc_format'} = 'base64ISO2709';
553         }
554         my $id = $record->subfield('999', 'c');
555         push @record_documents, [$id, $record_document];
556     }
557     return \@record_documents;
558 }
559
560 =head2 _field_mappings($facet, $suggestible, $sort, $target_name, $target_type, $range)
561
562     my @mappings = _field_mappings($facet, $suggestible, $sort, $target_name, $target_type, $range)
563
564 Get mappings, an internal data structure later used by
565 L<_process_mappings($mappings, $data, $record_document, $altscript)> to process MARC target
566 data for a MARC mapping.
567
568 The returned C<$mappings> is not to to be confused with mappings provided by
569 C<_foreach_mapping>, rather this sub accepts properties from a mapping as
570 provided by C<_foreach_mapping> and expands it to this internal data structure.
571 In the caller context (C<_get_marc_mapping_rules>) the returned C<@mappings>
572 is then applied to each MARC target (leader, control field data, subfield or
573 joined subfields) and integrated into the mapping rules data structure used in
574 C<marc_records_to_documents> to transform MARC records into Elasticsearch
575 documents.
576
577 =over 4
578
579 =item C<$facet>
580
581 Boolean indicating whether to create a facet field for this mapping.
582
583 =item C<$suggestible>
584
585 Boolean indicating whether to create a suggestion field for this mapping.
586
587 =item C<$sort>
588
589 Boolean indicating whether to create a sort field for this mapping.
590
591 =item C<$target_name>
592
593 Elasticsearch document target field name.
594
595 =item C<$target_type>
596
597 Elasticsearch document target field type.
598
599 =item C<$range>
600
601 An optional range as a string in the format "<START>-<END>" or "<START>",
602 where "<START>" and "<END>" are integers specifying a range that will be used
603 for extracting a substring from MARC data as Elasticsearch field target value.
604
605 The first character position is "1", and the range is inclusive,
606 so "1-3" means the first three characters of MARC data.
607
608 If only "<START>" is provided only one character at position "<START>" will
609 be extracted.
610
611 =back
612
613 =cut
614
615 sub _field_mappings {
616     my ($_self, $facet, $suggestible, $sort, $target_name, $target_type, $range) = @_;
617     my %mapping_defaults = ();
618     my @mappings;
619
620     my $substr_args = undef;
621     if ($range) {
622         # TODO: use value_callback instead?
623         my ($start, $end) = map(int, split /-/, $range, 2);
624         $substr_args = [$start];
625         push @{$substr_args}, (defined $end ? $end - $start + 1 : 1);
626     }
627     my $default_options = {};
628     if ($substr_args) {
629         $default_options->{substr} = $substr_args;
630     }
631
632     # TODO: Should probably have per type value callback/hook
633     # but hard code for now
634     if ($target_type eq 'boolean') {
635         $default_options->{value_callbacks} //= [];
636         push @{$default_options->{value_callbacks}}, sub {
637             my ($value) = @_;
638             # Trim whitespace at both ends
639             $value =~ s/^\s+|\s+$//g;
640             return $value ? 'true' : 'false';
641         };
642     }
643
644     my $mapping = [$target_name, $default_options];
645     push @mappings, $mapping;
646
647     my @suffixes = ();
648     push @suffixes, 'facet' if $facet;
649     push @suffixes, 'suggestion' if $suggestible;
650     push @suffixes, 'sort' if !defined $sort || $sort;
651
652     foreach my $suffix (@suffixes) {
653         my $mapping = ["${target_name}__$suffix"];
654         # TODO: Hack, fix later in less hideous manner
655         if ($suffix eq 'suggestion') {
656             push @{$mapping}, {%{$default_options}, property => 'input'};
657         }
658         else {
659             push @{$mapping}, $default_options;
660         }
661         push @mappings, $mapping;
662     }
663     return @mappings;
664 };
665
666 =head2 _get_marc_mapping_rules
667
668     my $mapping_rules = $self->_get_marc_mapping_rules()
669
670 Generates rules from mappings stored in database for MARC records to Elasticsearch JSON document conversion.
671
672 Since field retrieval is slow in C<MARC::Records> (all fields are itereted through for
673 each call to C<MARC::Record>->field) we create an optimized structure of mapping
674 rules keyed by MARC field tags holding all the mapping rules for that particular tag.
675
676 We can then iterate through all MARC fields for each record and apply all relevant
677 rules once per fields instead of retreiving fields multiple times for each mapping rule
678 which is terribly slow.
679
680 =cut
681
682 # TODO: This structure can be used for processing multiple MARC::Records so is currently
683 # rebuilt for each batch. Since it is cacheable it could also be stored in an in
684 # memory cache which it is currently not. The performance gain of caching
685 # would probably be marginal, but to do this could be a further improvement.
686
687 sub _get_marc_mapping_rules {
688     my ($self) = @_;
689     my $marcflavour = lc C4::Context->preference('marcflavour');
690     my $field_spec_regexp = qr/^([0-9]{3})([()0-9a-z]+)?(?:_\/(\d+(?:-\d+)?))?$/;
691     my $leader_regexp = qr/^leader(?:_\/(\d+(?:-\d+)?))?$/;
692     my $rules = {
693         'leader' => [],
694         'control_fields' => {},
695         'data_fields' => {},
696         'sum' => [],
697         'isbn' => [],
698         'defaults' => {}
699     };
700
701     $self->_foreach_mapping(sub {
702         my ($name, $type, $facet, $suggestible, $sort, $marc_type, $marc_field) = @_;
703         return if $marc_type ne $marcflavour;
704
705         if ($type eq 'sum') {
706             push @{$rules->{sum}}, $name;
707         }
708         elsif ($type eq 'isbn') {
709             push @{$rules->{isbn}}, $name;
710         }
711         elsif ($type eq 'boolean') {
712             # boolean gets special handling, if value doesn't exist for a field,
713             # it is set to false
714             $rules->{defaults}->{$name} = 'false';
715         }
716
717         if ($marc_field =~ $field_spec_regexp) {
718             my $field_tag = $1;
719
720             my @subfields;
721             my @subfield_groups;
722             # Parse and separate subfields form subfield groups
723             if (defined $2) {
724                 my $subfield_group = '';
725                 my $open_group = 0;
726
727                 foreach my $token (split //, $2) {
728                     if ($token eq "(") {
729                         if ($open_group) {
730                             Koha::Exceptions::Elasticsearch::MARCFieldExprParseError->throw(
731                                 "Unmatched opening parenthesis for $marc_field"
732                             );
733                         }
734                         else {
735                             $open_group = 1;
736                         }
737                     }
738                     elsif ($token eq ")") {
739                         if ($open_group) {
740                             if ($subfield_group) {
741                                 push @subfield_groups, $subfield_group;
742                                 $subfield_group = '';
743                             }
744                             $open_group = 0;
745                         }
746                         else {
747                             Koha::Exceptions::Elasticsearch::MARCFieldExprParseError->throw(
748                                 "Unmatched closing parenthesis for $marc_field"
749                             );
750                         }
751                     }
752                     elsif ($open_group) {
753                         $subfield_group .= $token;
754                     }
755                     else {
756                         push @subfields, $token;
757                     }
758                 }
759             }
760             else {
761                 push @subfields, '*';
762             }
763
764             my $range = defined $3 ? $3 : undef;
765             my @mappings = $self->_field_mappings($facet, $suggestible, $sort, $name, $type, $range);
766
767             if ($field_tag < 10) {
768                 $rules->{control_fields}->{$field_tag} //= [];
769                 push @{$rules->{control_fields}->{$field_tag}}, @mappings;
770             }
771             else {
772                 $rules->{data_fields}->{$field_tag} //= {};
773                 foreach my $subfield (@subfields) {
774                     $rules->{data_fields}->{$field_tag}->{subfields}->{$subfield} //= [];
775                     push @{$rules->{data_fields}->{$field_tag}->{subfields}->{$subfield}}, @mappings;
776                 }
777                 foreach my $subfield_group (@subfield_groups) {
778                     $rules->{data_fields}->{$field_tag}->{subfields_join}->{$subfield_group} //= [];
779                     push @{$rules->{data_fields}->{$field_tag}->{subfields_join}->{$subfield_group}}, @mappings;
780                 }
781             }
782         }
783         elsif ($marc_field =~ $leader_regexp) {
784             my $range = defined $1 ? $1 : undef;
785             my @mappings = $self->_field_mappings($facet, $suggestible, $sort, $name, $type, $range);
786             push @{$rules->{leader}}, @mappings;
787         }
788         else {
789             Koha::Exceptions::Elasticsearch::MARCFieldExprParseError->throw(
790                 "Invalid MARC field expression: $marc_field"
791             );
792         }
793     });
794     return $rules;
795 }
796
797 =head2 _foreach_mapping
798
799     $self->_foreach_mapping(
800         sub {
801             my ( $name, $type, $facet, $suggestible, $sort, $marc_type,
802                 $marc_field )
803               = @_;
804             return unless $marc_type eq 'marc21';
805             print "Data comes from: " . $marc_field . "\n";
806         }
807     );
808
809 This allows you to apply a function to each entry in the elasticsearch mappings
810 table, in order to build the mappings for whatever is needed.
811
812 In the provided function, the files are:
813
814 =over 4
815
816 =item C<$name>
817
818 The field name for elasticsearch (corresponds to the 'mapping' column in the
819 database.
820
821 =item C<$type>
822
823 The type for this value, e.g. 'string'.
824
825 =item C<$facet>
826
827 True if this value should be facetised. This only really makes sense if the
828 field is understood by the facet processing code anyway.
829
830 =item C<$sort>
831
832 True if this is a field that a) needs special sort handling, and b) if it
833 should be sorted on. False if a) but not b). Undef if not a). This allows,
834 for example, author to be sorted on but not everything marked with "author"
835 to be included in that sort.
836
837 =item C<$marc_type>
838
839 A string that indicates the MARC type that this mapping is for, e.g. 'marc21',
840 'unimarc', 'normarc'.
841
842 =item C<$marc_field>
843
844 A string that describes the MARC field that contains the data to extract.
845 These are of a form suited to Catmandu's MARC fixers.
846
847 =back
848
849 =cut
850
851 sub _foreach_mapping {
852     my ( $self, $sub ) = @_;
853
854     # TODO use a caching framework here
855     my $search_fields = Koha::Database->schema->resultset('SearchField')->search(
856         {
857             'search_marc_map.index_name' => $self->index,
858         },
859         {   join => { search_marc_to_fields => 'search_marc_map' },
860             '+select' => [
861                 'search_marc_to_fields.facet',
862                 'search_marc_to_fields.suggestible',
863                 'search_marc_to_fields.sort',
864                 'search_marc_map.marc_type',
865                 'search_marc_map.marc_field',
866             ],
867             '+as'     => [
868                 'facet',
869                 'suggestible',
870                 'sort',
871                 'marc_type',
872                 'marc_field',
873             ],
874         }
875     );
876
877     while ( my $search_field = $search_fields->next ) {
878         $sub->(
879             # Force lower case on indexed field names for case insensitive
880             # field name searches
881             lc($search_field->name),
882             $search_field->type,
883             $search_field->get_column('facet'),
884             $search_field->get_column('suggestible'),
885             $search_field->get_column('sort'),
886             $search_field->get_column('marc_type'),
887             $search_field->get_column('marc_field'),
888         );
889     }
890 }
891
892 =head2 process_error
893
894     die process_error($@);
895
896 This parses an Elasticsearch error message and produces a human-readable
897 result from it. This result is probably missing all the useful information
898 that you might want in diagnosing an issue, so the warning is also logged.
899
900 Note that currently the resulting message is not internationalised. This
901 will happen eventually by some method or other.
902
903 =cut
904
905 sub process_error {
906     my ($self, $msg) = @_;
907
908     warn $msg; # simple logging
909
910     # This is super-primitive
911     return "Unable to understand your search query, please rephrase and try again.\n" if $msg =~ /ParseException/;
912
913     return "Unable to perform your search. Please try again.\n";
914 }
915
916 =head2 _read_configuration
917
918     my $conf = _read_configuration();
919
920 Reads the I<configuration file> and returns a hash structure with the
921 configuration information. It raises an exception if mandatory entries
922 are missing.
923
924 The hashref structure has the following form:
925
926     {
927         'nodes' => ['127.0.0.1:9200', 'anotherserver:9200'],
928         'index_name' => 'koha_instance',
929     }
930
931 This is configured by the following in the C<config> block in koha-conf.xml:
932
933     <elasticsearch>
934         <server>127.0.0.1:9200</server>
935         <server>anotherserver:9200</server>
936         <index_name>koha_instance</index_name>
937     </elasticsearch>
938
939 =cut
940
941 sub _read_configuration {
942
943     my $configuration;
944
945     my $conf = C4::Context->config('elasticsearch');
946     Koha::Exceptions::Config::MissingEntry->throw(
947         "Missing 'elasticsearch' block in config file")
948       unless defined $conf;
949
950     if ( $conf && $conf->{server} ) {
951         my $nodes = $conf->{server};
952         if ( ref($nodes) eq 'ARRAY' ) {
953             $configuration->{nodes} = $nodes;
954         }
955         else {
956             $configuration->{nodes} = [$nodes];
957         }
958     }
959     else {
960         Koha::Exceptions::Config::MissingEntry->throw(
961             "Missing 'server' entry in config file for elasticsearch");
962     }
963
964     if ( defined $conf->{index_name} ) {
965         $configuration->{index_name} = $conf->{index_name};
966     }
967     else {
968         Koha::Exceptions::Config::MissingEntry->throw(
969             "Missing 'index_name' entry in config file for elasticsearch");
970     }
971
972     return $configuration;
973 }
974
975 =head2
976
977 my @facetable_fields = Koha::SearchEngine::Elasticsearch->get_facetable_fields();
978
979 Returns the list of Koha::SearchFields marked to be faceted in the ES configuration
980
981 =cut
982
983 sub get_facetable_fields {
984     my ($self) = @_;
985
986     # These should correspond to the ES field names, as opposed to the CCL
987     # things that zebra uses.
988     my @search_field_names = qw( author itype location su-geo se subject ccode holdingbranch homebranch );
989     my @faceted_fields = Koha::SearchFields->search(
990         { name => { -in => \@search_field_names }, facet_order => { '!=' => undef } }, { order_by => ['facet_order'] }
991     );
992     my @not_faceted_fields = Koha::SearchFields->search(
993         { name => { -in => \@search_field_names }, facet_order => undef }, { order_by => ['facet_order'] }
994     );
995     # This could certainly be improved
996     return ( @faceted_fields, @not_faceted_fields );
997 }
998
999 1;
1000
1001 __END__
1002
1003 =head1 AUTHOR
1004
1005 =over 4
1006
1007 =item Chris Cormack C<< <chrisc@catalyst.net.nz> >>
1008
1009 =item Robin Sheat C<< <robin@catalyst.net.nz> >>
1010
1011 =item Jonathan Druart C<< <jonathan.druart@bugs.koha-community.org> >>
1012
1013 =back
1014
1015 =cut