initial commit.
This commit is contained in:
580
node_modules/csvtojson/v1/core/Converter.js
generated
vendored
Normal file
580
node_modules/csvtojson/v1/core/Converter.js
generated
vendored
Normal file
@@ -0,0 +1,580 @@
|
||||
var util = require("util");
|
||||
var Transform = require("stream").Transform;
|
||||
var os = require("os");
|
||||
var stripBom = require('strip-bom');
|
||||
var eol = os.EOL;
|
||||
// var Processor = require("./Processor.js");
|
||||
var defParam = require("./defParam");
|
||||
var fileline = require("./fileline");
|
||||
var fileLineToCSVLine = require("./fileLineToCSVLine");
|
||||
var linesToJson = require("./linesToJson");
|
||||
var CSVError = require("./CSVError");
|
||||
var workerMgr = null;
|
||||
var _ = require('lodash');
|
||||
var rowSplit = require("./rowSplit");
|
||||
function Converter(params, options) {
|
||||
Transform.call(this, options);
|
||||
this._options = options || {};
|
||||
this.param = defParam(params);
|
||||
this.param._options = this._options;
|
||||
// this.resultObject = new Result(this);
|
||||
// this.pipe(this.resultObject); // it is important to have downstream for a transform otherwise it will stuck
|
||||
this.started = false;//indicate if parsing has started.
|
||||
this.recordNum = 0;
|
||||
this.lineNumber = 0; //file line number
|
||||
this._csvLineBuffer = "";
|
||||
this.lastIndex = 0; // index in result json array
|
||||
//this._pipe(this.lineParser).pipe(this.processor);
|
||||
// this.initNoFork();
|
||||
if (this.param.forked) {
|
||||
this.param.forked = false;
|
||||
this.workerNum = 2;
|
||||
}
|
||||
this.flushCb = null;
|
||||
this.processEnd = false;
|
||||
this.sequenceBuffer = [];
|
||||
this._needJson = null;
|
||||
this._needEmitResult = null;
|
||||
this._needEmitFinalResult = null;
|
||||
this._needEmitHeader = null;
|
||||
this._needEmitJson = null;
|
||||
this._needPush = null;
|
||||
this._needEmitCsv = null;
|
||||
this._csvTransf = null;
|
||||
this.finalResult = [];
|
||||
// this.on("data", function() {});
|
||||
this.on("error", emitDone(this));
|
||||
this.on("end", emitDone(this));
|
||||
this.initWorker();
|
||||
process.nextTick(function () {
|
||||
if (this._needEmitFinalResult === null) {
|
||||
this._needEmitFinalResult = this.listeners("end_parsed").length > 0;
|
||||
}
|
||||
if (this._needEmitResult === null) {
|
||||
this._needEmitResult = this.listeners("record_parsed").length > 0;
|
||||
}
|
||||
if (this._needEmitJson === null) {
|
||||
this._needEmitJson = this.listeners("json").length > 0;
|
||||
}
|
||||
if (this._needEmitHeader === null) {
|
||||
this._needEmitHeader = this.listeners("header").length > 0;
|
||||
}
|
||||
if (this._needEmitCsv === null) {
|
||||
this._needEmitCsv = this.listeners("csv").length > 0;
|
||||
}
|
||||
if (this._needJson === null) {
|
||||
this._needJson = this._needEmitJson || this._needEmitFinalResult || this._needEmitResult || this.transform || this._options.objectMode;
|
||||
}
|
||||
if (this._needPush === null) {
|
||||
this._needPush = this.listeners("data").length > 0 || this.listeners("readable").length > 0;
|
||||
// this._needPush=false;
|
||||
}
|
||||
this.param._needParseJson = this._needJson || this._needPush;
|
||||
}.bind(this));
|
||||
|
||||
return this;
|
||||
}
|
||||
|
||||
util.inherits(Converter, Transform);
|
||||
function emitDone(conv) {
|
||||
return function (err) {
|
||||
if (!conv._hasDone) {
|
||||
conv._hasDone = true;
|
||||
process.nextTick(function () {
|
||||
conv.emit('done', err);
|
||||
});
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
function bufFromString(str) {
|
||||
var length=Buffer.byteLength(str);
|
||||
var buffer = Buffer.allocUnsafe
|
||||
? Buffer.allocUnsafe(length)
|
||||
: new Buffer(length);
|
||||
buffer.write(str);
|
||||
return buffer;
|
||||
}
|
||||
|
||||
Converter.prototype._transform = function (data, encoding, cb) {
|
||||
data=this.prepareData(data);
|
||||
var idx =data.length-1;
|
||||
var left=null;
|
||||
/**
|
||||
* From Keyang:
|
||||
* The code below is to check if a single utf8 char (which could be multiple bytes) being split.
|
||||
* If the char being split, the buffer from two chunk needs to be concat
|
||||
* check how utf8 being encoded to understand the code below.
|
||||
* If anyone has any better way to do this, please let me know.
|
||||
*/
|
||||
if ((data[idx] & 1<<7) !=0){
|
||||
while ((data[idx] & 3<<6) === 128){
|
||||
idx--;
|
||||
}
|
||||
idx--;
|
||||
}
|
||||
if (idx !=data.length-1){
|
||||
left=data.slice(idx+1);
|
||||
data=data.slice(0,idx+1)
|
||||
var _cb=cb;
|
||||
var self=this;
|
||||
cb=function(){
|
||||
if (self._csvLineBuffer){
|
||||
self._csvLineBuffer=Buffer.concat([bufFromString(self._csvLineBuffer,"utf8"),left]);
|
||||
}else{
|
||||
self._csvLineBuffer=left;
|
||||
}
|
||||
_cb();
|
||||
}
|
||||
}
|
||||
data = data.toString("utf8");
|
||||
if (this.started === false) {
|
||||
this.started = true;
|
||||
data = stripBom(data);
|
||||
if (this.param.toArrayString) {
|
||||
if (this._needPush) {
|
||||
this.push("[" + eol, "utf8");
|
||||
}
|
||||
}
|
||||
}
|
||||
var self = this;
|
||||
this.preProcessRaw(data, function (d) {
|
||||
if (d && d.length > 0) {
|
||||
self.processData(d, cb);
|
||||
} else {
|
||||
cb();
|
||||
}
|
||||
});
|
||||
};
|
||||
|
||||
Converter.prototype.prepareData = function (data) {
|
||||
if (this._csvLineBuffer && this._csvLineBuffer.length>0){
|
||||
if (typeof this._csvLineBuffer === "string"){
|
||||
this._csvLineBuffer=bufFromString(this._csvLineBuffer);
|
||||
}
|
||||
return Buffer.concat([this._csvLineBuffer,data]);
|
||||
}else{
|
||||
return data;
|
||||
}
|
||||
// return this._csvLineBuffer + data;
|
||||
};
|
||||
|
||||
Converter.prototype.setPartialData = function (d) {
|
||||
this._csvLineBuffer = d;
|
||||
};
|
||||
|
||||
Converter.prototype.processData = function (data, cb) {
|
||||
var params = this.param;
|
||||
if (params.ignoreEmpty && !params._headers) {
|
||||
data = data.replace(/^\s+/, "");
|
||||
}
|
||||
var eol = this.param.eol;
|
||||
var fileLines = fileline(data, this.param);
|
||||
if (this.param.eol !== eol) {
|
||||
this.emit("eol", this.param.eol);
|
||||
}
|
||||
if (fileLines.lines.length > 0) {
|
||||
if (this.preProcessLine && typeof this.preProcessLine === "function") {
|
||||
fileLines.lines = this._preProcessLines(fileLines.lines, this.lastIndex);
|
||||
}
|
||||
if (!params._headers) { //header is not inited. init header
|
||||
this.processHead(fileLines, cb);
|
||||
} else {
|
||||
if (params.workerNum <= 1) {
|
||||
var lines = fileLineToCSVLine(fileLines, params);
|
||||
this.setPartialData(lines.partial);
|
||||
var jsonArr = linesToJson(lines.lines, params, this.recordNum);
|
||||
this.processResult(jsonArr);
|
||||
this.lastIndex += jsonArr.length;
|
||||
this.recordNum += jsonArr.length;
|
||||
cb();
|
||||
} else {
|
||||
this.workerProcess(fileLines, cb);
|
||||
}
|
||||
}
|
||||
} else {
|
||||
this.setPartialData(fileLines.partial);
|
||||
cb();
|
||||
}
|
||||
};
|
||||
|
||||
Converter.prototype._preProcessLines = function (lines, startIdx) {
|
||||
var rtn = [];
|
||||
for (var i = 0, len = lines.length; i < len; i++) {
|
||||
var result = this.preProcessLine(lines[i], startIdx + i + 1);
|
||||
if (typeof result === "string") {
|
||||
rtn.push(result);
|
||||
} else {
|
||||
rtn.push(lines[i]);
|
||||
this.emit("error", new Error("preProcessLine should return a string but got: " + JSON.stringify(result)));
|
||||
}
|
||||
}
|
||||
return rtn;
|
||||
};
|
||||
|
||||
Converter.prototype.initWorker = function () {
|
||||
var workerNum = this.param.workerNum - 1;
|
||||
if (workerNum > 0) {
|
||||
workerMgr = require("./workerMgr");
|
||||
this.workerMgr = workerMgr();
|
||||
this.workerMgr.initWorker(workerNum, this.param);
|
||||
}
|
||||
};
|
||||
|
||||
Converter.prototype.preRawData = function (func) {
|
||||
this.preProcessRaw = func;
|
||||
return this;
|
||||
};
|
||||
|
||||
Converter.prototype.preFileLine = function (func) {
|
||||
this.preProcessLine = func;
|
||||
return this;
|
||||
};
|
||||
|
||||
/**
|
||||
* workerpRocess does not support embeded multiple lines.
|
||||
*/
|
||||
Converter.prototype.workerProcess = function (fileLine, cb) {
|
||||
var self = this;
|
||||
var line = fileLine;
|
||||
var eol = this.getEol();
|
||||
this.setPartialData(line.partial);
|
||||
this.workerMgr.sendWorker(line.lines.join(eol) + eol, this.lastIndex, cb, function (results, lastIndex) {
|
||||
var buf;
|
||||
var cur = self.sequenceBuffer[0];
|
||||
if (cur.idx === lastIndex) {
|
||||
cur.result = results;
|
||||
var records = [];
|
||||
while (self.sequenceBuffer[0] && self.sequenceBuffer[0].result) {
|
||||
buf = self.sequenceBuffer.shift();
|
||||
records = records.concat(buf.result);
|
||||
}
|
||||
self.processResult(records);
|
||||
self.recordNum += records.length;
|
||||
} else {
|
||||
for (var i = 0, len = self.sequenceBuffer.length; i < len; i++) {
|
||||
buf = self.sequenceBuffer[i];
|
||||
if (buf.idx === lastIndex) {
|
||||
buf.result = results;
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
this.sequenceBuffer.push({
|
||||
idx: this.lastIndex,
|
||||
result: null
|
||||
});
|
||||
this.lastIndex += line.lines.length;
|
||||
};
|
||||
|
||||
Converter.prototype.processHead = function (fileLine, cb) {
|
||||
var params = this.param;
|
||||
if (params._headers) {
|
||||
return cb();
|
||||
}
|
||||
//dirty hack
|
||||
params._needFilterRow = false;
|
||||
// if header is not inited. init header
|
||||
var lines = fileLine.lines;
|
||||
var left = "";
|
||||
var headerRow = [];
|
||||
if (!params.noheader) {
|
||||
while (lines.length) {
|
||||
var line = left + lines.shift();
|
||||
var delimiter = params.delimiter;
|
||||
var row = rowSplit(line, params);
|
||||
if (params.delimiter !== delimiter) {
|
||||
this.emit("delimiter", params.delimiter);
|
||||
}
|
||||
if (row.closed) {
|
||||
headerRow = row.cols;
|
||||
left = "";
|
||||
break;
|
||||
} else {
|
||||
left = line + this.getEol();
|
||||
}
|
||||
}
|
||||
}
|
||||
params._needFilterRow = true;
|
||||
if (!params.noheader && headerRow.length === 0) { //if one chunk of data does not complete header row.
|
||||
this.setPartialData(left);
|
||||
return cb();
|
||||
}
|
||||
if (params.noheader) {
|
||||
if (params.headers) {
|
||||
params._headers = params.headers;
|
||||
} else {
|
||||
params._headers = [];
|
||||
}
|
||||
} else {
|
||||
if (params.headers) {
|
||||
params._headers = params.headers;
|
||||
} else {
|
||||
params._headers = headerRow;
|
||||
}
|
||||
}
|
||||
configIgnoreIncludeColumns(params);
|
||||
params._headers = require("./filterRow")(params._headers, params);
|
||||
if (this._needEmitHeader && this.param._headers) {
|
||||
this.emit("header", this.param._headers);
|
||||
}
|
||||
var delimiter = params.delimiter;
|
||||
var lines = fileLineToCSVLine(fileLine, params);
|
||||
if (params.delimiter !== delimiter) {
|
||||
this.emit("delimiter", params.delimiter);
|
||||
}
|
||||
this.setPartialData(lines.partial);
|
||||
if (this.param.workerNum > 1) {
|
||||
this.workerMgr.setParams(params);
|
||||
}
|
||||
var res = linesToJson(lines.lines, params, 0);
|
||||
// Put the header with the first row
|
||||
// if(res.length > 0) res[0].header = params._headers;
|
||||
this.processResult(res);
|
||||
this.lastIndex += res.length;
|
||||
this.recordNum += res.length;
|
||||
|
||||
cb();
|
||||
};
|
||||
function configIgnoreIncludeColumns(params) {
|
||||
if (params._postIgnoreColumns) {
|
||||
for (var i = 0; i < params.ignoreColumns.length; i++) {
|
||||
var ignoreCol = params.ignoreColumns[i];
|
||||
if (typeof ignoreCol === "string") {
|
||||
var idx = params._headers.indexOf(ignoreCol);
|
||||
if (idx > -1) {
|
||||
params.ignoreColumns[i] = idx;
|
||||
} else {
|
||||
params.ignoreColumns[i] = -1;
|
||||
}
|
||||
}
|
||||
}
|
||||
params.ignoreColumns.sort(function (a, b) { return b - a; });
|
||||
}
|
||||
if (params._postIncludeColumns) {
|
||||
for (var i = 0; i < params.includeColumns.length; i++) {
|
||||
var includeCol = params.includeColumns[i];
|
||||
if (typeof includeCol === "string") {
|
||||
var idx = params._headers.indexOf(includeCol);
|
||||
if (idx > -1) {
|
||||
params.includeColumns[i] = idx;
|
||||
} else {
|
||||
params.includeColumns[i] = -1;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
params.ignoreColumns = _.uniq(params.ignoreColumns);
|
||||
params.includeColumns = _.uniq(params.includeColumns);
|
||||
}
|
||||
|
||||
Converter.prototype.processResult = function (result) {
|
||||
for (var i = 0, len = result.length; i < len; i++) {
|
||||
var r = result[i];
|
||||
if (r.err) {
|
||||
this.emit("error", r.err);
|
||||
} else {
|
||||
this.emitResult(r);
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
Converter.prototype.emitResult = function (r) {
|
||||
var index = r.index;
|
||||
var header = this.param;
|
||||
var row = r.row;
|
||||
var result = r.json;
|
||||
var resultJson = null;
|
||||
var resultStr = null;
|
||||
if (typeof result === "string") {
|
||||
resultStr = result;
|
||||
} else {
|
||||
resultJson = result;
|
||||
}
|
||||
if (resultJson === null && this._needJson) {
|
||||
resultJson = JSON.parse(resultStr);
|
||||
if (typeof row === "string") {
|
||||
row = JSON.parse(row);
|
||||
}
|
||||
}
|
||||
if (this.transform && typeof this.transform === "function") {
|
||||
this.transform(resultJson, row, index);
|
||||
resultStr = null;
|
||||
}
|
||||
if (this._needEmitJson) {
|
||||
this.emit("json", resultJson, index);
|
||||
}
|
||||
if (this._needEmitCsv) {
|
||||
if (typeof row === "string") {
|
||||
row = JSON.parse(row);
|
||||
}
|
||||
this.emit("csv", row, index);
|
||||
}
|
||||
if (this.param.constructResult && this._needEmitFinalResult) {
|
||||
this.finalResult.push(resultJson);
|
||||
}
|
||||
if (this._needEmitResult) {
|
||||
this.emit("record_parsed", resultJson, row, index);
|
||||
}
|
||||
if (this.param.toArrayString && index > 0 && this._needPush) {
|
||||
this.push("," + eol);
|
||||
}
|
||||
if (this._options && this._options.objectMode) {
|
||||
this.push(resultJson);
|
||||
} else {
|
||||
if (this._needPush) {
|
||||
if (resultStr === null) {
|
||||
resultStr = JSON.stringify(resultJson);
|
||||
}
|
||||
this.push(!this.param.toArrayString ? resultStr + eol : resultStr, "utf8");
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
Converter.prototype.preProcessRaw = function (data, cb) {
|
||||
cb(data);
|
||||
};
|
||||
|
||||
// FIXME: lineNumber is not used.
|
||||
Converter.prototype.preProcessLine = function (line, lineNumber) {
|
||||
return line;
|
||||
};
|
||||
|
||||
Converter.prototype._flush = function (cb) {
|
||||
var self = this;
|
||||
this.flushCb = function () {
|
||||
self.emit("end_parsed", self.finalResult);
|
||||
if (self.workerMgr) {
|
||||
self.workerMgr.destroyWorker();
|
||||
}
|
||||
cb();
|
||||
if (!self._needPush) {
|
||||
self.emit("end");
|
||||
}
|
||||
};
|
||||
if (this._csvLineBuffer.length > 0) {
|
||||
var eol = this.getEol();
|
||||
if (this._csvLineBuffer[this._csvLineBuffer.length - 1] !== eol) {
|
||||
this._csvLineBuffer += eol;
|
||||
}
|
||||
this.processData(this._csvLineBuffer, function () {
|
||||
this.checkAndFlush();
|
||||
}.bind(this));
|
||||
} else {
|
||||
this.checkAndFlush();
|
||||
}
|
||||
return;
|
||||
};
|
||||
|
||||
Converter.prototype.checkAndFlush = function () {
|
||||
if (this._csvLineBuffer.length !== 0) {
|
||||
this.emit("error", CSVError.unclosed_quote(this.recordNum, this._csvLineBuffer), this._csvLineBuffer);
|
||||
}
|
||||
if (this.param.toArrayString && this._needPush) {
|
||||
this.push(eol + "]", "utf8");
|
||||
}
|
||||
if (this.workerMgr && this.workerMgr.isRunning()) {
|
||||
this.workerMgr.drain = function () {
|
||||
this.flushCb();
|
||||
}.bind(this);
|
||||
} else {
|
||||
this.flushCb();
|
||||
}
|
||||
};
|
||||
|
||||
Converter.prototype.getEol = function (data) {
|
||||
if (!this.param.eol && data) {
|
||||
for (var i = 0, len = data.length; i < len; i++) {
|
||||
if (data[i] === "\r") {
|
||||
if (data[i + 1] === "\n") {
|
||||
this.param.eol = "\r\n";
|
||||
} else {
|
||||
this.param.eol = "\r";
|
||||
}
|
||||
return this.param.eol;
|
||||
} else if (data[i] === "\n") {
|
||||
this.param.eol = "\n";
|
||||
return this.param.eol;
|
||||
}
|
||||
}
|
||||
this.param.eol = eol;
|
||||
}
|
||||
|
||||
return this.param.eol || eol;
|
||||
};
|
||||
|
||||
Converter.prototype.fromFile = function (filePath, cb, options) {
|
||||
var fs = require('fs');
|
||||
var rs = null;
|
||||
if (typeof cb ==="object" && typeof options === "undefined"){
|
||||
options=cb;
|
||||
cb=null;
|
||||
}
|
||||
this.wrapCallback(cb, function () {
|
||||
if (rs && rs.destroy) {
|
||||
rs.destroy();
|
||||
}
|
||||
});
|
||||
fs.exists(filePath, function (exist) {
|
||||
if (exist) {
|
||||
rs = fs.createReadStream(filePath,options);
|
||||
rs.pipe(this);
|
||||
} else {
|
||||
this.emit('error', new Error("File does not exist. Check to make sure the file path to your csv is correct."));
|
||||
}
|
||||
}.bind(this));
|
||||
return this;
|
||||
};
|
||||
|
||||
Converter.prototype.fromStream = function (readStream, cb) {
|
||||
if (cb && typeof cb === "function") {
|
||||
this.wrapCallback(cb);
|
||||
}
|
||||
readStream.pipe(this);
|
||||
return this;
|
||||
};
|
||||
|
||||
Converter.prototype.transf = function (func) {
|
||||
this.transform = func;
|
||||
return this;
|
||||
};
|
||||
|
||||
Converter.prototype.fromString = function (csvString, cb) {
|
||||
if (typeof csvString !== "string") {
|
||||
if (cb && typeof cb ==="function"){
|
||||
return cb(new Error("Passed CSV Data is not a string."));
|
||||
}
|
||||
}
|
||||
if (cb && typeof cb === "function") {
|
||||
this.wrapCallback(cb, function () {
|
||||
});
|
||||
}
|
||||
process.nextTick(function () {
|
||||
this.end(csvString);
|
||||
}.bind(this));
|
||||
return this;
|
||||
};
|
||||
|
||||
Converter.prototype.wrapCallback = function (cb, clean) {
|
||||
if (clean === undefined) {
|
||||
clean = function () { };
|
||||
}
|
||||
if (cb && typeof cb === "function") {
|
||||
this.once("end_parsed", function (res) {
|
||||
if (!this.hasError) {
|
||||
cb(null, res);
|
||||
}
|
||||
}.bind(this));
|
||||
}
|
||||
this.once("error", function (err) {
|
||||
this.hasError = true;
|
||||
if (cb && typeof cb === "function") {
|
||||
cb(err);
|
||||
}
|
||||
clean();
|
||||
}.bind(this));
|
||||
};
|
||||
|
||||
module.exports = Converter;
|
||||
Reference in New Issue
Block a user