import { Import } from '../import' ; import fs from 'fs-extra' ; import logger from '../logger' ; import path from 'path' ; import { pipeline } from 'stream' ; import { promisify } from 'util' ; import { createCityvizorParser , createCityvizorTransformer , CityvizorFileType , } from './parser' ; import { DatabaseWriter } from '../db-writer' ; import { PostprocessingTransformer } from '../postprocessing-transformer' ; export async function importCityvizor ( options ) { const dirFiles = await fs . readdir ( options . importDir ) ; const dataFile = dirFiles . filter ( file => file . match ( /^.*data.*\.csv/i ) ) [ 0 ] ; const eventsFile = dirFiles . filter ( file => file . match ( /^.*events.*\.csv/i ) ) [ 0 ] ; const paymentsFile = dirFiles . filter ( file => file . match ( /^.*payments.*\.csv/i ) ) [ 0 ] ; const accountingFile = dirFiles . filter ( file => file . match ( /^.*accounting.*\.csv/i ) ) [ 0 ] ; if ( ! options . append ) { if ( paymentsFile ) { await options . transaction ( 'data.payments' ) . where ( { profileId : options . profileId , year : options . year } ) . delete ( ) ; logger . log ( 'Deleted previous payment records from the DB.' ) ; } if ( eventsFile ) { await options . transaction ( 'data.events' ) . where ( { profileId : options . profileId , year : options . year } ) . delete ( ) ; logger . log ( 'Deleted previous event records from the DB.' ) ; } if ( accountingFile ) { await options . transaction ( 'data.accounting' ) . where ( { profileId : options . profileId , year : options . year } ) . delete ( ) ; logger . log ( 'Deleted previous accounting records from the DB.' ) ; } if ( dataFile ) { await options . transaction ( 'data.payments' ) . where ( { profileId : options . profileId , year : options . year } ) . delete ( ) ; await options . transaction ( 'data.accounting' ) . where ( { profileId : options . profileId , year : options . year } ) . delete ( ) ; logger . log ( 'Deleted previous payment and accounting records from the DB.' ) ; } } for ( const fileType of [ [ dataFile , CityvizorFileType . DATA ] , [ eventsFile , CityvizorFileType . EVENTS ] , [ paymentsFile , CityvizorFileType . PAYMENTS ] , [ accountingFile , CityvizorFileType . ACCOUNTING ] , ] . filter ( ( [ f ] ) => f ) as [ string , CityvizorFileType ] [ ] ) { const [ file , type ] = fileType ; const filePath = path . join ( options . importDir , file ) ; const fileReader = fs . createReadStream ( filePath ) ; const cvParser = createCityvizorParser ( type ) ; const cvTransformer = createCityvizorTransformer ( type , options ) ; await promisify ( pipeline ) ( fileReader , cvParser , cvTransformer , new PostprocessingTransformer ( ) , new DatabaseWriter ( options ) ) ; } }