Exemplo de implementação do import de dados para o SOS-RS usando NodeJS / Webstreams API

csvParser.ts

import { Prisma } from '@prisma/client';
import { parse } from 'csv';
import { createReadStream } from 'node:fs';
import { Readable, Transform } from 'node:stream';
import { TransformStream, WritableStream } from 'node:stream/web';

/**
 * Padrão utilizado
 * `nome_do_local, endereco, whatsapp, lat, lng, itens_disponiveis, itens_em_falta`
 */
// const fileSourceStream = createReadStream(__dirname+'/planilha_porto_alegre - comida.csv');

const sourceStreamHttpRequest = fetch('https://docs.google.com/spreadsheets/d/18hY52i65lIdLE2UsugjnKdnE_ubrBCI6nCR0XQurSBk/gviz/tq?tqx=out:csv&sheet=planilha_porto_alegre')


  const csvParser = parse({},function(err,data){
    if(err) console.error(err);
    return data
  });

const transformCsvLineToShelter = new TransformStream({
  async transform(chunk,controller){
    const [nome_do_local,endereco,whatsapp,lat,lng,_itens_disponiveis,itens_em_falta] = chunk;
    const shelter: Partial<Prisma.ShelterCreateInput> = {
      name: nome_do_local,
      address: endereco,
      contact: whatsapp,
      latitude: lat,
      longitude: lng,
      verified: false,
    }
    const shelterSuppliesByCategory = await detectSupplyCategoryUsingAI(itens_em_falta);
    console.log("🚀 ~ transform ~ shelterSuppliesByCategory:", shelterSuppliesByCategory)
    controller.enqueue(shelter);
  },
});

/**
 * https://stackoverflow.com/questions/44672942/stream-response-to-file-using-fetch-api-and-fs-createwritestream
 */
function responseToReadable(response: Response) {
  const rs = new Readable();
  if (!response.body) {
    rs.push(null);
    return rs;
  }
  const reader = response.body.getReader();
  rs._read = async () => {
    const result = await reader.read();
    if (!result.done) {
      rs.push(Buffer.from(result.value));
    } else {
      rs.push(null);
      return;
    }
  };
  return rs;
}

export async function parseCsv(){
  let sheltersSaved = 0;

  Readable.toWeb(responseToReadable(await sourceStreamHttpRequest))
    .pipeThrough(Transform.toWeb(csvParser))
    .pipeThrough(transformCsvLineToShelter)
    .pipeTo(new WritableStream({
      async write(shelter){
        console.log(`${++sheltersSaved} processados.`)
      },
       close(){
        console.log('Job done');
      }
    }))
    ;
}
//TODO: integrar com alguma LLM para categorizar com base nas categorias existentes no banco
async function detectSupplyCategoryUsingAI(input):Promise<{name:string,categoryName?:string}[]>{  
  if(typeof input !== 'string'){
  return [];
  }
  return input.split(',').filter(Boolean).map(item => ({
    name:item.trim(),
    categoryName:''
  }));
  
  
}

parseCsv()