<commit_before>// --------------------------------------------------------------------------------------
var exp = {}
// --------------------------------------------------------------------------------------
// get invalid records for old data
// and save them to database
// --------------------------------------------------------------------------------------
exp.process_old_data = function (database) {
  var log_root = "/home/etlog/logs/fticks/";
  var etlog_log_root = "/home/etlog/logs/";

  // find the lowest date in database and go from that date to present
  var ref_date;
  var current = new Date();
  var curr_min = new Date(current.getFullYear(), current.getMonth(), current.getUTCDate(), 0, 0, 0, 0);   // current day hh:mm:ss:ms set to 00:00:00:000

  // find all, sort by timestamp, display only timestamp, display one document only
  database.logs.find({ query : {}, $orderby : { timestamp : 1 } } , { timestamp : 1, _id : 0 }, { limit : 1 },
  function(err, doc) {
    var ref_date = doc;

    ref_date = String(ref_date[0]["timestamp"]);    // get only string representation of date

    var fields = ref_date.split(" ");
    var months = {      // months dict for date constructor
      "Jan" : 0,
      "Feb" : 1,
      "Mar" : 2,
      "Apr" : 3,
      "May" : 4,
      "Jun" : 5,
      "Jul" : 6,
      "Aug" : 7,
      "Sep" : 8,
      "Oct" : 9,
      "Nov" : 10,
      "Dec" : 11
    }

    // ------------------------------------
    var min = new Date(fields[3], months[fields[1]], fields[2], 0, 0, 0, 0);        // hh:mm:ss:ms set to 0
    var max = new Date(fields[3], months[fields[1]], Number(fields[2]) + 1, 0, 0, 0, 0);    // next day, hh:mm:ss:ms set to 0
                                                                                    // search uses lower than max condition !
    // this date handling should guarantee correct interval for all processed records

    while(min < curr_min) {
      // use max for UTC correction
      // using min could result in a bad day because of midnight and utc offset !
      date = "-" + max.getFullYear() + "-" + zero_pad((Number(max.getMonth()) + 1), 2) + "-" + zero_pad(max.getUTCDate(), 2);   

      log_file = log_root + "fticks" + date;      // original file, which contains invalid records
      err_file = etlog_log_root + "transform/err" + date; // /home/etlog/logs/transform/err-2015-01-25

      get_record_numbers(err_file, log_file, read_lines, save_to_db, database, min);

      min.setDate(min.getDate() + 1);  // continue
      max.setDate(max.getDate() + 1);  // continue
    }
    console.log("cron task invalid_records finished processing old data");
  });
};
// --------------------------------------------------------------------------------------
// run once a day
// --------------------------------------------------------------------------------------
// get invalid records for previous day
// and save them to database
// --------------------------------------------------------------------------------------
exp.process_current_data = function (database) {
  var log_root = "/home/etlog/logs/fticks/";
  var etlog_log_root = "/home/etlog/logs/";
  
  var curr = new Date();        // current day
  var prev_min = new Date(curr.getFullYear(), curr.getMonth(), curr.getUTCDate() - 1, 0, 0, 0, 0);       // previous day hh:mm:ss:ms set to 00:00:00:000
  // this date handling should guarantee correct interval for all processed records

  var date = "-" + prev_min.getFullYear() + "-" + zero_pad((Number(prev_min.getMonth()) + 1), 2) + "-" + zero_pad(prev_min.getDate(), 2);
  // getDate to compensate for UTC offset !!

  var log_file = log_root + "fticks" + date;      // original file, which contains invalid records
  var err_file = etlog_log_root + "transform/err" + date; // /home/etlog/logs/transform/err-2015-01-25

  get_record_numbers(err_file, log_file, read_lines, save_to_db, database, prev_min);
};
// --------------------------------------------------------------------------------------
// get invalid record line numbers and call next processing function
// --------------------------------------------------------------------------------------
function get_record_numbers(err_file, log_file, read_lines, save_to_db, database, db_date)
{
  var arr = [];
  var fs = require('fs');
  var process = require('process');

  try {
    fs.accessSync(err_file, fs.F_OK);
  }
  catch(e) {
    console.log("error log file " + err_file + " does not exist");  // most likely
    return;
  }

  var lineReader = require('readline').createInterface({
    input: require('fs').createReadStream(err_file) // read file
  });

  lineReader.on('line', function (line) {       // read by lines
    arr.push(line.split(':')[1]);   // add line number of invalid record to array
  });
  
  lineReader.on('close', function() {   // when done
    read_lines(log_file, arr, save_to_db, database, db_date);
  });
}
// --------------------------------------------------------------------------------------
// save invalid records to database as array
// --------------------------------------------------------------------------------------
function save_to_db(database, arr, db_date)
{
  database.invalid_records.update({ timestamp : db_date, records : arr }, { timestamp : db_date, records : arr }, { upsert : true },
  function(err, result) {
    if(err)  
      console.log(err);
  });
}
// --------------------------------------------------------------------------------------
// read invalid lines from original file
// --------------------------------------------------------------------------------------
function read_lines(log_file, numbers, save_to_db, database, db_date)
{
  arr = [];
  var fs = require('fs');
  var process = require('process');

  try {
    fs.accessSync(log_file, fs.F_OK);
  }
  catch(e) {
    console.log("log file " + log_file + " does not exist");  // most likely
    return;
  }

  fs.readFile(log_file, function (err, data) {      // read file
    if (err) throw err;

    var lines = data.toString('utf-8').split("\n"); // split by newlines

    for(var number in numbers) {
      arr.push(lines[numbers[number] - 1]);         // indexed from 0
    }

    save_to_db(database, arr, db_date);            // save when done
  });
}
// --------------------------------------------------------------------------------------
// add zero padding of defined width
// --------------------------------------------------------------------------------------
function zero_pad(n, width, z)
{
  z = z || '0';
  n = n + '';
  return n.length >= width ? n : new Array(width - n.length + 1).join(z) + n;
}
// --------------------------------------------------------------------------------------
module.exports = exp;
<commit_msg>do not buffer whole file in memory<commit_after>// --------------------------------------------------------------------------------------
var exp = {}
// --------------------------------------------------------------------------------------
// get invalid records for old data
// and save them to database
// --------------------------------------------------------------------------------------
exp.process_old_data = function (database) {
  var log_root = "/home/etlog/logs/fticks/";
  var etlog_log_root = "/home/etlog/logs/";

  // find the lowest date in database and go from that date to present
  var ref_date;
  var current = new Date();
  var curr_min = new Date(current.getFullYear(), current.getMonth(), current.getUTCDate(), 0, 0, 0, 0);   // current day hh:mm:ss:ms set to 00:00:00:000

  // find all, sort by timestamp, display only timestamp, display one document only
  database.logs.find({ query : {}, $orderby : { timestamp : 1 } } , { timestamp : 1, _id : 0 }, { limit : 1 },
  function(err, doc) {
    var ref_date = doc;

    ref_date = String(ref_date[0]["timestamp"]);    // get only string representation of date

    var fields = ref_date.split(" ");
    var months = {      // months dict for date constructor
      "Jan" : 0,
      "Feb" : 1,
      "Mar" : 2,
      "Apr" : 3,
      "May" : 4,
      "Jun" : 5,
      "Jul" : 6,
      "Aug" : 7,
      "Sep" : 8,
      "Oct" : 9,
      "Nov" : 10,
      "Dec" : 11
    }

    // ------------------------------------
    var min = new Date(fields[3], months[fields[1]], fields[2], 0, 0, 0, 0);        // hh:mm:ss:ms set to 0
    var max = new Date(fields[3], months[fields[1]], Number(fields[2]) + 1, 0, 0, 0, 0);    // next day, hh:mm:ss:ms set to 0
                                                                                    // search uses lower than max condition !
    // this date handling should guarantee correct interval for all processed records

    while(min < curr_min) {
      // use max for UTC correction
      // using min could result in a bad day because of midnight and utc offset !
      date = "-" + max.getFullYear() + "-" + zero_pad((Number(max.getMonth()) + 1), 2) + "-" + zero_pad(max.getUTCDate(), 2);   

      log_file = log_root + "fticks" + date;      // original file, which contains invalid records
      err_file = etlog_log_root + "transform/err" + date; // /home/etlog/logs/transform/err-2015-01-25

      get_record_numbers(err_file, log_file, read_lines, save_to_db, database, min);

      min.setDate(min.getDate() + 1);  // continue
      max.setDate(max.getDate() + 1);  // continue
    }
    console.log("cron task invalid_records finished processing old data");
  });
};
// --------------------------------------------------------------------------------------
// run once a day
// --------------------------------------------------------------------------------------
// get invalid records for previous day
// and save them to database
// --------------------------------------------------------------------------------------
exp.process_current_data = function (database) {
  var log_root = "/home/etlog/logs/fticks/";
  var etlog_log_root = "/home/etlog/logs/";
  
  var curr = new Date();        // current day
  var prev_min = new Date(curr.getFullYear(), curr.getMonth(), curr.getUTCDate() - 1, 0, 0, 0, 0);       // previous day hh:mm:ss:ms set to 00:00:00:000
  // this date handling should guarantee correct interval for all processed records

  var date = "-" + prev_min.getFullYear() + "-" + zero_pad((Number(prev_min.getMonth()) + 1), 2) + "-" + zero_pad(prev_min.getDate(), 2);
  // getDate to compensate for UTC offset !!

  var log_file = log_root + "fticks" + date;      // original file, which contains invalid records
  var err_file = etlog_log_root + "transform/err" + date; // /home/etlog/logs/transform/err-2015-01-25

  get_record_numbers(err_file, log_file, read_lines, save_to_db, database, prev_min);
};
// --------------------------------------------------------------------------------------
// get invalid record line numbers and call next processing function
// --------------------------------------------------------------------------------------
function get_record_numbers(err_file, log_file, read_lines, save_to_db, database, db_date)
{
  var arr = [];
  var fs = require('fs');
  var process = require('process');

  try {
    fs.accessSync(err_file, fs.F_OK);
  }
  catch(e) {
    console.log("error log file " + err_file + " does not exist");  // most likely
    return;
  }

  var lineReader = require('readline').createInterface({
    input: require('fs').createReadStream(err_file) // read file
  });

  lineReader.on('line', function (line) {       // read by lines
    arr.push(line.split(':')[1]);   // add line number of invalid record to array
  });
  
  lineReader.on('close', function() {   // when done
    read_lines(log_file, arr, save_to_db, database, db_date);
  });
}
// --------------------------------------------------------------------------------------
// save invalid records to database as array
// --------------------------------------------------------------------------------------
function save_to_db(database, arr, db_date)
{
  database.invalid_records.update({ timestamp : db_date, records : arr }, { timestamp : db_date, records : arr }, { upsert : true },
  function(err, result) {
    if(err)  
      console.log(err);
  });
}
// --------------------------------------------------------------------------------------
// read invalid lines from original file
// --------------------------------------------------------------------------------------
function read_lines(log_file, numbers, save_to_db, database, db_date)
{
  arr = [];
  var fs = require('fs');
  var process = require('process');

  try {
    fs.accessSync(log_file, fs.F_OK);
  }
  catch(e) {
    console.log("log file " + log_file + " does not exist");  // most likely
    return;
  }

  var lineReader = require('readline').createInterface({
      input: require('fs').createReadStream(log_file)
  });

  var idx = 0;  // index in numbers array
  var invalid_line = numbers[idx];      // first invalid line
  var line_num = 1; // line number in file

  lineReader.on('line', function (line) {   // read file line by line
    if(line_num++ == invalid_line) {
      arr.push(line);
      idx++;
      invalid_line = numbers[idx];
    }
  });

  lineReader.on('close', function() {   // when done
    save_to_db(database, arr, db_date);            // save when done
  });
}
// --------------------------------------------------------------------------------------
// add zero padding of defined width
// --------------------------------------------------------------------------------------
function zero_pad(n, width, z)
{
  z = z || '0';
  n = n + '';
  return n.length >= width ? n : new Array(width - n.length + 1).join(z) + n;
}
// --------------------------------------------------------------------------------------
module.exports = exp;

{'fraction_non_alphanumeric': 0.31317243384704857, 'fraction_numerical': 0.013448676940971212, 'mean_word_length': 1.4190258484262352, 'pattern_counts': {'":': 0, '<': 5, '<?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': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6851343', 'n_tokens_mistral': 3850, 'n_tokens_neox': 3665, 'n_words': 1060}