main.js

const fs = require('fs');
const util = require('util');
const readline = require('node:readline');

const papa = require('papaparse');
const AWS = require('aws-sdk');
const file = fs.createReadStream('csv/{file-name}.csv');

// * Usar perfil de configuración, por defecto se usa "default"
const credentials = new AWS.SharedIniFileCredentials({ profile: 'default' });
AWS.config.update({ region: 'us-east-2', credentials });

// * Inicializamos el servicio de DynamoDB
const ddb = new AWS.DynamoDB({ apiVersion: '2012-08-10' });

// * Nombre de la tabla en DynamoDB
const tableName = "{table-name}";

// * Verifica si hay duplicados en la data
const checkDuplicates = (data) => {
    let setCheck = new Set();
    let set = new Set();

    for (let i = 0; i < data.length; i++) {
        if (setCheck.has(data[i].InternalID)) {
            console.log('============================================================');

            console.log(`=== Duplicado en la posición ${i} del arreglo ===`, data[i]);

            console.log('============================================================\n');
        } else {
            setCheck.add(data[i].InternalID);
            set.add(data[i]);
        }
    }


    console.log('=== Tamaño original del arreglo ===', data.length);
    console.log('=== Tamaño del arreglo ya limpiado ===', setCheck.size);
    console.log('=== Numero de duplicados ===', data.length - setCheck.size);

    return [...set];
}

// * Parte el array en chunks de tamaño especificado
const chunkArray = (arr, size) => {
    return Array.from({ length: Math.ceil(arr.length / size) }, (v, i) => arr.slice(i * size, i * size + size));
}

// * Convierte en promesa la escritura en DynamoDB
const write2Dynamo = (params) => {
    return new Promise((resolve, reject) => {
        ddb.batchWriteItem(params, function (err, data) {
            if (err) {
                console.log("=== Error al guardar en DynamoDB ===", err);
                return reject(err);
            } else {
                console.log("=== Guardado en DynamoDB ===", data);
                return resolve(data);
            }
        });
    });
}

// * Procesa la data del csv
const processData = async (results, save) => {
    // * Verificamos si hay duplicados en el arreglo
    const array = checkDuplicates(results.data);

    const dynamoItems = array.map(item => {
        /*
            * Necesitamos obtener los valores y llaves de cada item,
            * para luego identificar el tipo de dato del valor y crear el objeto
            * con el schema de Dynamo: { PutRequest: { Item: dynamoSchema } } 
        */
        let dynamoSchema = Object.entries(item).reduce((acc, [key, value]) => {
            let type = '';//isNaN(value) ? 'S' : 'N';

            let cleanKey = key.replace(/\s/g, '');

            return ({
                ...acc, [cleanKey]: {
                    [type]: value
                }
            })
        }, {});

        return { PutRequest: { Item: dynamoSchema } }
    });

    const chunks = chunkArray(dynamoItems, 25);

    let registers = 0;
    let error = 0;

    for (let i = 0; i < chunks.length; i++) {
        try {
            console.clear();

            console.log('\n============================================================\n');

            console.log('=== Numero de items ===', dynamoItems.length);
            console.log('=== Registros guardados ===', registers += chunks[i].length);
            console.log('=== Tamaño del chunk ===', chunks[i].length);
            console.log('=== Chunks guardados ===', `${i + 1} / ${chunks.length}`);
            console.log('=== Errores de chunk ===', error);

            for (let j = 0; j < chunks[i].length; j++) {
                console.log(`\n=== Información del chunk ${i + 1} ===`);

                console.log(`=== Numero de item en el chunk: ${j + 1} ===`);
                console.log(`=== Total de keys del item: ${Object.keys(chunks[i][j]?.PutRequest.Item).length} ===`);
                console.log(`=== ${Object.keys(chunks[i][j]?.PutRequest.Item)} ===`);

                console.log('\n=== Información del item ===', chunks[i][j]?.PutRequest.Item);
            }

            console.log('\n============================================================\n');

            if (save) {
                await write2Dynamo({
                    RequestItems: {
                        [tableName]: chunks[i]
                    }
                });
            }
        } catch (error) {
            console.log('\n============================================================\n');

            console.log(`Error en el chunk ${i + 1}:`, error);
            error++;

            // * Ciclamos la información del chunk
            for (let j = 0; j < chunks[j].length; j++) {
                console.log(chunks[i][j]?.PutRequest);
            }

            console.log('\n============================================================\n');

            return;
        }
    }

    console.log('=== Finalizando proceso y guardado de logs ===');
}

// * Pregunta si se desea guardar en DynamoDB
const question = (results) => {
    const rl = readline.createInterface({
        input: process.stdin,
        output: process.stdout,
    });

    rl.question(`¿Desea 1)guardar los registros en DynamoDB o solo 2)generar el log? (1/2): `, async (answer) => {
        const save = answer.toLowerCase();

        if (save !== '1' && save !== '2') {
            console.log('=== Opción no válida, finalizando proceso ===');
            return rl.close();
        }

        await processData(results, save === '1');

        return rl.close();
    });
}

// * Inicializa el log
const initLog = () => {
    return new Promise((resolve, reject) => {
        try {
            // * Verifica si existe la carpeta logs
            if (!fs.existsSync(__dirname + '/logs')) {
                fs.mkdirSync(__dirname + '/logs');
            }

            // * Crea el archivo de log
            const log_file = fs.createWriteStream(__dirname + `/logs/${new Date().getTime()}.log`, { flags: 'w' });
            const log_stdout = process.stdout;

            // * Sobre escribimos la función console.log
            console.log = (d, ...args) => {
                log_file.write(util.format(d, ...args) + '\n');
                log_stdout.write(util.format(d, ...args) + '\n');
            };

            console.log('=== Iniciando proceso y guardado de logs ===\n');

            return resolve();
        } catch (error) {
            return reject(error);
        }
    });
}

// * Obtener los registros que contengan un valor semejante a un string en el campo Description
/*const scanData = async (value) => {
    const params = {
        TableName: tableName,
        FilterExpression: 'contains(Description, :value)',
        ExpressionAttributeValues: {
            ':value': { S: value }
        }
    };

    const data = await ddb.scan(params).promise();

    return data;
}

scanData('{value-to-scan}').then((data) => {
    console.log('=== Data escaneada ===', data.Items.length);
}).catch((error) => {
    console.log('=== Error al escanear la data ===', error);
});*/

// * Inicializamos la lectura del archivo csv y su procesamiento
initLog().then(() => {
    // * Formateamos el archivo csv
    papa.parse(file, {
        header: true,
        skipEmptyLines: true,
        encoding: 'utf-8',
        complete: question
    });
}).catch((error) => {
    console.log('=== Error al inicializar el log ===', error);
});