<commit_before>'use strict';

var _ = require('lodash');
var indexName = require('./../utils/processor_utils').indexName;

function newProcessor(context, opConfig, jobConfig) {
    var context = context;
    var opConfig = opConfig;

    var logger = jobConfig.logger;

    // index_prefix is require if timeseries
    if (opConfig.timeseries && !opConfig.index_prefix) {
        throw new Error("timeseries requires an index_prefix");
    }

    /*
     * Additional configuration fields. Should validate once a schema is available.
     * update - boolean. set to true if the ES request should be an update
     * upsert_fields - array or strings. Will pull the specified fields from the data record
     *    and create an upsert request
     * id_field - string. Name of the field that contains the id to use for the inserted record.
     * preserve_id - boolean. if incoming data is from ES the existing ID of the record will be preserved.
     */
    return function(data) {
        var fromElastic = false;
        var dataArray;

        if (data.hits && data.hits.hits) {
            fromElastic = true;
            dataArray = data.hits.hits
        } else {
            dataArray = data;
        }

        if (!opConfig.type && !fromElastic) {
            throw ('type must be specified in elasticsearch index selector config if data is not from elasticsearch');
        }
        var formatted = [];

        function generateRequest(start) {
            var record;
            if (fromElastic) {
                record = dataArray[start]._source;
            } else {
                record = dataArray[start];
            }

            var indexSpec = {};

            var meta = {
                "_index": indexName(record, opConfig),
                "_type": opConfig.type ? opConfig.type : data.hits.hits[start]._type
            };

            if (opConfig.preserve_id) {
                meta._id = data.hits.hits[start]._id;
            }
            else if (opConfig.id_field) {
                meta._id = dataArray[start][opConfig.id_field];
            }

            if (opConfig.update || opConfig.upsert) {
                indexSpec['update'] = meta;
            }
            else if (opConfig.delete) {
                indexSpec['delete'] = meta;
            }
            else {
                if (opConfig.create) {
                    indexSpec['create'] = meta;
                }
                else {
                    indexSpec['index'] = meta;
                }
            }

            formatted.push(indexSpec);

            if (opConfig.update || opConfig.upsert) {
                var update = {};

                if (opConfig.upsert) {
                    // The upsert field is what is inserted if the key doesn't already exist
                    update.upsert = record;
                }

                // This will merge this record with the existing record.
                if (opConfig.update_fields.length > 0) {
                    update.doc = {};
                    opConfig.update_fields.forEach(function(field) {<commit_msg>index selector id_field now looks in the right spot resolves# 277 (#278)<commit_after>'use strict';

var _ = require('lodash');
var indexName = require('./../utils/processor_utils').indexName;

function newProcessor(context, opConfig, jobConfig) {
    var opConfig = opConfig;
    var logger = jobConfig.logger;

    // index_prefix is require if timeseries
    if (opConfig.timeseries && !opConfig.index_prefix) {
        throw new Error("timeseries requires an index_prefix");
    }

    /*
     * Additional configuration fields. Should validate once a schema is available.
     * update - boolean. set to true if the ES request should be an update
     * upsert_fields - array or strings. Will pull the specified fields from the data record
     *    and create an upsert request
     * id_field - string. Name of the field that contains the id to use for the inserted record.
     * preserve_id - boolean. if incoming data is from ES the existing ID of the record will be preserved.
     */
    return function(data) {
        var fromElastic = false;
        var dataArray;

        if (data.hits && data.hits.hits) {
            fromElastic = true;
            dataArray = data.hits.hits
        } else {
            dataArray = data;
        }

        if (!opConfig.type && !fromElastic) {
            throw ('type must be specified in elasticsearch index selector config if data is not from elasticsearch');
        }
        var formatted = [];

        function generateRequest(start) {
            var record;
            if (fromElastic) {
                record = dataArray[start]._source;
            } else {
                record = dataArray[start];
            }

            var indexSpec = {};

            var meta = {
                "_index": indexName(record, opConfig),
                "_type": opConfig.type ? opConfig.type : data.hits.hits[start]._type
            };

            if (opConfig.preserve_id) {
                //TODO don't have checks for this to only run when fromElastic is true
                meta._id = data.hits.hits[start]._id;
            }
            else if (opConfig.id_field) {
                if (fromElastic) {
                    meta._id = dataArray[start]._source[opConfig.id_field];
                }
                else {
                    meta._id = dataArray[start][opConfig.id_field];
                }
            }

            if (opConfig.update || opConfig.upsert) {
                indexSpec['update'] = meta;
            }
            else if (opConfig.delete) {
                indexSpec['delete'] = meta;
            }
            else {
                if (opConfig.create) {
                    indexSpec['create'] = meta;
                }
                else {
                    indexSpec['index'] = meta;
                }
            }

            formatted.push(indexSpec);

            if (opConfig.update || opConfig.upsert) {
                var update = {};

                if (opConfig.upsert) {
                    // The upsert field is what is inserted if the key doesn't already exist
                    update.upsert = record;
                }

                // This will merge this record with the existing record.
                if (opConfig.update_fields.length > 0) {
                    update.doc = {};
                    opConfig.update_fields.forEach(function(field) {
{'fraction_non_alphanumeric': 0.08989805375347544, 'fraction_numerical': 0.001235712079085573, 'mean_word_length': 1.5204359673024523, 'pattern_counts': {'":': 4, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '6605334', 'n_tokens_mistral': 1703, 'n_tokens_neox': 1605, 'n_words': 490}