example search for Koha database
[webpac2] / run.pl
diff --git a/run.pl b/run.pl
index 4299b6e..ea744e9 100755 (executable)
--- a/run.pl
+++ b/run.pl
@@ -7,13 +7,12 @@ use File::Temp qw/tempdir/;
 use lib './lib';
 
 use WebPAC::Common 0.02;
-use WebPAC::Parser 0.04;
-use WebPAC::Lookup 0.03;
-use WebPAC::Input 0.11;
-use WebPAC::Store 0.03;
-use WebPAC::Normalize 0.11;
-use WebPAC::Output::TT;
-use WebPAC::Validate 0.06;
+use WebPAC::Parser 0.08;
+use WebPAC::Input 0.16;
+use WebPAC::Store 0.15;
+use WebPAC::Normalize 0.22;
+#use WebPAC::Output::TT;
+use WebPAC::Validate 0.11;
 use WebPAC::Output::MARC;
 use WebPAC::Config;
 use Getopt::Long;
@@ -22,8 +21,9 @@ use Time::HiRes qw/time/;
 use File::Slurp;
 use Data::Dump qw/dump/;
 use Storable qw/dclone/;
+use Pod::Usage qw/pod2usage/;
+use LWP::Simple qw//;
 
-use Proc::Queue size => 1;
 use POSIX ":sys_wait_h"; # imports WNOHANG
 
 =head1 NAME
@@ -32,7 +32,7 @@ run.pl - start WebPAC indexing
 
 B<this command will probably go away. Don't get used to it!>
 
-Options:
+=head1 OPTIONS
 
 =over 4
 
@@ -68,17 +68,22 @@ and subfield usage for each input
 
 turn on extra validation of imput records, see L<WebPAC::Validation>
 
-=item --marc-normalize conf/normalize/mapping.pl
+You can use special variables C<$database> and $C<$input> in this parametar
+like C<--validate 'conf/validate/$database-$input'> to construct filename
 
-This option specifies normalisation file for MARC creation
+=item --validate-delimiters path/to/validate_delimiters_file
 
-=item --marc-output out/marc/test.marc
+this option is used with C<--validate> to turn on extra validation of
+delimiters. If file is non existant, it will be created on first run.
 
-Optional path to output file
+=item --marc-generate
+
+Generate MARC file. This will automatically be on if file contains C<marc*> directives.
+You can use this option as C<--no-marc-generate> to disable MARC generation.
 
 =item --marc-lint
 
-By default turned on if C<--marc-normalize> is used. You can disable lint
+By default turned on if normalisation file has C<marc*> directives. You can disable lint
 messages with C<--no-marc-lint>.
 
 =item --marc-dump
@@ -98,6 +103,10 @@ Create just links
 
 Create merged index of databases which have links
 
+=item --mirror http://www.example.com
+
+Tries to download input path files from mirror URI
+
 =back
 
 =cut
@@ -111,12 +120,15 @@ my $debug = 0;
 my $only_filter;
 my $stats = 0;
 my $validate_path;
-my ($marc_normalize, $marc_output);
+my $validate_delimiters_path;
+my $marc_generate = 1;
 my $marc_lint = 1;
 my $marc_dump = 0;
 my $parallel = 0;
 my $only_links = 0;
 my $merge = 0;
+my $mirror;
+my $help;
 
 my $log = _new WebPAC::Common()->_get_logger();
 
@@ -126,27 +138,41 @@ GetOptions(
        "clean" => \$clean,
        "one=s" => \$only_filter,
        "only=s" => \$only_filter,
-       "config" => \$config_path,
+       "config=s" => \$config_path,
        "debug+" => \$debug,
        "stats" => \$stats,
        "validate=s" => \$validate_path,
-       "marc-normalize=s" => \$marc_normalize,
-       "marc-output=s" => \$marc_output,
+       "validate-delimiters=s" => \$validate_delimiters_path,
+       "marc-generate!" => \$marc_generate,
        "marc-lint!" => \$marc_lint,
        "marc-dump!" => \$marc_dump,
        "parallel=i" => \$parallel,
        "only-links!" => \$only_links,
        "merge" => \$merge,
+       "mirror=s" => \$mirror,
+       "help" => \$help,
 );
 
+$marc_generate = 0 if ( $validate_delimiters_path );
+
+pod2usage(-verbose => 2) if ($help);
+
 my $config = new WebPAC::Config( path => $config_path );
 
+WebPAC::Normalize::_debug( $debug - 1 ) if $debug > 1;
+
 #print "config = ",dump($config) if ($debug);
 
 die "no databases in config file!\n" unless ($config->databases);
 
 $log->info( "-" x 79 );
 
+my $log_file = 'log';
+
+if (-e $log_file ) {   # && -s $log_file > 5 * 1024 * 1024) {
+       $log->info("moved old log with ", -s $log_file, " bytes to '${log_file}.old'");
+       rename $log_file, "${log_file}.old" || $log->logwarn("can't rename $log_file to ${log_file}.old: $!");
+}
 
 my $estcmd_fh;
 my $estcmd_path = './estcmd-merge.sh';
@@ -157,27 +183,29 @@ if ($merge) {
        $log->info("created merge batch file $estcmd_path");
 }
 
-
 my $validate;
 $validate = new WebPAC::Validate(
-       path => $validate_path,
-) if ($validate_path);
-
+       delimiters => $config->webpac('delimiters'),
+) if ($validate_path || $validate_delimiters_path);
 
 my $use_indexer = $config->use_indexer;
+$stats ||= $validate;
 if ($stats) {
-       $log->debug("option --stats disables update of indexing engine...");
+       $log->debug("disabled indexing for stats collection");
        $use_indexer = undef;
-} else {
+} elsif ( $use_indexer ) {
        $log->info("using $use_indexer indexing engine...");
 }
 
-# disable indexing when creating marc
-$use_indexer = undef if ($marc_normalize);
-
 # parse normalize files and create source files for lookup and normalization
 
-my $parser = new WebPAC::Parser( config => $config );
+my ($only_database,$only_input) = split(m#/#, $only_filter) if $only_filter;
+
+my $parser = new WebPAC::Parser(
+       config => $config,
+       only_database => $only_database,
+       only_input => $only_input,
+);
 
 my $total_rows = 0;
 my $start_t = time();
@@ -185,13 +213,24 @@ my $start_t = time();
 my @links;
 
 if ($parallel) {
+       eval 'use Proc::Queue size => 1;';
+       die $@ if $@;
        $log->info("Using $parallel processes for speedup");
        Proc::Queue::size($parallel);
 }
 
-while (my ($database, $db_config) = each %{ $config->databases }) {
+sub create_ds_config {
+       my ($db_config, $database, $input, $mfn) = @_;
+       my $c = dclone( $db_config );
+       $c->{_} = $database || $log->logconfess("need database");
+       $c->{_mfn} = $mfn || $log->logconfess("need mfn");
+       $c->{input} = $input || $log->logconfess("need input");
+       return $c;
+}
+
+foreach my $database ( sort keys %{ $config->databases } ) {
+       my $db_config = $config->databases->{$database};
 
-       my ($only_database,$only_input) = split(m#/#, $only_filter) if ($only_filter);
        next if ($only_database && $database !~ m/$only_database/i);
 
        if ($parallel) {
@@ -204,7 +243,7 @@ while (my ($database, $db_config) = each %{ $config->databases }) {
        }
 
        my $indexer;
-       if ($use_indexer) {
+       if ($use_indexer && $parser->have_rules( 'search', $database )) {
 
                my $cfg_name = $use_indexer;
                $cfg_name =~ s/\-.*$//;
@@ -220,27 +259,24 @@ while (my ($database, $db_config) = each %{ $config->databases }) {
                if ($use_indexer eq 'hyperestraier') {
 
                        # open Hyper Estraier database
-                       use WebPAC::Output::Estraier '0.10';
+                       require WebPAC::Output::Estraier;
                        $indexer = new WebPAC::Output::Estraier( %{ $indexer_config } );
                
                } elsif ($use_indexer eq 'hyperestraier-native') {
 
                        # open Hyper Estraier database
-                       use WebPAC::Output::EstraierNative;
+                       require WebPAC::Output::EstraierNative;
                        $indexer = new WebPAC::Output::EstraierNative( %{ $indexer_config } );
 
                } elsif ($use_indexer eq 'kinosearch') {
 
-                       # open KinoSearch
-                       use WebPAC::Output::KinoSearch;
-                       $indexer_config->{clean} = 1 unless (-e $indexer_config->{index_path});
-                       $indexer = new WebPAC::Output::KinoSearch( %{ $indexer_config } );
+                       die "no longer supported";
 
                } else {
                        $log->logdie("unknown use_indexer: $use_indexer");
                }
 
-               $log->logide("can't continue without valid indexer") unless ($indexer);
+               $log->logdie("can't continue without valid indexer") unless ($indexer);
        }
 
 
@@ -274,50 +310,98 @@ while (my ($database, $db_config) = each %{ $config->databases }) {
        #
        # now WebPAC::Store
        #
-       my $abs_path = abs_path($0);
-       $abs_path =~ s#/[^/]*$#/#;
+       my $store = new WebPAC::Store({
+               debug => $debug,
+       });
 
-       my $db_path = $config->get('webpac')->{db_path} . '/' . $database;
 
-       if ($clean) {
-               $log->info("creating new database '$database' in $db_path");
-               rmtree( $db_path ) || $log->warn("can't remove $db_path: $!");
-       } else {
-               $log->info("working on database '$database' in $db_path");
-       }
+       #
+       # prepare output
+       #
+       my @outputs = force_array( $db_config->{output}, sub {
+               $log->error("Database $database doesn't have any outputs defined. Do you want to remove it from configuration?" );
+       } );
 
-       my $db = new WebPAC::Store(
-               path => $db_path,
-               database => $database,
-               debug => $debug,
-       );
+       my @output_modules;
+
+       foreach my $output ( @outputs ) {
+
+#warn '## output = ',dump( $output );
+
+               my $module = $output->{module} || $log->logdie("need module in output section of $database");
+               $module = 'WebPAC::Output::' . $module unless $module =~ m/::/;
+       
+               $log->debug("loading output module $module");
+               eval "require $module";
+
+               # add database to arugemnts for output filter
+               $output->{database} = $database;
+               $output->{clean} = $clean;
+
+               $log->debug("calling $module->new(",dump( $output ),")");
+               my $out = new $module->new( $output );
+               if ( $out->init ) {
+                       push @output_modules, $out;
+               } else {
+                       $log->warn("SKIPPED $module");
+               }
+       }
 
 
        #
        # now, iterate through input formats
        #
 
-       my @inputs;
-       if (ref($db_config->{input}) eq 'ARRAY') {
-               @inputs = @{ $db_config->{input} };
-       } elsif ($db_config->{input}) {
-               push @inputs, $db_config->{input};
-       } else {
+
+       my @inputs = force_array( $db_config->{input}, sub {
                $log->info("database $database doesn't have inputs defined");
-       }
+       } );
+
+       if ( -e 'out/debug' ) { # FIXME flag?
+               my $out;
+               foreach my $i ( @inputs ) {
+                       warn dump( $i );
+                       next unless defined $i->{normalize};
+                       warn dump( $i->{normalize} );
+                       foreach my $normalize ( @{ $i->{normalize} } ) {
+                               my $path = $normalize->{path};
+                               $out .= qq/\n##\n## $path\n##\n\n/;
+                               $out .= read_file( $path );
+                       }
+               }
+               my $all = "out/debug/all-normalize.pl";
+               write_file( $all, $out );
+               warn "### all normalize for this input saved to: $all";
+       };
 
        foreach my $input (@inputs) {
 
-               next if ($only_input && ($input->{name} !~ m#$only_input#i && $input->{type} !~ m#$only_input#i));
+               my $input_name = $input->{name} || $log->logdie("input without a name isn't valid: ",dump($input));
+
+               if ( $input->{skip} ) {
+                       $log->info("skip $input_name");
+                       next;
+               }
+
+               next if ($only_input && ($input_name !~ m#$only_input#i && $input->{type} !~ m#$only_input#i));
 
                my $type = lc($input->{type});
 
-               die "I know only how to handle input types ", join(",", $config->webpac('inputs') ), " not '$type'!\n" unless (grep(/$type/, $config->webpac('inputs')));
+               # FIXME check if input module exists
+               my $input_module = $input->{module};
 
-               my $input_module = $config->webpac('inputs')->{$type};
+               if ( ! $input_module ) {
+                       if ( grep(/$type/, $config->webpac('inputs')) ) {
+                               $input_module = $config->webpac('inputs')->{$type};
+                       } else {
+                               $log->logdie("I know only how to handle input types ", join(",", $config->webpac('inputs') ), " not '$type'!" );
+                       }
+               }
+
+               my @lookups = $parser->have_lookup_create($database, $input);
 
-               $log->info("working on input '$input->{name}' in $input->{path} [type: $input->{type}] using $input_module",
-                       $input->{lookup} ? "lookup '$input->{lookup}'" : ""
+               $log->info("working on $database/$input_name with $input_module on $input->{path}",
+                       @lookups ? " creating lookups: ".join(", ", @lookups) : ""
                );
 
                if ($stats) {
@@ -326,173 +410,312 @@ while (my ($database, $db_config) = each %{ $config->databases }) {
                        delete($input->{modify_file});
                }
 
-               warn "parser->depends = ", dump( $parser->{depends} );
-               warn "depends on: ", dump( $parser->depends($database, $input->{name}) );
-               warn "lookup_create_rules = ", dump( $parser->lookup_create_rules($database, $input->{name}) );
-               warn "parser->_lookup_create = ", dump( $parser->{_lookup_create} );
+               if ( $mirror ) {
+                       my $path = $input->{path} || die "no input path in ",dump( $input );
 
-               my $lookup;
+                       my $base = $path;
+                       $base =~ s{/[^/]+$}{};
+                       mkpath $base unless -e $base;
+
+                       my $rc = LWP::Simple::mirror( "$mirror/$path", $path );
+                       if (LWP::Simple::is_error( $rc )) {
+                               die "can't mirror $mirror/$path -> $path [$rc]";
+                       } else {
+                               $log->info( "mirror ", $path, " [$rc] ", -s $path, " bytes" );
+                       }
+                       
+               }
 
                my $input_db = new WebPAC::Input(
                        module => $input_module,
-                       encoding => $config->webpac('webpac_encoding'),
                        limit => $limit || $input->{limit},
                        offset => $offset,
-                       lookup_coderef => sub {
-                               my $rec = shift || return;
-                               $lookup->add( $rec );
-                       },
                        recode => $input->{recode},
                        stats => $stats,
                        modify_records => $input->{modify_records},
                        modify_file => $input->{modify_file},
+                       input_config => $input,
                );
                $log->logdie("can't create input using $input_module") unless ($input);
 
+               if (defined( $input->{lookup} )) {
+                       $log->warn("$database/$input_name has depriciated lookup definition, removing it...");
+                       delete( $input->{lookup} );
+               }
+
+               my $lookup_coderef;
+
+               if (@lookups) {
+
+                       my $rules = $parser->lookup_create_rules($database, $input) || $log->logdie("no rules found for $database/$input");
+
+                       $lookup_coderef = sub {
+                               my $rec = shift || die "need rec!";
+                               my $mfn = $rec->{'000'}->[0] || die "need mfn in 000";
+
+                               WebPAC::Normalize::data_structure(
+                                       row => $rec,
+                                       rules => $rules,
+                                       config => create_ds_config( $db_config, $database, $input, $mfn ),
+                               );
+
+                               #warn "current lookup: ", dump(WebPAC::Normalize::_get_lookup());
+                       };
+
+                       WebPAC::Normalize::_set_lookup( undef );
+
+                       $log->debug("created lookup_coderef using:\n$rules");
+
+               };
+
+               my $lookup_jar;
+
                my $maxmfn = $input_db->open(
                        path => $input->{path},
-                       code_page => $input->{encoding},        # database encoding
+                       input_encoding => $input->{encoding},   # database encoding
+                       lookup_coderef => $lookup_coderef,
+                       lookup => $lookup_jar,
                        %{ $input },
+                       load_row => sub {
+                               my $a = shift;
+                               return $store->load_row(
+                                       database => $database,
+                                       input => $input_name,
+                                       id => $a->{id},
+                               );
+                       },
+                       save_row => sub {
+                               my $a = shift;
+                               return $store->save_row(
+                                       database => $database,
+                                       input => $input_name,
+                                       id => $a->{id},
+                                       row => $a->{row},
+                               );
+                       },
+
                );
 
+               my $lookup_data = WebPAC::Normalize::_get_lookup();
+
+               if (defined( $lookup_data->{$database}->{$input_name} )) {
+                       $log->debug("created following lookups: ", sub { dump( $lookup_data ) } );
+
+                       foreach my $key (keys %{ $lookup_data->{$database}->{$input_name} }) {
+                               $store->save_lookup(
+                                       database => $database,
+                                       input => $input_name,
+                                       key => $key,
+                                       data => $lookup_data->{$database}->{$input_name}->{$key},
+                               );
+                       }
+               }
+
                my $report_fh;
                if ($stats || $validate) {
-                       my $path = "out/report/" . $database . '-' . $input->{name} . '.txt';
+                       my $out_report = 'out/report'; # FIXME move to config
+                       mkpath $out_report unless -e $out_report;
+                       my $path = "$out_report/${database}-${input_name}.txt";
                        open($report_fh, '>', $path) || $log->logdie("can't open $path: $!");
 
-                       print $report_fh "Report for database '$database' input '$input->{name}' records ",
+                       print $report_fh "Report for database '$database' input '$input_name' records ",
                                $offset || 1, "-", $limit || $input->{limit} || $maxmfn, "\n\n";
                        $log->info("Generating report file $path");
+
+                       if ( $validate ) {
+                               $validate->read_validate_file( $validate->fill_in( $validate_path, database => $database, input => $input_name ) ) if ( $validate_path );
+                               $validate->read_validate_delimiters_file( $validate->fill_in( $validate_delimiters_path, database => $database, input => $input_name ) ) if ( $validate_delimiters_path );
+                       }
                }
 
-               my @norm_array = ref($input->{normalize}) eq 'ARRAY' ?
-                       @{ $input->{normalize} } : ( $input->{normalize} );
+               my $marc;
+               if ($marc_generate && $parser->have_rules( 'marc', $database, $input_name )) {
+                       
+                       my $out_marc = 'out/marc'; # FIXME move to config
+                       mkpath $out_marc unless -e $out_marc;
 
-               if ($marc_normalize) {
-                       @norm_array = ( {
-                               path => $marc_normalize,
-                               output => $marc_output || 'out/marc/' . $database . '-' . $input->{name} . '.marc',
-                       );
+                       $marc = new WebPAC::Output::MARC(
+                               path => "$out_marc/${database}-${input_name}.marc",
+                               lint => $marc_lint,
+                               dump => $marc_dump,
+                       );
                }
 
-               foreach my $normalize (@norm_array) {
-
-                       my $normalize_path = $normalize->{path} || $log->logdie("can't find normalize path in config");
+               my $rules = $parser->normalize_rules($database,$input_name);
+               if ( ! $rules ) {
+                       $log->logwarn("no normalize rules for $database/$input_name", $input_db->input_module->can('normalize') ? " using normalize from input module" : '');
+                       next;
+               }
 
-                       $log->logdie("Found '$normalize_path' as normalization file which isn't supported any more!") unless ( $normalize_path =~ m!\.pl$!i );
+               $log->debug("parsed normalize rules:\n$rules");
+
+               # reset position in database
+               $input_db->seek(1);
+
+               # generate name of config key for indexer (strip everything after -)
+               my $indexer_config = $use_indexer;
+               $indexer_config =~ s/^(\w+)-?.*$/$1/g if ($indexer_config);
+
+               my $lookup_hash;
+               my $depends = $parser->depends($database,$input_name);
+       
+               if ($depends) {
+                       $log->debug("$database/$input_name depends on: ", dump($depends)) if ($depends);
+                       $log->logdie("parser->depends didn't return HASH") unless (ref($depends) eq 'HASH');
+
+                       foreach my $db (keys %$depends) {
+                               foreach my $i (keys %{$depends->{$db}}) {
+                                       foreach my $k (keys %{$depends->{$db}->{$i}}) {
+                                               my $t = time();
+                                               $log->debug("loading lookup $db/$i");
+                                               $lookup_hash->{$db}->{$i}->{$k} = $store->load_lookup(
+                                                       database => $db,
+                                                       input => $i,
+                                                       key => $k,
+                                               );
+                                               $log->debug(sprintf("lookup $db/$i took %.2fs", time() - $t));
+                                       }
+                               }
+                       }
 
-                       my $rules = read_file( $normalize_path ) or die "can't open $normalize_path: $!";
+                       $log->debug("lookup_hash = ", sub { dump( $lookup_hash ) });
+               }
 
-                       $log->info("Using $normalize_path for normalization...");
 
-                       my $marc = new WebPAC::Output::MARC(
-                               path => $normalize->{output},
-                               lint => $marc_lint,
-                               dump => $marc_dump,
-                       ) if ($normalize->{output});
+               # setup input name for all output filters
+               foreach my $out ( @output_modules ) {
+                       if ( $out->can('input') ) {
+                               $out->input( $input_name );
+                       } else {
+                               $log->warn("output filter ",ref($out)," doesn't support input name");
+                       }
+               }
 
-                       # reset position in database
-                       $input_db->seek(1);
 
-                       # generate name of config key for indexer (strip everything after -)
-                       my $indexer_config = $use_indexer;
-                       $indexer_config =~ s/^(\w+)-?.*$/$1/g if ($indexer_config);
+               foreach my $pos ( 0 ... $input_db->size ) {
 
-                       foreach my $pos ( 0 ... $input_db->size ) {
+                       my $row = $input_db->fetch || next;
 
-                               my $row = $input_db->fetch || next;
+                       $total_rows++;
 
-                               my $mfn = $row->{'000'}->[0];
+                       my $mfn = $row->{'000'}->[0];
 
-                               if (! $mfn || $mfn !~ m#^\d+$#) {
-                                       $log->warn("record $pos doesn't have valid MFN but '$mfn', using $pos");
-                                       $mfn = $pos;
-                                       push @{ $row->{'000'} }, $pos;
-                               }
+                       if (! $mfn || $mfn !~ m{^\d+$}) {
+                               $log->warn("record $pos doesn't have valid MFN but '$mfn', using $pos");
+                               $mfn = $pos;
+                               push @{ $row->{'000'} }, $pos;
+                       }
 
+                       foreach my $out ( @output_modules ) {
+                               $out->add_row( $mfn, $row ) if $out->can('add_row');
+                       }
 
-                               if ($validate) {
-                                       if ( my $errors = $validate->validate_errors( $row, $input_db->dump ) ) {
-                                               $log->error( "MFN $mfn validation error:\n",
-                                                       $validate->report_error( $errors )
-                                               );
-                                       }
+                       if ($validate) {
+                               if ( my $errors = $validate->validate_rec( $row, $input_db->dump_ascii ) ) {
+                                       $log->error( "MFN $mfn validation error:\n",
+                                               $validate->report_error( $errors )
+                                       );
                                }
+                               next;   # validation doesn't create any output
+                       }
 
-                               my $ds_config = dclone($db_config);
-
-                               # default values -> database key
-                               $ds_config->{_} = $database;
-
-                               # current mfn
-                               $ds_config->{_mfn} = $mfn;
+                       my $ds;
 
-                               # attach current input
-                               $ds_config->{input} = $input;
+                       if ($rules) {
 
-                               my $ds = WebPAC::Normalize::data_structure(
+                               $ds = WebPAC::Normalize::data_structure(
                                        row => $row,
                                        rules => $rules,
-                                       lookup => $lookup ? $lookup->lookup_hash : undef,
-                                       config => $ds_config,
+                                       lookup => $lookup_hash,
+                                       config => create_ds_config( $db_config, $database, $input, $mfn ),
                                        marc_encoding => 'utf-8',
+                                       load_row_coderef => sub {
+                                               my ($database,$input,$mfn) = @_;
+#warn "### load_row($database,$input,$mfn) from data_structure\n";
+                                               return $store->load_row(
+                                                       database => $database,
+                                                       input => $input,
+                                                       id => $mfn,
+                                               );
+                                       },
                                );
 
-                               $db->save_ds(
+                       } elsif ( $input_db->input_module->can('normalize') ) {
+                               $ds = $input_db->input_module->normalize( $mfn );
+                       }
+
+                       if ( $ds ) {
+                               $log->debug("ds = ", sub { dump($ds) });
+
+                               $store->save_ds(
+                                       database => $database,
+                                       input => $input_name,
                                        id => $mfn,
                                        ds => $ds,
-                                       prefix => $input->{name},
-                               ) if ($ds && !$stats);
+                               ) if !$stats;
 
                                $indexer->add(
-                                       id => $input->{name} . "/" . $mfn,
+                                       id => "${input_name}/${mfn}",
                                        ds => $ds,
                                        type => $config->get($indexer_config)->{type},
-                               ) if ($indexer && $ds);
+                               ) if $indexer;
 
-                               if ($marc) {
-                                       my $i = 0;
-
-                                       while (my $fields = WebPAC::Normalize::_get_marc_fields( fetch_next => 1 ) ) {
-                                               $marc->add(
-                                                       id => $mfn . ( $i ? "/$i" : '' ),
-                                                       fields => $fields,
-                                                       leader => WebPAC::Normalize::marc_leader(),
-                                                       row => $row,
-                                               );
-                                               $i++;
-                                       }
-
-                                       $log->info("Created $i instances of MFN $mfn\n") if ($i > 1);
+                               foreach my $out ( @output_modules ) {
+                                       $out->add( $mfn, $ds ) if $out->can('add');
                                }
 
-                               $total_rows++;
+                       } else {
+                               $log->warn("record $pos didn't produce any output after normalization rules!") unless $marc;
                        }
-
-                       if ($validate) {
-                               my $errors = $validate->report;
-                               if ($errors) {
-                                       $log->info("validation errors:\n$errors\n" );
-                                       print $report_fh "$errors\n" if ($report_fh);
+                       if ($marc) {
+                               my $i = 0;
+
+                               while (my $fields = WebPAC::Normalize::MARC::_get_marc_fields( fetch_next => 1 ) ) {
+                                       $marc->add(
+                                               id => $mfn . ( $i ? "/$i" : '' ),
+                                               fields => $fields,
+                                               leader => WebPAC::Normalize::MARC::_get_marc_leader(),
+                                               row => $row,
+                                       );
+                                       $i++;
                                }
+
+                               $log->info("Created $i instances of MFN $mfn\n") if ($i > 1);
                        }
 
-                       if ($stats) {
-                               my $s = $input_db->stats;
-                               $log->info("statistics of fields usage:\n$s");
-                               print $report_fh "Statistics of fields usage:\n$s" if ($report_fh);
+               }
+
+               if ($validate) {
+                       my $errors = $validate->report;
+                       if ($errors) {
+                               $log->info("validation errors:\n$errors\n" );
+                               print $report_fh "$errors\n" if ($report_fh);
                        }
 
-                       # close MARC file
-                       $marc->finish if ($marc);
+                       print $report_fh "\nAll possible subfields/delimiter templates:\n", $validate->delimiters_templates( report => 1, current_input => 1 ), "\n\n";
 
-                       # close report
-                       close($report_fh) if ($report_fh)
+                       # must be last thing that touches $validate for this input
+                       $validate->reset;
                }
 
+               if ($stats) {
+                       my $s = $input_db->stats;
+                       $log->info("statistics of fields usage:\n$s");
+                       print $report_fh "Statistics of fields usage:\n$s" if ($report_fh);
+               }
+
+               # close MARC file
+               $marc->finish if ($marc);
+
+               # close report
+               close($report_fh) if ($report_fh);
        }
 
-       eval { $indexer->finish } if ($indexer && $indexer->can('finish'));
+       $indexer->finish if $indexer && $indexer->can('finish');
+
+       foreach my $out ( @output_modules ) {
+               $out->finish if $out->can('finish');
+       }
 
        my $dt = time() - $start_t;
        $log->info("$total_rows records ", $indexer ? "indexed " : "",
@@ -516,6 +739,9 @@ if ($parallel) {
        $log->info("all parallel processes finished");
 }
 
+# save new delimiters if needed
+$validate->save_delimiters_templates if ( $validate_delimiters_path );
+
 #
 # handle links or merge after indexing
 #