Press n or j to go to the next uncovered block, b, p or k for the previous block.
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 | 1x 1x 1x 1x 1x 3x 3x 3x 3x 3x 3x 3x 3x 3x 3x 3x 3x 3x 3x 4x 4x 4x 4x 4x 4x 4x 4x 4x 15x 15x 11x 11x 11x 11x 11x 11x 11x 11x 11x 11x 11x 15x 7x 7x 15x 15x 15x 15x 12x 12x 12x 15x 4x 4x 4x 4x 4x 3x 3x 3x 3x 3x 3x 3x 3x 3x 3x 3x 3x 3x 3x 1x 1x 1x 1x 1x 1x 1x 3x | import { Transform } from 'node:stream'
const BREAK_LINE_SYMBOL = "\n"
const INDEX_NOT_FOUND = -1
export default class CSVToNDJSON extends Transform {
#delimiter = ''
#headers = []
#buffer = Buffer.alloc(0)
constructor({
delimiter = ',',
headers
}) {
super()
this.#delimiter = delimiter
this.#headers = headers
}
* #updateBuffer(chunk) {
// it'll ensure if we got a chunk that is not completed
// and doesnt have a breakline
// will concat with the previous read chunk
// 1st time = 01,
// 2st time = ,erick,adreress\n
// try parsing and returning data!
this.#buffer = Buffer.concat([this.#buffer, chunk])
let breaklineIndex = 0
while(breaklineIndex !== INDEX_NOT_FOUND) {
breaklineIndex = this.#buffer.indexOf(Buffer.from(BREAK_LINE_SYMBOL))
if(breaklineIndex === INDEX_NOT_FOUND) break;
const lineToProcessIndex = breaklineIndex + BREAK_LINE_SYMBOL.length
const line = this.#buffer.subarray(0, lineToProcessIndex)
const lineData = line.toString()
// I'll remove from the main buffer the data
// we already processed!
this.#buffer = this.#buffer.subarray(lineToProcessIndex)
// if it's an empty line ignore this line
if(lineData === BREAK_LINE_SYMBOL) continue;
const NDJSONLine = []
const headers = Array.from(this.#headers)
for(const item of lineData.split(this.#delimiter)) {
const key = headers.shift()
const value = item.replace(BREAK_LINE_SYMBOL, "")
if(key === value) break;
NDJSONLine.push(`"${key}":"${value}"`)
}
if(!NDJSONLine.length) continue;
const ndJSONData = NDJSONLine.join(',')
yield Buffer.from('{'.concat(ndJSONData).concat('}').concat(BREAK_LINE_SYMBOL))
}
}
_transform(chunk, enc, callback) {
for(const item of this.#updateBuffer(chunk)) {
this.push(item);
}
return callback()
}
// when it finishes processing
// this.push(null) on the readable side
// or .end()
_final(callback) {
if(!this.#buffer.length) return callback()
for(const item of this.#updateBuffer(Buffer.from(BREAK_LINE_SYMBOL))) {
this.push(item)
}
callback()
}
} |