<commit_before>/**
 *
 * @licstart  The following is the entire license notice for the JavaScript code in this file. 
 *
 * Load records into a data store while filtering, preprocessing and matching & merging them in the process
 *
 * Copyright (c) 2015 University Of Helsinki (The National Library Of Finland)
 *
 * This file is part of record-loader
 *
 * record-loader is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as
 * published by the Free Software Foundation, either version 3 of the
 * License, or (at your option) any later version.
 *  
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Affero General Public License for more details.
 *  
 * You should have received a copy of the GNU Affero General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 *
 * @licend  The above is the entire license notice
 * for the JavaScript code in this file.
 *
 **/

/* istanbul ignore next: umd wrapper */
(function(root, factory) {

    'use strict';

    if (typeof define === 'function' && define.amd) {
        define(['es6-polyfills/lib/promise', 'es6-polyfills/lib/object', 'loglevel', 'loglevel-std-streams', 'jjv', 'jjve', '../resources/config-schema.json'], factory);
    } else if (typeof module === 'object' && module.exports) {
        module.exports = factory(require('es6-polyfills/lib/promise'), require('es6-polyfills/lib/object'), require('loglevel'), require('loglevel-std-streams'), require('jjv'), require('jjve'), require('../resources/config-schema.json'));
    }

}(this, factory));

function factory(Promise, Object, log, loglevelStandardStreams, jjv, jjve, schema)
{
    
    'use strict';

    return function(input_data, modules, config) {

	var processed_records_indices = [];
	var transaction_retries;
	var results = {
	    processed: 0,
	    skipped: 0,
	    recordStore: {
		created: 0,
		updated: 0,
		deleted: 0
	    },
	    merged: 0,
	    transactions: []
	};

	function validateConfig()
	{

	    var env = jjv();
	    var je = jjve(env);
	    var errors = env.validate(schema, config, {
		useDefault: true
	    });
	    
	    if (errors) {
		throw new Error(JSON.stringify(je(schema, config, errors), undefined, 4));
	    }
	    
	}
	
	function initialise()
	{	    

	    function initialiseModule(module_type, is_processor) {
		if (is_processor) {
		    if (module_type in modules.processors) {

			modules.processors[module_type] = modules.processors[module_type](config.moduleParameters.processors[module_type]);
			
			if (module_type === 'match' || module_type === 'load') {
			    modules.processors[module_type].setRecordStore(modules.recordStore);
			}

			if (module_type === 'merge' || module_type === 'load') {
			    modules.processors[module_type].setResultsLevel(config.processing.resultsLevel);
			}

			if (modules.converter) {
			    modules.processors[module_type].setConverter(modules.converter);
			}
			
		    }
		} else if (module_type in modules) {
		    modules[module_type] = modules[module_type](config.moduleParameters[module_type]);
		}
	    }

	    modules = modules === undefined ? {} : modules;
	    config = config === undefined ? {} : config;

	    try {
		validateConfig();
	    } catch (excp) {
		throw new Error("Configuration doesn't validate against schema: "+excp.message);
	    }

	    loglevelStandardStreams(log);
	    log.setLevel(config.logging.level);

	    log.debug('Configuration: '+JSON.stringify( config, null, 2 ));

	    Object.keys(modules).filter(function(key) {
		return key !== 'processors';
	    }).forEach(function(key) {
		initialiseModule(key);
	    });

	    if ('processors' in modules) {
		Object.keys(modules.processors).forEach(function(key) {
		    initialiseModule(key, 1);
		});
	    }

	}   
	
	function processRecordSet(previous_results_or_first_run)
	{
	    log.debug('processRecordSet');
	    if (previous_results_or_first_run === undefined) {
		return Promise.resolve(results);
	    } else {
		if (typeof previous_results_or_first_run === 'object') {
		    
		    var previous_results = Array.isArray(previous_results_or_first_run) ? previous_results_or_first_run : [previous_results_or_first_run];

		    if (config.processing.resultsLevel !== 'total' && !results.hasOwnProperty('records')) {
			results.records = [];
		    }

		    previous_results.forEach(function(previous_result) {

			var record_level_results;

			if (config.processing.resultsLevel !== 'total') {

			    record_level_results = JSON.parse(JSON.stringify(previous_result));

			    delete record_level_results.processed;
			    delete record_level_results.skipped;
			    delete record_level_results.transactions;
			    
			    results.records.push(record_level_results);

			}
		
			Object.keys(previous_result).forEach(function(key) {
			    if (key === 'recordStore') {
				Object.keys(previous_result.recordStore).forEach(function(record_store_key) {
				    results.recordStore[record_store_key] += previous_result.recordStore[record_store_key].length;
				});
			    } else if (key === 'merged') {
				results.merged += previous_result.merged.length > 1 ?  previous_result.merged.length : 0;
			    } else if (previous_result[key]) {
				/**
				 * @internal Adds to 'processed' or 'skipped'
				 */
				results[key]++;
			    }
			});

		    });
		    
		}

		return getRecordData().then(handleRecordData).then(processRecordSet);
	    }
	}

	function getRecordData()
	{
	    log.debug('getRecordData');
	    return modules.recordSet.next().then(function(record_data) {
		if (record_data !== undefined && processed_records_indices.indexOf(record_data.index) >= 0) {
		    log.debug('Skipping already processed record at index ' + record_data.index);
		    return getRecordData();
		} else {
		    return record_data;
		}
	    });
	}

	function handleRecordData(record_data)
	{
	    log.debug('handleRecordData');
	    if (record_data === undefined ) {
		return Promise.resolve();
	    } else if (Array.isArray(record_data)) {
		transaction_retries = 0;
		return startRecordDataTransaction(record_data);
	    } else {
		return processRecord(record_data);
	    }
	}

	function startRecordDataTransaction(records)<commit_msg>Fixed a bug in transaction where already processed records were reprocessed.<commit_after>/**
 *
 * @licstart  The following is the entire license notice for the JavaScript code in this file. 
 *
 * Load records into a data store while filtering, preprocessing and matching & merging them in the process
 *
 * Copyright (c) 2015 University Of Helsinki (The National Library Of Finland)
 *
 * This file is part of record-loader
 *
 * record-loader is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as
 * published by the Free Software Foundation, either version 3 of the
 * License, or (at your option) any later version.
 *  
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Affero General Public License for more details.
 *  
 * You should have received a copy of the GNU Affero General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 *
 * @licend  The above is the entire license notice
 * for the JavaScript code in this file.
 *
 **/

/* istanbul ignore next: umd wrapper */
(function(root, factory) {

    'use strict';

    if (typeof define === 'function' && define.amd) {
        define(['es6-polyfills/lib/promise', 'es6-polyfills/lib/object', 'loglevel', 'loglevel-std-streams', 'jjv', 'jjve', '../resources/config-schema.json'], factory);
    } else if (typeof module === 'object' && module.exports) {
        module.exports = factory(require('es6-polyfills/lib/promise'), require('es6-polyfills/lib/object'), require('loglevel'), require('loglevel-std-streams'), require('jjv'), require('jjve'), require('../resources/config-schema.json'));
    }

}(this, factory));

function factory(Promise, Object, log, loglevelStandardStreams, jjv, jjve, schema)
{
    
    'use strict';

    return function(input_data, modules, config) {

	var processed_records_indices = [];
	var transaction_retries;
	var results = {
	    processed: 0,
	    skipped: 0,
	    recordStore: {
		created: 0,
		updated: 0,
		deleted: 0
	    },
	    merged: 0,
	    transactions: []
	};

	function validateConfig()
	{

	    var env = jjv();
	    var je = jjve(env);
	    var errors = env.validate(schema, config, {
		useDefault: true
	    });
	    
	    if (errors) {
		throw new Error(JSON.stringify(je(schema, config, errors), undefined, 4));
	    }
	    
	}
	
	function initialise()
	{	    

	    function initialiseModule(module_type, is_processor) {
		if (is_processor) {
		    if (module_type in modules.processors) {

			modules.processors[module_type] = modules.processors[module_type](config.moduleParameters.processors[module_type]);
			
			if (module_type === 'match' || module_type === 'load') {
			    modules.processors[module_type].setRecordStore(modules.recordStore);
			}

			if (module_type === 'merge' || module_type === 'load') {
			    modules.processors[module_type].setResultsLevel(config.processing.resultsLevel);
			}

			if (modules.converter) {
			    modules.processors[module_type].setConverter(modules.converter);
			}
			
		    }
		} else if (module_type in modules) {
		    modules[module_type] = modules[module_type](config.moduleParameters[module_type]);
		}
	    }

	    modules = modules === undefined ? {} : modules;
	    config = config === undefined ? {} : config;

	    try {
		validateConfig();
	    } catch (excp) {
		throw new Error("Configuration doesn't validate against schema: "+excp.message);
	    }

	    loglevelStandardStreams(log);
	    log.setLevel(config.logging.level);

	    log.debug('Configuration: '+JSON.stringify( config, null, 2 ));

	    Object.keys(modules).filter(function(key) {
		return key !== 'processors';
	    }).forEach(function(key) {
		initialiseModule(key);
	    });

	    if ('processors' in modules) {
		Object.keys(modules.processors).forEach(function(key) {
		    initialiseModule(key, 1);
		});
	    }

	}   
	
	function processRecordSet(previous_results_or_first_run)
	{
	    log.debug('processRecordSet');
	    if (previous_results_or_first_run === undefined) {
		return Promise.resolve(results);
	    } else {
		if (typeof previous_results_or_first_run === 'object') {
		    
		    var previous_results = Array.isArray(previous_results_or_first_run) ? previous_results_or_first_run : [previous_results_or_first_run];

		    if (config.processing.resultsLevel !== 'total' && !results.hasOwnProperty('records')) {
			results.records = [];
		    }

		    previous_results.forEach(function(previous_result) {

			var record_level_results;

			if (config.processing.resultsLevel !== 'total') {

			    record_level_results = JSON.parse(JSON.stringify(previous_result));

			    delete record_level_results.processed;
			    delete record_level_results.skipped;
			    delete record_level_results.transactions;
			    
			    results.records.push(record_level_results);

			}
		
			Object.keys(previous_result).forEach(function(key) {
			    if (key === 'recordStore') {
				Object.keys(previous_result.recordStore).forEach(function(record_store_key) {
				    results.recordStore[record_store_key] += previous_result.recordStore[record_store_key].length;
				});
			    } else if (key === 'merged') {
				results.merged += previous_result.merged.length > 1 ?  previous_result.merged.length : 0;
			    } else if (previous_result[key]) {
				/**
				 * @internal Adds to 'processed' or 'skipped'
				 */
				results[key]++;
			    }
			});

		    });
		    
		}

		return getRecordData().then(handleRecordData).then(processRecordSet);
	    }
	}

	function getRecordData()
	{
	    log.debug('getRecordData');
	    return modules.recordSet.next().then(function(record_data) {

		var index;

		if (record_data !== undefined) {
		    
		    index = Array.isArray(record_data) ? record_data[0].index : record_data.index;
		    
		    if (processed_records_indices.indexOf(index) >= 0) {
			log.debug('Skipping already processed record at index ' + record_data.index);
			return getRecordData();
		    }
		    
		}
		
		return record_data;

	    });
	}

	function handleRecordData(record_data)
	{
	    log.debug('handleRecordData');
	    if (record_data === undefined ) {
		return Promise.resolve();
	    } else if (Array.isArray(record_data)) {
		transaction_retries = 0;
		return startRecordDataTransaction(record_data);
	    } else {
		return processRecord(record_data);
	    }
	}

	function startRecordDataTransaction(records)
{'fraction_non_alphanumeric': 0.12332688588007737, 'fraction_numerical': 0.003481624758220503, 'mean_word_length': 3.4897533865925667, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6758876', 'n_tokens_mistral': 4288, 'n_tokens_neox': 3752, 'n_words': 901}