<commit_before>'use strict';
var h = require('highland');
var _ = require('lodash');
var moment = require('moment');

var gulp = require('gulp');
var sheets = require('stream-google-spreadsheet');
var es = require('vinyl-elasticsearch');
var s3 = require('vinyl-s3');
var s3Glob = require('s3-glob');
var amqp = require('vinyl-amqp');
let File = require('vinyl');
const googleAnalytics = require('./sinks/googleAnalytics');
const echo = require('./sinks/echo');

var runRecipe = require('./runRecipe');

module.exports = function(connections, codecs, recipes) {
  var channels = {
    googleAnalytics,

    'google-sheets': sheets,
    googleSheets: sheets,

    gulp: gulp,

    elasticsearch: {

      /**
       * Process the search results into individual Vinyl files. This can probably
       * go into vinyl-elasticsearch, but we'd need to take some of the sifttt-specific
       * out first:
       */

      src: (glob, opts) => {
        let docinfoFields = opts.docinfo_fields || ['_index', '_type', '_id'];
        let docinfoTarget = opts.docinfo_target || '@metadata';

        return es.src(glob, opts)
        .flatMap(function (file) {

          /**
           * The search returns an array of results in hits.hits, so we'll
           * convert that to an array of Vinyl files...
           */

          let hits = file.data.hits.hits;
          let aggregations = file.data.aggregations;

          if (!hits.length && aggregations) {
            return h(push => {
              let file = new File({
                contents: new Buffer(JSON.stringify(aggregations))
              });
              file.data = aggregations;
              file.stat = {
                size: file.contents.length
              };
              push(null, file);
              push(null, h.nil);
            });
          }

          return h((push/*, next*/) => {
            hits.forEach(_data => {
              let data = _data._source;
              let metadata = _.pick(_data, docinfoFields);

              data[docinfoTarget] = metadata;
              let file = new File({
                path: metadata._id,
                contents: new Buffer(JSON.stringify(data))
              });

              file.data = data;
              file.stat = {
                size: file.contents.length
              };

              /**
               * This is not quite right; it should be set in the output
               * phase, and should come from fields defined by options.
               * To get that to work we'd need to add a final step that
               * uses the whole evaljson/params business:
               */

              file.id = metadata._id;
              file.index = metadata._index;
              file.type = metadata._type;

              push(null, file);
            });

            push(null, h.nil);
          });
        })
        ;
      },

      /**
       * Insert a shim before dest() in vinyl-elasticsearch, which
       * calculates the index to put the event in, the type and an id,
       * if they don't already exist:
       */

      dest: function(glob, opts) {
        return h.pipeline(function(stream) {
          return stream
            .map(function(file) {
              if (!file.index) {

                /**
                 * [TODO] Make the field name and format string configurable:
                 */

                var timestamp = file.data.startTime;

                file.index = (opts.index || 'logstash-') +
                  moment(timestamp).format('YYYY-MM-DD');
              }
              return file;
            })
            .map(function(file) {
              if (!file.id) {
                if (file.data && file.data.id) {

                  /**
                   * [TODO] Make the field name configurable:<commit_msg>use index option as an index name template that incorporate timestamp<commit_after>'use strict';
var h = require('highland');
var _ = require('lodash');
var moment = require('moment');

var gulp = require('gulp');
var sheets = require('stream-google-spreadsheet');
var es = require('vinyl-elasticsearch');
var s3 = require('vinyl-s3');
var s3Glob = require('s3-glob');
var amqp = require('vinyl-amqp');
let File = require('vinyl');
const googleAnalytics = require('./sinks/googleAnalytics');
const echo = require('./sinks/echo');

var runRecipe = require('./runRecipe');

module.exports = function(connections, codecs, recipes) {
  var channels = {
    googleAnalytics,

    'google-sheets': sheets,
    googleSheets: sheets,

    gulp: gulp,

    elasticsearch: {

      /**
       * Process the search results into individual Vinyl files. This can probably
       * go into vinyl-elasticsearch, but we'd need to take some of the sifttt-specific
       * out first:
       */

      src: (glob, opts) => {
        let docinfoFields = opts.docinfo_fields || ['_index', '_type', '_id'];
        let docinfoTarget = opts.docinfo_target || '@metadata';

        return es.src(glob, opts)
        .flatMap(function (file) {

          /**
           * The search returns an array of results in hits.hits, so we'll
           * convert that to an array of Vinyl files...
           */

          let hits = file.data.hits.hits;
          let aggregations = file.data.aggregations;

          if (!hits.length && aggregations) {
            return h(push => {
              let file = new File({
                contents: new Buffer(JSON.stringify(aggregations))
              });
              file.data = aggregations;
              file.stat = {
                size: file.contents.length
              };
              push(null, file);
              push(null, h.nil);
            });
          }

          return h((push/*, next*/) => {
            hits.forEach(_data => {
              let data = _data._source;
              let metadata = _.pick(_data, docinfoFields);

              data[docinfoTarget] = metadata;
              let file = new File({
                path: metadata._id,
                contents: new Buffer(JSON.stringify(data))
              });

              file.data = data;
              file.stat = {
                size: file.contents.length
              };

              /**
               * This is not quite right; it should be set in the output
               * phase, and should come from fields defined by options.
               * To get that to work we'd need to add a final step that
               * uses the whole evaljson/params business:
               */

              file.id = metadata._id;
              file.index = metadata._index;
              file.type = metadata._type;

              push(null, file);
            });

            push(null, h.nil);
          });
        })
        ;
      },

      /**
       * Insert a shim before dest() in vinyl-elasticsearch, which
       * calculates the index to put the event in, the type and an id,
       * if they don't already exist:
       */

      dest: function(glob, opts) {
        return h.pipeline(function(stream) {
          return stream
            .map(function(file) {
              if (!file.index) {

                /**
                 * [TODO] Make the field name configurable:
                 */

                var timestamp = file.data.startTime;

                file.index = moment(timestamp).format(opts.index || '[logstash]-YYYY.MM.DD');
              }
              return file;
            })
            .map(function(file) {
              if (!file.id) {
                if (file.data && file.data.id) {

                  /**
                   * [TODO] Make the field name configurable:
{'fraction_non_alphanumeric': 0.11535943433285321, 'fraction_numerical': 0.0010475317533062722, 'mean_word_length': 1.5341738553417386, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 11, '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': '865030', 'n_tokens_mistral': 2055, 'n_tokens_neox': 1964, 'n_words': 575}