<commit_before>var aws = require('aws-sdk');
var _ = require('underscore');
var util = require('util');
var SimpleFileWriter = require('simple-file-writer');
var log4js = require('log4js');
var path = require('path');
var fs = require('fs');
var events = require('events');

var EventEmitter = events.EventEmitter;

var logger = log4js.getLogger('redshift-bulk-insert');

module.exports = RedshiftBulkInsert;

var DELIMITER = '|';
var ESCAPE = '\\';
var NEWLINE = '\n';
var NULL = '\\N';
var SUFFIX = '.log';

util.inherits(RedshiftBulkInsert, EventEmitter);

function RedshiftBulkInsert(options) {

	var pathToLogs = options.pathToLogs;

	this._datastore = options.datastore;
	this._tableName = options.tableName;
	this._fields = options.fields;
	this._pathToLogs = pathToLogs;

	var idleFlushPeriod = options.idleFlushPeriod;
	this._idleFlushPeriod = _.isNumber(idleFlushPeriod) ? idleFlushPeriod : 10000;

	this._awsBucketName = options.awsBucketName;
	this._awsAccessKeyId = options.awsAccessKeyId;
	this._awsSecretAccessKey = options.awsSecretAccessKey;
	this._awsrRegion = options.awsRegion;

	this._hasEventsInFile = false;
	this._activeFlushOps = 0;

	var fileName = this._getLogFileName();
	var pathToFile = path.join(pathToLogs, fileName);

	this._fileName = fileName;
	this._file = new SimpleFileWriter(pathToFile);
	this._s3 = this._createS3();

	this.startIdleFlushMonitor();
}

RedshiftBulkInsert.prototype._getLogFileName = function() {
	return this._tableName + '_' + Date.now() + SUFFIX;
}

RedshiftBulkInsert.prototype._createS3 = function() {
	var options = {
		region: this._awsRegion,
		accessKeyId: this._awsAccessKeyId,
		secretAccessKey: this._awsSecretAccessKey
	};
	return new aws.S3(options);
}

RedshiftBulkInsert.prototype.flush = function() {

	logger.info('flush');

	if (this._hasEventsInFile) {

		this.activeFlushOps++;

		var oldFileName = this._fileName;
		var pathToLogs = this._pathToLogs;

		var newFileName = this._getLogFileName();
		this._fileName = newFileName;
		var newPathToFile = path.join(pathToLogs, newFileName);
		this._file.setupFile(newPathToFile);

		var sendToS3 = _.bind(this._sendToS3, this, oldFileName);
		var oldPathToFile = path.join(pathToLogs, oldFileName);
		fs.readFile(oldPathToFile, sendToS3);
		this._hasEventsInFile = false;
	}

	this.startIdleFlushMonitor();
};

RedshiftBulkInsert.prototype._sendToS3 = function(fileName, err, body) {
	if (err) {
		logger.error(err);
		return;
	}
	var params = {
		Body: body,
		Key: fileName,
		Bucket: this._awsBucketName
	};
	var onSentToS3 = _.bind(this._onSentToS3, this, fileName);
	this._s3.putObject(params, onSentToS3);
};

RedshiftBulkInsert.prototype._onSentToS3 = function(fileName, err, data) {
	if (err) {
		logger.error(err);
		return;
	}
	var onSentCopyToRedshift = _.bind(this._onSentCopyToRedshift, this, fileName);
	var query = this._getCopyQuery(fileName);
	this._datastore.query(query, onSentCopyToRedshift);
};

RedshiftBulkInsert.prototype._getCopyQuery = function(fileName) {
	return 'COPY '
		+ this._tableName
		+ ' ('
		+ this._fields.join(', ')
		+ ')'
		+ ' FROM '
		+ "'"
		+ 's3://'
		+ this._awsBucketName
		+ '/'
		+ fileName
		+ "'"
		+ ' CREDENTIALS '
		+ "'aws_access_key_id="
		+ this._awsAccessKeyId
		+ ';'
		+ 'aws_secret_access_key='
		+ this._awsSecretAccessKey
		+ "' ";
		// + ' DELIMITER '
		// + "'"
		// + DELIMITER
		// + "'"
		// + ' NULL AS '
		// + "'"
		// + NULL
		// + "'"
		// + ' ESCAPE '
		// + "'"
		// + ESCAPE
		// + "'";
};

RedshiftBulkInsert.prototype._onSentCopyToRedshift = function(fileName, err, result) {
	if (err) {
		logger.error(err);
	}
	this.emit('flush', err, result, '', Date.now(), this);
	var pathToFile = path.join(this._pathToLogs, fileName);
	fs.unlink(pathToFile, function() {
		_activeFlushOps--;
	});
};

RedshiftBulkInsert.prototype.insert = function(row) {
	var line = this._rowToLine(row);
	this._file.write(line + NEWLINE);
	this._hasEventsInFile = true;
};

RedshiftBulkInsert.prototype._rowToLine = function(row) {
	var lineBuff = [];
	for (var i = 0, l = row.length; i < l; i++) {
		var value = row[i];
		value = this._escapeValue(value);
		lineBuff.push(value);<commit_msg>Bug fix in lib/RedshiftBulkInsert.js<commit_after>var aws = require('aws-sdk');
var _ = require('underscore');
var util = require('util');
var SimpleFileWriter = require('simple-file-writer');
var log4js = require('log4js');
var path = require('path');
var fs = require('fs');
var events = require('events');

var EventEmitter = events.EventEmitter;

var logger = log4js.getLogger('redshift-bulk-insert');

module.exports = RedshiftBulkInsert;

var DELIMITER = '|';
var ESCAPE = '\\';
var NEWLINE = '\n';
var NULL = '\\N';
var SUFFIX = '.log';

util.inherits(RedshiftBulkInsert, EventEmitter);

function RedshiftBulkInsert(options) {

	var pathToLogs = options.pathToLogs;

	this._datastore = options.datastore;
	this._tableName = options.tableName;
	this._fields = options.fields;
	this._pathToLogs = pathToLogs;

	var idleFlushPeriod = options.idleFlushPeriod;
	this._idleFlushPeriod = _.isNumber(idleFlushPeriod) ? idleFlushPeriod : 10000;

	this._awsBucketName = options.awsBucketName;
	this._awsAccessKeyId = options.awsAccessKeyId;
	this._awsSecretAccessKey = options.awsSecretAccessKey;
	this._awsrRegion = options.awsRegion;

	this._hasEventsInFile = false;
	this._activeFlushOps = 0;

	var fileName = this._getLogFileName();
	var pathToFile = path.join(pathToLogs, fileName);

	this._fileName = fileName;
	this._file = new SimpleFileWriter(pathToFile);
	this._s3 = this._createS3();

	this.startIdleFlushMonitor();
}

RedshiftBulkInsert.prototype._getLogFileName = function() {
	return this._tableName + '_' + Date.now() + SUFFIX;
}

RedshiftBulkInsert.prototype._createS3 = function() {
	var options = {
		region: this._awsRegion,
		accessKeyId: this._awsAccessKeyId,
		secretAccessKey: this._awsSecretAccessKey
	};
	return new aws.S3(options);
}

RedshiftBulkInsert.prototype.flush = function() {

	logger.info('flush');

	if (this._hasEventsInFile) {

		this.activeFlushOps++;

		var oldFileName = this._fileName;
		var pathToLogs = this._pathToLogs;

		var newFileName = this._getLogFileName();
		this._fileName = newFileName;
		var newPathToFile = path.join(pathToLogs, newFileName);
		this._file.setupFile(newPathToFile);

		var sendToS3 = _.bind(this._sendToS3, this, oldFileName);
		var oldPathToFile = path.join(pathToLogs, oldFileName);
		fs.readFile(oldPathToFile, sendToS3);
		this._hasEventsInFile = false;
	}

	this.startIdleFlushMonitor();
};

RedshiftBulkInsert.prototype._sendToS3 = function(fileName, err, body) {
	if (err) {
		logger.error(err);
		return;
	}
	var params = {
		Body: body,
		Key: fileName,
		Bucket: this._awsBucketName
	};
	var onSentToS3 = _.bind(this._onSentToS3, this, fileName);
	this._s3.putObject(params, onSentToS3);
};

RedshiftBulkInsert.prototype._onSentToS3 = function(fileName, err, data) {
	if (err) {
		logger.error(err);
		return;
	}
	var onSentCopyToRedshift = _.bind(this._onSentCopyToRedshift, this, fileName);
	var query = this._getCopyQuery(fileName);
	this._datastore.query(query, onSentCopyToRedshift);
};

RedshiftBulkInsert.prototype._getCopyQuery = function(fileName) {
	return 'COPY '
		+ this._tableName
		+ ' ('
		+ this._fields.join(', ')
		+ ')'
		+ ' FROM '
		+ "'"
		+ 's3://'
		+ this._awsBucketName
		+ '/'
		+ fileName
		+ "'"
		+ ' CREDENTIALS '
		+ "'aws_access_key_id="
		+ this._awsAccessKeyId
		+ ';'
		+ 'aws_secret_access_key='
		+ this._awsSecretAccessKey
		+ "' ";
		// + ' DELIMITER '
		// + "'"
		// + DELIMITER
		// + "'"
		// + ' NULL AS '
		// + "'"
		// + NULL
		// + "'"
		// + ' ESCAPE '
		// + "'"
		// + ESCAPE
		// + "'";
};

RedshiftBulkInsert.prototype._onSentCopyToRedshift = function(fileName, err, result) {
	if (err) {
		logger.error(err);
	}
	this.emit('flush', err, result, '', Date.now(), this);
	var pathToFile = path.join(this._pathToLogs, fileName);
	fs.unlink(pathToFile, _.bind(this._onLogFileRemoved, this));
};

RedshiftBulkInsert.prototype._onLogFileRemoved = function() {
	this.activeFlushOps--;
};

RedshiftBulkInsert.prototype.insert = function(row) {
	var line = this._rowToLine(row);
	this._file.write(line + NEWLINE);
	this._hasEventsInFile = true;
};

RedshiftBulkInsert.prototype._rowToLine = function(row) {
	var lineBuff = [];
	for (var i = 0, l = row.length; i < l; i++) {
		var value = row[i];
		value = this._escapeValue(value);
		lineBuff.push(value);
{'fraction_non_alphanumeric': 0.15004141521713407, 'fraction_numerical': 0.0056798012069577564, 'mean_word_length': 4.424903722721438, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 3, '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': '781786', 'n_tokens_mistral': 3205, 'n_tokens_neox': 3074, 'n_words': 521}