Неполный CSV на JSON через FTP

Question

Получил код, который подключается к FTP-серверу, преобразует данные в CSV и сохраняет их в массиве, который позже будет использоваться для обновления базы данных.

Я имею дело со многими большими CSV-файлами а при доступе с FTP они возвращают неполные! Я вывожу длину массива, и она каждый раз различается (но обычно близка к полной)

Удаление .end() не меняет результат, поэтому преждевременное отключение не происходит ...

Загрузка файлов CSV вручную (не вариант для разработчиков) приводит к чтению всего файла.

Любые идеи приветствуются, ура :)

Вот код

var FTPClient = require('ftp');
var c = new FTPClient();
var csv = require("csvtojson");

c.connect({
    host: '****',
    user: '****',
    password: '****',
    debug: console.log,
});

dataArray = []
c.on('ready', function() {
    c.get('/file.csv', function(err, stream) {
        if (err) throw err;
        stream.once('close', function() {
            console.log(dataArray.length)
            c.end();
        });
        stream.on('data', function(chunk) {
            csv({
                    noheader: false,
                    headers: ["Header1", "Header2", ]
                })
                .fromString(chunk.toString())
                .subscribe(function(jsonObj) {
                    return new Promise(function(resolve, reject) {
                        dataArray.push(jsonObj)
                        resolve()
                    })
                })
                .on("end", function() {
                    console.log("done");
                });
        });
    });
});

Answer 1

Сейчас я использую это для потоковой передачи и анализа

const PromiseFtp = require('promise-ftp'),
    ftp = new PromiseFtp();
const csv = require("csvtojson");

var jsonArray = []
var credentails = {
    host: '*****',
    user: '*****',
    password: '*****',
}
var csvFile = {
    formatA: ['filesA.csv'],
    formatB: ['filesB.csv'],
    formatC: ['filesC.csv']
}

async function fptUpdate(file) {
    await ftp.connect(credentails).then(function() {
        return ftp.get(file);
    }).then(function(stream) {
        return new Promise(function(resolve, reject) {
            stream.pipe(csv(headers))
                .on('data', (data) => {
                    let string = data.toString('utf8')
                    let json = JSON.parse(string)
                    jsonArray.push(json);
                })
                .on('end', () => {
                    resolve()
                });
        });
    }).then(function() {
        return ftp.end();
    });
}

async function start() {
    for (const file of csvFiles) {
        await fptUpdate(file)
        console.log(jsonArray.length)
        jsonArray = []
        console.log('Processed file: ' + file)
    }
}
start()