marndt/avicenna
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

(data-hub): add etl feature

Browse Source
This commit is contained in:
Marcel Arndt
2024-07-09 17:23:08 +02:00
parent 092efb1f53
commit 8be07e1f9b
57 changed files with 5211 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
service/data-hub/data-connector/src/app.controller.ts
+21 -6
View File
@@ -1,12 +1,27 @@
import { Controller, Get } from '@nestjs/common';
import { AppService } from './app.service';
import { Controller, Get, Query } from '@nestjs/common';
import { ImporterService } from './modules/feat-importers/importer.service';
@Controller()
export class AppController {
constructor(private readonly appService: AppService) {}
constructor(private readonly importer: ImporterService) {}
@Get()
getHello(): string {
return this.appService.getHello();
@Get('etl')
etl(@Query('from') from: string, @Query('to') to: string) {
if (!from || !to) {
return {
error: 'from and to must be defined',
};
}
const interval = {
start: new Date(from),
end: new Date(to),
};
return this.importer.runImport(interval);
}
// Call Extractors
// Call Transformers
// Call Loaders
}
service/data-hub/data-connector/src/app.module.ts
+12 -1
View File
@@ -1,9 +1,20 @@
import { Module } from '@nestjs/common';
import { EventEmitterModule } from '@nestjs/event-emitter';
import { ScheduleModule } from '@nestjs/schedule';
import { AppController } from './app.controller';
import { AppService } from './app.service';
import { PrismaModule } from './core/database/prisma.module';
import { FeatureImporterModule } from './modules/feat-importers/feature-importer.module';
import { ConfigModule } from './core/config/config.module';
@Module({
imports: [],
imports: [
ConfigModule,
PrismaModule.forRoot(),
ScheduleModule.forRoot(),
EventEmitterModule.forRoot(),
FeatureImporterModule,
],
controllers: [AppController],
providers: [AppService],
})
service/data-hub/data-connector/src/app.service.ts
+164 -3
View File
@@ -1,8 +1,169 @@
import { Injectable } from '@nestjs/common';
// import { TicketService } from '@dks/avicenna/api-feature-ticket-system';
import { Injectable, Logger } from '@nestjs/common';
import { Cron, SchedulerRegistry } from '@nestjs/schedule';
import {
Interval,
addDays,
eachDayOfInterval,
endOfDay,
format,
set,
startOfDay,
} from 'date-fns';
import { concatMap, from, lastValueFrom, take, tap } from 'rxjs';
import { ImporterService } from './modules/feat-importers/importer.service';
import { chunk } from './utils';
@Injectable()
export class AppService {
getHello(): string {
return 'Hello World!';
private readonly logger = new Logger(AppService.name);
@Cron('0 */5 2-20 * * *')
async fetchDataContinously() {
const now = new Date();
const fetchInterval = {
start: startOfDay(addDays(now, -14)),
end: endOfDay(addDays(now, 7)),
};
this.importer
.runImport(fetchInterval)
.pipe(
take(1),
tap(() =>
this.logger.log(
`import took ${(Date.now() - now.getTime()) / 1000}s`,
),
),
// switchMap(() => from(this.ticketService.createTickets(fetchInterval))),
)
.subscribe();
}
// @Interval(5 * 60 * 1000)
async fetchDataInIntervals() {
const timeouts = this.schedulerRegistry.getTimeouts();
this.logger.log(`Timeouts: ${timeouts.join(', ')}`);
if (timeouts.length) return;
const now = new Date();
const fetchInterval: Interval = {
start: set(new Date(), {
year: now.getFullYear(),
month: 0,
date: 1,
}),
end: endOfDay(now),
};
const intervals = [
...this.splitIntervalIntoPeriods(fetchInterval, 14),
].reverse();
intervals.forEach((interval, index) => {
const readableInterval = `${format(
new Date(interval.start),
'yyyy-MM-dd',
)} - ${format(new Date(interval.end), 'yyyy-MM-dd')}`;
const timeoutName = `Year: ${readableInterval}`;
const callback = () => {
this.logger.log(`Timeout "${timeoutName}"`);
from([interval])
.pipe(
concatMap((interval) => this.importer.runImport(interval)),
// switchMap(() => from(this.ticketService.createTickets(interval))),
)
.subscribe();
};
this.addTimeout(timeoutName, 60 * 1000 * index + 60 * 1000, callback);
});
}
// @Cron('0 2,7,12,17,22,27,32,37,42,47,52,57 2-20 * * *')
async fetchDataSixWeeks() {
const now = new Date();
const fetchInterval: Interval = {
start: startOfDay(addDays(now, -90)),
end: endOfDay(now),
};
const intervals = this.splitIntervalIntoPeriods(fetchInterval, 14);
await lastValueFrom(
from(intervals).pipe(
concatMap((interval, index) =>
this.importer
.runImport(interval)
.pipe(
tap(() =>
this.logger.log(
`${index + 1}. import three months took ${
(Date.now() - now.getTime()) / 1000
}s`,
),
),
),
),
),
);
// await this.ticketService.createTickets(fetchInterval);
}
// @Cron('* * 0 * * *')
async fetchDataYearly() {
const now = new Date();
const fetchInterval: Interval = {
start: set(new Date(), {
year: now.getFullYear(),
month: 0,
date: 1,
}),
end: endOfDay(now),
};
const intervals = this.splitIntervalIntoPeriods(fetchInterval);
from(intervals)
.pipe(
concatMap((interval, index) =>
this.importer
.runImport(interval)
.pipe(
tap(() =>
this.logger.log(
`${index + 1}. import year took ${
(Date.now() - now.getTime()) / 1000
}s`,
),
),
),
),
// switchMap(() => from(this.ticketService.createTickets(fetchInterval))),
)
.subscribe();
}
constructor(
private readonly importer: ImporterService,
// private readonly ticketService: TicketService,
private readonly schedulerRegistry: SchedulerRegistry,
) {}
private splitIntervalIntoPeriods(
interval: Interval,
chunkSize = 30,
): Interval[] {
return chunk(eachDayOfInterval(interval), chunkSize).map((dates) => ({
start: startOfDay(dates.at(0) as Date),
end: endOfDay(dates.at(-1) as Date),
}));
}
private addTimeout(
name: string,
milliseconds: number,
callback: () => void,
): void {
const timeout = setTimeout(callback, milliseconds);
this.schedulerRegistry.addTimeout(name, timeout);
}
}
service/data-hub/data-connector/src/core/config/config.module.ts
+11
View File
@@ -0,0 +1,11 @@
import { Global, Module } from '@nestjs/common';
import { ConfigModule as NestConfigModule } from '@nestjs/config';
import { ConfigService } from './config.service';
@Global()
@Module({
imports: [NestConfigModule.forRoot()],
providers: [ConfigService],
exports: [ConfigService],
})
export class ConfigModule {}
service/data-hub/data-connector/src/core/config/config.service.ts
+27
View File
@@ -0,0 +1,27 @@
import { Injectable } from '@nestjs/common';
import { ConfigService as NestConfigService } from '@nestjs/config';
@Injectable()
export class ConfigService extends NestConfigService {
dispoliveBaseUrl(): string {
return this.get('DISPO_LIVE_BASE_URL');
}
dispoliveUsername(): string {
return this.get('DISPO_LIVE_USERNAME');
}
dispolivePassword(): string {
return this.get('DISPO_LIVE_PASSWORD');
}
dyflexisV0BaseUrl(): string {
return this.get('DYFLEXIS_V0_BASE_URL');
}
dyflexisV0ApiKey(): string {
return this.get('DYFLEXIS_V0_API_KEY');
}
dyflexisV2BaseUrl(): string {
return this.get('DYFLEXIS_V2_BASE_URL');
}
dyflexisV2ApiKey(): string {
return this.get('DYFLEXIS_V2_API_KEY');
}
}
service/data-hub/data-connector/src/core/database/constants.ts
+1
View File
@@ -0,0 +1 @@
export const PRISMA_SERVICE_OPTIONS = 'PRISMA_SERVICE_OPTIONS';
service/data-hub/data-connector/src/core/database/pagination.interface.ts
+9
View File
@@ -0,0 +1,9 @@
export interface PaginationInterface<T> {
items: T[];
meta: {
totalItems: number;
items: number;
totalPages: number;
page: number;
};
}
service/data-hub/data-connector/src/core/database/prisma-module.options.ts
+51
View File
@@ -0,0 +1,51 @@
import { ModuleMetadata, Type } from "@nestjs/common";
import { Prisma } from "@prisma/client";
export interface PrismaModuleOptions {
/**
* If true, registers `PrismaModule` as a global module.
* See: https://docs.nestjs.com/modules#global-modules
*/
isGlobal?: boolean;
prismaServiceOptions?: PrismaServiceOptions;
}
export interface PrismaServiceOptions {
/**
* Pass options directly to the `PrismaClient`.
* See: https://www.prisma.io/docs/reference/api-reference/prisma-client-reference/#prismaclient
*/
prismaOptions?: Prisma.PrismaClientOptions;
/**
* If true, `PrismaClient` explicitly creates a connection pool and your first query will respond instantly.
*
* For most use cases the lazy connect behavior of `PrismaClient` will do. The first query of `PrismaClient` creates the connection pool.
* See: https://www.prisma.io/docs/concepts/components/prisma-client/working-with-prismaclient/connection-management
*/
explicitConnect?: boolean;
/**
* Apply Prisma middlewares to perform actions before or after db queries.
*
* See: https://www.prisma.io/docs/concepts/components/prisma-client/middleware
*/
middlewares?: Array<Prisma.Middleware>;
}
export interface PrismaOptionsFactory {
createPrismaOptions(): Promise<PrismaServiceOptions> | PrismaServiceOptions;
}
export interface PrismaModuleAsyncOptions
extends Pick<ModuleMetadata, 'imports'> {
isGlobal?: boolean;
useExisting?: Type<PrismaOptionsFactory>;
useClass?: Type<PrismaOptionsFactory>;
useFactory?: (
// eslint-disable-next-line @typescript-eslint/no-explicit-any
...args: any[]
) => Promise<PrismaServiceOptions> | PrismaServiceOptions;
// eslint-disable-next-line @typescript-eslint/no-explicit-any
inject?: any[];
}
service/data-hub/data-connector/src/core/database/prisma.module.ts
+100
View File
@@ -0,0 +1,100 @@
import { DynamicModule, Module, Provider } from '@nestjs/common';
import { PRISMA_SERVICE_OPTIONS } from './constants';
import {
PrismaModuleAsyncOptions,
PrismaModuleOptions,
PrismaOptionsFactory,
} from './prisma-module.options';
import { PrismaService } from './prisma.service';
@Module({
providers: [PrismaService],
exports: [PrismaService],
})
export class PrismaModule {
static forRoot(options: PrismaModuleOptions = {}): DynamicModule {
return {
global: options.isGlobal,
module: PrismaModule,
providers: [
{
provide: PRISMA_SERVICE_OPTIONS,
useValue: options.prismaServiceOptions,
},
],
};
}
static forRootAsync(options: PrismaModuleAsyncOptions): DynamicModule {
return {
global: options.isGlobal,
module: PrismaModule,
imports: options.imports || [],
providers: this.createAsyncProviders(options),
};
}
private static createAsyncProviders(
options: PrismaModuleAsyncOptions,
): Provider[] {
if (options.useExisting || options.useFactory) {
return this.createAsyncOptionsProvider(options);
}
if (options.useClass) {
return [
...this.createAsyncOptionsProvider(options),
{
provide: options.useClass,
useClass: options.useClass,
},
];
}
return [...this.createAsyncOptionsProvider(options)];
}
private static createAsyncOptionsProvider(
options: PrismaModuleAsyncOptions,
): Provider[] {
if (options.useFactory) {
return [
{
provide: PRISMA_SERVICE_OPTIONS,
useFactory: options.useFactory,
inject: options.inject || [],
},
];
}
if (options.useExisting) {
return [
{
provide: PRISMA_SERVICE_OPTIONS,
useFactory: async (optionsFactory: PrismaOptionsFactory) =>
await optionsFactory.createPrismaOptions(),
inject: [options.useExisting],
},
];
}
if (options.useClass) {
return [
{
provide: PRISMA_SERVICE_OPTIONS,
useFactory: async (optionsFactory: PrismaOptionsFactory) =>
await optionsFactory.createPrismaOptions(),
inject: [options.useClass],
},
];
}
return [
{
provide: PRISMA_SERVICE_OPTIONS,
useFactory: async (optionsFactory: PrismaOptionsFactory) =>
await optionsFactory.createPrismaOptions(),
inject: [],
},
];
}
}
service/data-hub/data-connector/src/core/database/prisma.service.spec.ts
+18
View File
@@ -0,0 +1,18 @@
import { Test } from '@nestjs/testing';
import { PrismaService } from './prisma.service';
describe('PrismaService', () => {
let service: PrismaService;
beforeEach(async () => {
const module = await Test.createTestingModule({
providers: [PrismaService],
}).compile();
service = module.get(PrismaService);
});
it('should be defined', () => {
expect(service).toBeTruthy();
});
});
service/data-hub/data-connector/src/core/database/prisma.service.ts
+62
View File
@@ -0,0 +1,62 @@
import {
Inject,
Injectable,
Logger,
OnModuleInit,
Optional,
} from '@nestjs/common';
import { PrismaClient } from '@prisma/client';
import { PrismaClientOptions } from '@prisma/client/runtime/library';
import { PRISMA_SERVICE_OPTIONS } from './constants';
import { PrismaServiceOptions } from './prisma-module.options';
@Injectable()
export class PrismaService extends PrismaClient implements OnModuleInit {
private readonly logger = new Logger('Prisma');
constructor(
@Optional()
@Inject(PRISMA_SERVICE_OPTIONS)
private readonly prismaServiceOptions: PrismaServiceOptions = {},
) {
super({
...prismaServiceOptions.prismaOptions,
// Make sure queries are always emitted as events, so that we can log them.
log: [
{ emit: 'event', level: 'query' },
...(prismaServiceOptions.prismaOptions?.log
? prismaServiceOptions.prismaOptions.log
: (['info', 'warn', 'error'] as PrismaClientOptions['log'])),
],
});
if (this.prismaServiceOptions.middlewares) {
this.prismaServiceOptions.middlewares.forEach((middleware) =>
this.$use(middleware),
);
}
/*
Sadly typescript does not recognize the Prisma Client configuration passed in the super call.
As Prisma defines the events generically, based on the configuration, the event type, typescript inferred, is wrong.
*/
// eslint-disable-next-line @typescript-eslint/ban-ts-comment
//@ts-ignore
this.$on('query', (e: QueryEvent) => {
const { timestamp, query, params, duration } = e;
const message = `${timestamp.toISOString()} ${query} ${params} ${duration}`;
// this.logger.log(message);
});
}
async onModuleInit() {
if (this.prismaServiceOptions.explicitConnect) {
await this.$connect();
}
}
// async enableShutdownHooks(app: INestApplication) {
// this.$on('beforeExit', async () => await app.close());
// }
}
service/data-hub/data-connector/src/modules/feat-extractors/data-access-extractors.module.ts
+19
View File
@@ -0,0 +1,19 @@
import { HttpModule } from '@nestjs/axios';
import { Module } from '@nestjs/common';
import { PrismaModule } from 'src/core/database/prisma.module';
import { DispoLiveExtractorsService } from './dispo-live-extractors.service';
import { DyflexisExtractorsService } from './dyflexis-extractors.service';
import { ExtractorsService } from './extractors.service';
import { SourceRepository } from './source.repository';
@Module({
imports: [PrismaModule, HttpModule],
providers: [
ExtractorsService,
DispoLiveExtractorsService,
DyflexisExtractorsService,
SourceRepository,
],
exports: [ExtractorsService],
})
export class DataAccessExtractorsModule {}
service/data-hub/data-connector/src/modules/feat-extractors/dispo-live-extractors.service.ts
+481
View File
@@ -0,0 +1,481 @@
import { HttpService } from '@nestjs/axios';
import { Injectable, Logger } from '@nestjs/common';
import { Prisma } from '@prisma/client';
import {
Interval,
eachDayOfInterval,
endOfDay,
format,
startOfDay,
} from 'date-fns';
import {
Observable,
combineLatest,
concatMap,
delay,
forkJoin,
from,
map,
of,
reduce,
switchMap,
} from 'rxjs';
import { EmployeeDispoLiveRaw, PatientRaw, TourRaw } from './source.interface';
import { SourceRepository } from './source.repository';
import { chunk } from 'src/utils';
import { ConfigService } from 'src/core/config/config.service';
@Injectable()
export class DispoLiveExtractorsService {
private readonly logger = new Logger(DispoLiveExtractorsService.name);
constructor(
private readonly http: HttpService,
private readonly repository: SourceRepository,
private readonly config: ConfigService,
) {}
extractTours(
interval: Interval,
jobId: string,
): Observable<Prisma.BatchPayload> {
const intervals = chunk(eachDayOfInterval(interval), 30).map((dates) => ({
start: startOfDay(dates.at(0)),
end: endOfDay(dates.at(-1)),
}));
return this.fetchTours(intervals).pipe(
map((rawTours) => {
return rawTours;
}),
switchMap((rawTours) =>
from(this.repository.createTours(rawTours, jobId)),
),
);
}
private fetchTours(intervals: Interval[]): Observable<TourRaw[]> {
const archiveEndpoint = 'custom/getArchive/getArchive';
const carsEndpoint = 'data/stammdatenFahrzeuge';
const fetchCars = (
authCookies: string,
): Observable<{ id: string; name: string }[]> =>
this.http
.post<{ data: { _id: string; name: string }[] }>(
`${this.config.dispoliveBaseUrl()}/${carsEndpoint}`,
{},
{
headers: {
'Content-Type': 'application/json',
Cookie: authCookies,
},
},
)
.pipe(
map((response) =>
response.data.data.map(({ _id, name }) => ({ id: _id, name })),
),
);
const fetchToursInInterval = (interval: Interval, authCookies: string) =>
this.http
.post<TourRaw[]>(
`${this.config.dispoliveBaseUrl()}/${archiveEndpoint}`,
{
dateFrom: format(new Date(interval.start), 'yyyy-MM-dd'),
dateTo: format(new Date(interval.end), 'yyyy-MM-dd'),
},
{
headers: {
'Content-Type': 'application/json',
Cookie: authCookies,
},
},
)
.pipe(
map((response) =>
response.data.map((tour) => {
const {
_id,
check,
createTime,
empfangenVonId,
endgen,
serienId,
serienKategorName,
serienKategorie,
zadStatus,
abrDate,
abrNum,
abrTime,
ausrueckZeit,
startTime,
startDate,
date,
date_iso,
dauergenehmigung,
direction,
distanceValue,
besetztKm,
gefahren,
gesamtPreis,
infektion,
itDesc,
itKrz,
itName,
kkId,
krankenkasse,
patId,
patName,
patSurname,
patStreet,
patZip,
patCity,
startInstitution,
startStreet,
startZip,
startCity,
targetInstitution,
targetStreet,
targetCity,
targetZip,
transportKrz,
transportName,
transportart,
verordnungsId,
verordnungsName,
type,
startBegin,
startEnd,
frei,
ziel,
zielBegin,
endTime,
endDate,
einsatzNummer,
carName,
carNo,
beifahrer,
beifahrerName,
fahrer,
fahrerName,
color,
} = tour;
return {
id: _id ? `${_id}` : undefined,
check: check ? `${check}` : undefined,
createTime: createTime ? `${createTime}` : undefined,
empfangenVonId: empfangenVonId
? `${empfangenVonId}`
: undefined,
endgen: endgen ? `${endgen}` : undefined,
serienId: serienId ? `${serienId}` : undefined,
serienKategorName: serienKategorName
? `${serienKategorName}`
: undefined,
serienKategorie: serienKategorie
? `${serienKategorie}`
: undefined,
zadStatus: zadStatus ? `${zadStatus}` : undefined,
abrDate: abrDate ? `${abrDate}` : undefined,
abrNum: abrNum ? `${abrNum}` : undefined,
abrTime: abrTime ? `${abrTime}` : undefined,
ausrueckZeit: ausrueckZeit ? `${ausrueckZeit}` : undefined,
startTime: startTime ? `${startTime}` : undefined,
startDate: startDate ? `${startDate}` : undefined,
date: date ? `${date}` : undefined,
date_iso: date_iso ? `${date_iso}` : undefined,
dauergenehmigung: dauergenehmigung
? `${dauergenehmigung}`
: undefined,
direction: direction ? `${direction}` : undefined,
distanceValue: distanceValue ? `${distanceValue}` : undefined,
besetztKm: besetztKm ? `${besetztKm}` : undefined,
gefahren: gefahren ? `${gefahren}` : undefined,
gesamtPreis: gesamtPreis ? `${gesamtPreis}` : undefined,
infektion: infektion ? `${infektion}` : undefined,
itDesc: itDesc ? `${itDesc}` : undefined,
itKrz: itKrz ? `${itKrz}` : undefined,
itName: itName ? `${itName}` : undefined,
kkId: kkId ? `${kkId}` : undefined,
krankenkasse: krankenkasse ? `${krankenkasse}` : undefined,
patId: patId ? `${patId}` : undefined,
patName: patName ? `${patName}` : undefined,
patSurname: patSurname ? `${patSurname}` : undefined,
patStreet: patStreet ? `${patStreet}` : undefined,
patZip: patZip ? `${patZip}` : undefined,
patCity: patCity ? `${patCity}` : undefined,
startInstitution: startInstitution
? `${startInstitution}`
: undefined,
startStreet: startStreet ? `${startStreet}` : undefined,
startZip: startZip ? `${startZip}` : undefined,
startCity: startCity ? `${startCity}` : undefined,
targetInstitution: targetInstitution
? `${targetInstitution}`
: undefined,
targetStreet: targetStreet ? `${targetStreet}` : undefined,
targetCity: targetCity ? `${targetCity}` : undefined,
targetZip: targetZip ? `${targetZip}` : undefined,
transportKrz: transportKrz ? `${transportKrz}` : undefined,
transportName: transportName ? `${transportName}` : undefined,
transportart: transportart ? `${transportart}` : undefined,
verordnungsId: verordnungsId ? `${verordnungsId}` : undefined,
verordnungsName: verordnungsName
? `${verordnungsName}`
: undefined,
type: type ? `${type}` : undefined,
startBegin: startBegin ? `${startBegin}` : undefined,
startEnd: startEnd ? `${startEnd}` : undefined,
frei: frei ? `${frei}` : undefined,
ziel: ziel ? `${ziel}` : undefined,
zielBegin: zielBegin ? `${zielBegin}` : undefined,
endTime: endTime ? `${endTime}` : undefined,
endDate: endDate ? `${endDate}` : undefined,
einsatzNummer: einsatzNummer ? `${einsatzNummer}` : undefined,
carName: carName ? `${carName}` : undefined,
carNo: carNo ? `${carNo}` : undefined,
beifahrer: beifahrer ? `${beifahrer}` : undefined,
beifahrerName: beifahrerName ? `${beifahrerName}` : undefined,
fahrer: fahrer ? `${fahrer}` : undefined,
fahrerName: fahrerName ? `${fahrerName}` : undefined,
color: color ? `${color}` : undefined,
};
}),
),
);
return this.dispoliveLogin(
this.config.dispoliveUsername(),
this.config.dispolivePassword(),
).pipe(
switchMap((authCookies) =>
forkJoin([of(authCookies), fetchCars(authCookies)]),
),
switchMap(([authCookies, cars]) =>
from(intervals).pipe(
delay(300),
concatMap((interval) => fetchToursInInterval(interval, authCookies)),
reduce((full, tours) => [...full, ...tours], []),
map((tours) =>
tours.map(({ carNo, ...tour }) => {
return {
...tour,
carNo,
carName:
cars.find(({ id }) => id === carNo)?.name ?? tour.carName,
};
}),
),
),
),
);
}
extractEmployees(jobId: string): Observable<Prisma.BatchPayload> {
return this.dispoliveLogin(
this.config.dispoliveUsername(),
this.config.dispolivePassword(),
).pipe(
switchMap((authCookies) => this.fetchEmployees(authCookies)),
switchMap((rawEmployees) =>
from(this.repository.createEmployeesDispoLive(rawEmployees, jobId)),
),
);
}
private fetchEmployees(
authCookies: string,
): Observable<EmployeeDispoLiveRaw[]> {
const employeeEndpoint = 'data/mitarbeiter';
return this.http
.post<{ data: EmployeeDispoLiveRaw[] }>(
`${this.config.dispoliveBaseUrl()}/${employeeEndpoint}`,
{ sort: { surname: 1, status_name: -1 } },
{
headers: {
'Content-Type': 'application/json',
Cookie: authCookies,
},
},
)
.pipe(
map((response) =>
response.data.data.map((employee) => {
const {
_id,
pScheinAblaufDatum,
name,
surname,
persId,
importPersId,
qualifikation,
} = employee;
return {
id: _id,
pScheinAblaufDatum,
name,
surname,
persId,
importPersId,
qualifikation,
};
}),
),
);
}
extractPatients(jobId: string): Observable<Prisma.BatchPayload> {
return this.dispoliveLogin(
this.config.dispoliveUsername(),
this.config.dispolivePassword(),
).pipe(
switchMap((authCookies) => this.fetchPatients(authCookies)),
switchMap((rawPatients) =>
from(this.repository.createPatients(rawPatients, jobId)),
),
);
}
private fetchPatients(authCookies: string): Observable<PatientRaw[]> {
const patientEndpoint = 'data/patientenStammdaten';
const careDegreeEndpoint = 'data/pflegestufe';
const disabilityMarkEndpoint = 'data/schwerbehindertenausweis';
const fetchCareDegrees = (
authCookies: string,
): Observable<{ id: string; careDegree: string }[]> =>
this.http
.post<{ data: { _id: string; pflegestufe: string }[] }>(
`${this.config.dispoliveBaseUrl()}/${careDegreeEndpoint}`,
{},
{
headers: {
'Content-Type': 'application/json',
Cookie: authCookies,
},
},
)
.pipe(
map((response) =>
response.data.data.map(({ _id, pflegestufe }) => ({
id: _id,
careDegree: pflegestufe,
})),
),
);
const fetchDisabilityMarks = (
authCookies: string,
): Observable<{ id: string; disabilityMark: string }[]> =>
this.http
.post<{ data: { _id: string; behinderungShortcut: string }[] }>(
`${this.config.dispoliveBaseUrl()}/${disabilityMarkEndpoint}`,
{},
{
headers: {
'Content-Type': 'application/json',
Cookie: authCookies,
},
},
)
.pipe(
map((response) =>
response.data.data.map(({ _id, behinderungShortcut }) => ({
id: _id,
disabilityMark: behinderungShortcut,
})),
),
);
return combineLatest([
fetchCareDegrees(authCookies),
fetchDisabilityMarks(authCookies),
]).pipe(
switchMap(([careDegrees, disabilityMarks]) =>
this.http
.post<{ data: PatientRaw[] }>(
`${this.config.dispoliveBaseUrl()}/${patientEndpoint}`,
{ sort: { surname: 1, status_name: -1 } },
{
headers: {
'Content-Type': 'application/json',
Cookie: authCookies,
},
},
)
.pipe(
map((response) =>
response.data.data.map((patient) => {
const {
_id,
name,
surname,
street,
zip,
city,
birthday,
kkId,
krankenkasse,
pflegestufe,
behinderung1,
behinderung2,
} = patient;
return {
id: _id,
name,
surname,
street,
zip,
city,
birthday,
kkId,
krankenkasse,
pflegestufe: careDegrees.find(({ id }) => id === pflegestufe)
?.careDegree,
behinderung1: disabilityMarks.find(
({ id }) => id === behinderung1,
)?.disabilityMark,
behinderung2: disabilityMarks.find(
({ id }) => id === behinderung2,
)?.disabilityMark,
};
}),
),
),
),
);
}
private dispoliveLogin(
username: string,
password: string,
): Observable<string> {
const endpoint = 'login';
return this.http
.post(
`${this.config.dispoliveBaseUrl()}/${endpoint}`,
{
username,
password,
},
{
headers: {
'Content-Type': 'application/json',
},
},
)
.pipe(
map((response) =>
response.headers['set-cookie']
// To retrieve only the necessary information of the 'set-cookie' header, we combine them with the same separator, which is used inside one 'set-cookie' request (e.g "key=value; path=/; httponly")
.join(';')
.split(';')
.filter((cookie) => cookie.indexOf('koa.sid') > -1)
.join('; '),
),
);
}
}
service/data-hub/data-connector/src/modules/feat-extractors/dyflexis-extractors.service.ts
+343
View File
@@ -0,0 +1,343 @@
import { HttpService } from '@nestjs/axios';
import { Injectable } from '@nestjs/common';
import { Prisma } from '@prisma/client';
import {
Interval,
eachDayOfInterval,
eachMonthOfInterval,
format,
getMonth,
getYear,
} from 'date-fns';
import {
EMPTY,
Observable,
catchError,
concatMap,
expand,
from,
map,
of,
reduce,
switchMap,
} from 'rxjs';
import {
AttendanceRegistrationRaw,
EmployeeDyflexisRaw,
PlannedTimeRaw,
WorkTimeRaw,
} from './source.interface';
import { SourceRepository } from './source.repository';
import { ConfigService } from 'src/core/config/config.service';
@Injectable()
export class DyflexisExtractorsService {
constructor(
private readonly http: HttpService,
private readonly repository: SourceRepository,
private readonly config: ConfigService,
) {}
extractEmployees(jobId: string): Observable<Prisma.BatchPayload> {
return this.fetchEmployees().pipe(
switchMap((rawEmployees) =>
from(this.repository.createEmployeesDyflexis(rawEmployees, jobId)),
),
);
}
fetchEmployees(): Observable<EmployeeDyflexisRaw[]> {
const endpoint = '/employee';
const getEmployeesPage = (pageNumber: number) =>
this.http
.get<{ response: { data: EmployeeDyflexisRaw[] } }>(
`${this.config.dyflexisV0BaseUrl()}${endpoint}?page=${pageNumber}`,
{
headers: {
'Content-Type': 'application/json',
'X-Dyflexis-AuthToken': `${this.config.dyflexisV0ApiKey()}`,
},
},
)
.pipe(
map((response) =>
response.data.response.data.map((employee) => {
const {
id,
firstname,
surname,
contract_id,
contract_start,
contract_end,
contract_hours_week,
contract_salary_hour,
} = employee;
return {
id,
firstname,
surname,
contract_id,
contract_start,
contract_end,
contract_hours_week,
contract_salary_hour,
};
}),
),
);
return new Observable<EmployeeDyflexisRaw[]>((observer) =>
getEmployeesPage(0)
.pipe(
expand((data, i) => (data.length ? getEmployeesPage(i + 1) : EMPTY)),
reduce((acc, data) => [...acc, ...data], []),
catchError((error) => {
observer.error(error);
return of(error);
}),
)
.subscribe((employees) => {
observer.next(employees);
observer.complete();
}),
);
}
extractAttendanceRegistrations(
interval: Interval,
jobId: string,
): Observable<Prisma.BatchPayload> {
const daysInInterval = eachDayOfInterval(interval);
return from(daysInInterval).pipe(
concatMap((day) => this.fetchAttendanceRegistrationsForDay(day)),
reduce(
(full, attendanceRegistrations) => [
...full,
...attendanceRegistrations,
],
[],
),
switchMap((rawAttendanceRegistrations) =>
from(
this.repository.createAttendanceRegistrations(
rawAttendanceRegistrations,
jobId,
),
),
),
);
}
fetchAttendanceRegistrationsForDay(
date: Date | string,
): Observable<AttendanceRegistrationRaw[]> {
const endpoint = '/attendance-registration';
return this.http
.get<{ attendanceRegistration: AttendanceRegistrationRaw[] }>(
`${this.config.dyflexisV2BaseUrl()}${endpoint}/${format(
new Date(date),
'yyyy-MM-dd',
)}`,
{
headers: {
'Content-Type': 'application/json',
Authorization: `Token ${this.config.dyflexisV2ApiKey()}`,
},
},
)
.pipe(
map((response) =>
response.data.attendanceRegistration.map((attendanceRegistration) => {
const {
attendanceRegistrationId,
employeeId,
personnelNumber,
dateTime,
event,
} = attendanceRegistration;
return {
attendanceRegistrationId: attendanceRegistrationId
? `${attendanceRegistrationId}`
: undefined,
employeeId: employeeId ? `${employeeId}` : undefined,
personnelNumber: personnelNumber
? `${personnelNumber}`
: undefined,
dateTime: dateTime ? `${dateTime}` : undefined,
event: event ? `${event}` : undefined,
};
}),
),
);
}
extractWorkTimes(
interval: Interval,
jobId: string,
): Observable<Prisma.BatchPayload> {
const monthsInInterval = eachMonthOfInterval(interval);
return from(monthsInInterval).pipe(
concatMap((month) =>
this.fetchWorkTimeForMonth(getMonth(month) + 1, getYear(month)),
),
reduce((full, workTimes) => [...full, ...workTimes], []),
switchMap((rawWorkTimes) =>
from(this.repository.createWorkTimes(rawWorkTimes, jobId)),
),
);
}
private fetchWorkTimeForMonth(
month: number,
year: number,
): Observable<WorkTimeRaw[]> {
const endpoint = '/worked';
const getWorkTimePage = (pageNumber: number) =>
this.http
.get<{ response: { data: WorkTimeRaw[] } }>(
`${this.config.dyflexisV0BaseUrl()}${endpoint}?month=${month}&year=${year}&page=${pageNumber}`,
{
headers: {
'Content-Type': 'application/json',
'X-Dyflexis-AuthToken': `${this.config.dyflexisV0ApiKey()}`,
},
},
)
.pipe(
map((response) =>
response.data.response.data.map((workTime) => {
const {
id,
user_id,
firstname,
surname,
department_id,
department_name,
start_date,
end_date,
pauze,
duration,
} = workTime;
return {
id,
user_id,
firstname,
surname,
department_id,
department_name,
start_date,
end_date,
pauze,
duration,
};
}),
),
);
return new Observable<WorkTimeRaw[]>((observer) =>
getWorkTimePage(0)
.pipe(
expand((data, i) => (data.length ? getWorkTimePage(i + 1) : EMPTY)),
reduce((acc, data) => [...acc, ...data], []),
catchError((error) => {
observer.error(error);
return of(error);
}),
)
.subscribe((workTimes) => {
observer.next(workTimes);
observer.complete();
}),
);
}
extractPlannedTimes(
interval: Interval,
jobId: string,
): Observable<Prisma.BatchPayload> {
const monthsInInterval = eachMonthOfInterval(interval);
return from(monthsInInterval).pipe(
concatMap((month) =>
this.fetchPlannedTimeForMonth(getMonth(month) + 1, getYear(month)),
),
reduce((full, plannedTimes) => [...full, ...plannedTimes], []),
switchMap((rawPlannedTimes) =>
from(this.repository.createPlannedTimes(rawPlannedTimes, jobId)),
),
);
}
private fetchPlannedTimeForMonth(
month: number,
year: number,
): Observable<PlannedTimeRaw[]> {
const endpoint = '/planned';
const getPlannedTimePage = (pageNumber: number) =>
this.http
.get<{ response: { data: PlannedTimeRaw[] } }>(
`${this.config.dyflexisV0BaseUrl()}${endpoint}?month=${month}&year=${year}&page=${pageNumber}`,
{
headers: {
'Content-Type': 'application/json',
'X-Dyflexis-AuthToken': `${this.config.dyflexisV0ApiKey()}`,
},
},
)
.pipe(
map((response) =>
response.data.response.data.map((plannedTime) => {
const {
id,
user_id,
firstname,
surname,
department_id,
department_name,
start_date,
end_date,
pauze,
duration,
deleted,
mark,
} = plannedTime;
return {
id,
user_id,
firstname,
surname,
department_id,
department_name,
start_date,
end_date,
pauze,
duration,
deleted,
mark,
};
}),
),
);
return new Observable<PlannedTimeRaw[]>((observer) =>
getPlannedTimePage(0)
.pipe(
expand((data, i) =>
data.length ? getPlannedTimePage(i + 1) : EMPTY,
),
reduce((acc, data) => [...acc, ...data], []),
catchError((error) => {
observer.error(error);
return of(error);
}),
)
.subscribe((plannedTimes) => {
observer.next(plannedTimes);
observer.complete();
}),
);
}
}
service/data-hub/data-connector/src/modules/feat-extractors/extractors.service.ts
+141
View File
@@ -0,0 +1,141 @@
import { Injectable, Logger } from '@nestjs/common';
import {
AttendanceRegistrationSource,
EmployeeDispoLiveSource,
EmployeeDyflexisSource,
PatientSource,
PlannedTimeSource,
Prisma,
TourSource,
WorkTimeSource,
} from '@prisma/client';
import { Interval } from 'date-fns';
import { Observable, catchError, combineLatest, map } from 'rxjs';
import { DispoLiveExtractorsService } from './dispo-live-extractors.service';
import { DyflexisExtractorsService } from './dyflexis-extractors.service';
import { SourceRepository } from './source.repository';
@Injectable()
export class ExtractorsService {
private logger = new Logger(ExtractorsService.name);
constructor(
private readonly dispoLiveExtractors: DispoLiveExtractorsService,
private readonly dyflexisExtractors: DyflexisExtractorsService,
private readonly repository: SourceRepository,
) {}
clearSource(): Promise<Prisma.BatchPayload[]> {
return this.repository.clearSource();
}
extractTours(
interval: Interval,
jobId: string,
): Observable<Prisma.BatchPayload> {
return this.dispoLiveExtractors.extractTours(interval, jobId).pipe(
catchError((error) => {
this.logger.error(error);
throw error;
}),
);
}
extractEmployees(jobId: string): Observable<Prisma.BatchPayload> {
return combineLatest([
this.dispoLiveExtractors.extractEmployees(jobId),
this.dyflexisExtractors.extractEmployees(jobId),
]).pipe(
map(([{ count: countDispoLive }, { count: countDyflexis }]) => {
return {
count: countDispoLive + countDyflexis,
};
}),
catchError((error) => {
this.logger.error(error);
throw error;
}),
);
}
extractPatients(jobId: string): Observable<Prisma.BatchPayload> {
return this.dispoLiveExtractors.extractPatients(jobId).pipe(
catchError((error) => {
this.logger.error(error);
throw error;
}),
);
}
extractPlannedTimes(
interval: Interval,
jobId: string,
): Observable<Prisma.BatchPayload> {
return this.dyflexisExtractors.extractPlannedTimes(interval, jobId).pipe(
catchError((error) => {
this.logger.error(error);
throw error;
}),
);
}
extractWorkTimes(
interval: Interval,
jobId: string,
): Observable<Prisma.BatchPayload> {
return this.dyflexisExtractors.extractWorkTimes(interval, jobId).pipe(
catchError((error) => {
this.logger.error(error);
throw error;
}),
);
}
extractAttendanceRegistrations(
interval: Interval,
jobId: string,
): Observable<Prisma.BatchPayload> {
return this.dyflexisExtractors
.extractAttendanceRegistrations(interval, jobId)
.pipe(
catchError((error) => {
this.logger.error(error);
throw error;
}),
);
}
findToursByJobId(jobId: string): Promise<TourSource[]> {
return this.repository.findToursByJobId(jobId);
}
findPatientsByJobId(jobId: string): Promise<PatientSource[]> {
return this.repository.findPatientsByJobId(jobId);
}
findPlannedTimesByJobId(jobId: string): Promise<PlannedTimeSource[]> {
return this.repository.findPlannedTimesByJobId(jobId);
}
findWorkTimesByJobId(jobId: string): Promise<WorkTimeSource[]> {
return this.repository.findWorkTimesByJobId(jobId);
}
findAttendanceRegistrationsByJobId(
jobId: string,
): Promise<AttendanceRegistrationSource[]> {
return this.repository.findAttendanceRegistrationsByJobId(jobId);
}
findEmployeesDispoLiveByJobId(
jobId: string,
): Promise<EmployeeDispoLiveSource[]> {
return this.repository.findEmployeesDispoLiveByJobId(jobId);
}
findEmployeesDyflexisByJobId(
jobId: string,
): Promise<EmployeeDyflexisSource[]> {
return this.repository.findEmployeesDyflexisByJobId(jobId);
}
}
service/data-hub/data-connector/src/modules/feat-extractors/source.interface.ts
+142
View File
@@ -0,0 +1,142 @@
export interface TourRaw {
_id?: string;
check?: string;
createTime?: string; // yyyy-MM-dd HH:mm:ss 'Europe/Berlin' OR yyyy-MM-dd 'Europe/Berlin'
empfangenVonId?: string;
endgen?: string; // yyyy-MM-dd
serienId?: string;
serienKategorName?: string;
serienKategorie?: string;
zadStatus?: string;
abrDate?: string; // yyyy-MM-dd
abrNum?: string;
abrTime?: string; // HH:mm
ausrueckZeit?: string; // HH:mm 'Europe/Berlin'
startTime?: string; // HH:mm 'Europe/Berlin' (is after ausrueckZeit)
startDate?: string; // yyyy-MM-dd HH:mm:ss 'Europe/Berlin'
date?: string; // yyyy-MM-dd
date_iso?: string; // ISO 8601 String
dauergenehmigung?: string;
direction?: string;
distanceValue?: string; // meters (probably int)
besetztKm?: string; // kilometers (float)
gefahren?: string; // Boolean as number
gesamtPreis?: string; // float
infektion?: string; // Boolean as number
itDesc?: string;
itKrz?: string;
itName?: string;
kkId?: string;
krankenkasse?: string;
patId?: string;
patName?: string;
patSurname?: string;
patStreet?: string;
patZip?: string;
patCity?: string;
startInstitution?: string;
startStreet?: string;
startZip?: string;
startCity?: string;
targetInstitution?: string;
targetStreet?: string;
targetZip?: string;
targetCity?: string;
transportKrz?: string;
transportName?: string;
transportart?: string; // ID
verordnungsId?: string;
verordnungsName?: string;
type?: string;
startBegin?: string; // HH:mm 'Europe/Berlin'
startEnd?: string; // HH:mm 'Europe/Berlin'
frei?: string; // HH:mm 'Europe/Berlin'
ziel?: string; // HH:mm 'Europe/Berlin'
zielBegin?: string; // HH:mm 'Europe/Berlin'
endTime?: string; // HH:mm 'Europe/Berlin'
endDate?: string; // yyyy-MM-dd HH:mm:ss 'Europe/Berlin'
einsatzNummer?: string;
carName?: string;
carNo?: string; // ID
beifahrerName?: string;
beifahrer?: string; // ID
fahrerName?: string;
fahrer?: string; // ID
color?: string; // Hex rgb
}
export interface EmployeeDispoLiveRaw {
_id?: string;
pScheinAblaufDatum?: string; // yyyy-MM-dd
name?: string;
surname?: string;
persId?: string;
importPersId?: string; // Personal ID aus Dyflexis
qualifikation?: string; // ID
}
export interface PatientRaw {
_id?: string;
birthday?: string; // yyyy-MM-dd
name?: string;
surname?: string;
street?: string;
zip?: string;
city?: string;
kkId?: string;
krankenkasse?: string;
pflegestufe?: string;
behinderung1?: string;
behinderung2?: string;
}
export interface AttendanceRegistrationRaw {
attendanceRegistrationId?: string;
employeeId?: string;
personnelNumber?: string;
dateTime?: string; // ISO 8601 String with timezone
event?: string;
}
export interface PlannedTimeRaw {
id?: string;
user_id?: string;
firstname?: string;
surname?: string;
department_id?: string;
department_name?: string;
start_date?: string; // yyyy-MM-dd HH:mm:ss 'Europe/Berlin'
end_date?: string; // yyyy-MM-dd HH:mm:ss 'Europe/Berlin'
pauze?: string;
duration?: string;
deleted?: string;
mark?: string;
}
export interface WorkTimeRaw {
id?: string;
user_id?: string;
firstname?: string;
surname?: string;
department_id?: string;
department_name?: string;
start_date?: string; // yyyy-MM-dd HH:mm:ss 'Europe/Berlin'
end_date?: string; // yyyy-MM-dd HH:mm:ss 'Europe/Berlin'
pauze?: string;
duration?: string;
}
export interface EmployeeDyflexisRaw {
id?: string;
firstname?: string;
surname?: string;
contract_id?: string;
contract_start?: string; // yyyy-MM-dd
contract_end?: string; // yyyy-MM-dd
contract_hours_week?: string; // Float string
contract_salary_hour?: string; // Float string
}
service/data-hub/data-connector/src/modules/feat-extractors/source.repository.ts
+163
View File
@@ -0,0 +1,163 @@
import { Injectable } from '@nestjs/common';
import {
AttendanceRegistrationSource,
EmployeeDispoLiveSource,
EmployeeDyflexisSource,
PatientSource,
PlannedTimeSource,
Prisma,
TourSource,
WorkTimeSource,
} from '@prisma/client';
import { PrismaService } from 'src/core/database/prisma.service';
import {
AttendanceRegistrationRaw,
EmployeeDispoLiveRaw,
EmployeeDyflexisRaw,
PatientRaw,
PlannedTimeRaw,
TourRaw,
WorkTimeRaw,
} from './source.interface';
@Injectable()
export class SourceRepository {
constructor(private readonly prisma: PrismaService) {}
clearSource(): Promise<Prisma.BatchPayload[]> {
return Promise.all([
this.prisma.tourSource.deleteMany(),
this.prisma.employeeDispoLiveSource.deleteMany(),
this.prisma.employeeDyflexisSource.deleteMany(),
this.prisma.workTimeSource.deleteMany(),
this.prisma.plannedTimeSource.deleteMany(),
this.prisma.attendanceRegistrationSource.deleteMany(),
this.prisma.patientSource.deleteMany(),
]);
}
createTours(
tours: TourRaw[],
jobId: string,
): Prisma.PrismaPromise<Prisma.BatchPayload> {
return this.prisma.tourSource.createMany({
data: tours.map((data) => ({ jobId, ...data })),
});
}
createEmployeesDispoLive(
employees: EmployeeDispoLiveRaw[],
jobId: string,
): Prisma.PrismaPromise<Prisma.BatchPayload> {
return this.prisma.employeeDispoLiveSource.createMany({
data: employees.map((data) => ({ jobId, ...data })),
});
}
createPatients(
patients: PatientRaw[],
jobId: string,
): Promise<Prisma.BatchPayload> {
return this.prisma.patientSource.createMany({
data: patients.map((data) => ({ jobId, ...data })),
});
}
createEmployeesDyflexis(
employees: EmployeeDyflexisRaw[],
jobId: string,
): Prisma.PrismaPromise<Prisma.BatchPayload> {
return this.prisma.employeeDyflexisSource.createMany({
data: employees.map((data) => ({ jobId, ...data })),
});
}
createAttendanceRegistrations(
attendanceRegistrations: AttendanceRegistrationRaw[],
jobId: string,
): Prisma.PrismaPromise<Prisma.BatchPayload> {
return this.prisma.attendanceRegistrationSource.createMany({
data: attendanceRegistrations.map((data) => ({ jobId, ...data })),
});
}
createWorkTimes(
workTimes: WorkTimeRaw[],
jobId: string,
): Prisma.PrismaPromise<Prisma.BatchPayload> {
return this.prisma.workTimeSource.createMany({
data: workTimes.map((data) => ({ jobId, ...data })),
});
}
createPlannedTimes(
plannedTimes: PlannedTimeRaw[],
jobId: string,
): Prisma.PrismaPromise<Prisma.BatchPayload> {
return this.prisma.plannedTimeSource.createMany({
data: plannedTimes.map((data) => ({ jobId, ...data })),
});
}
findToursByJobId(jobId: string): Promise<TourSource[]> {
return this.prisma.tourSource.findMany({
where: {
jobId,
},
});
}
findPatientsByJobId(jobId: string): Promise<PatientSource[]> {
return this.prisma.patientSource.findMany({
where: {
jobId,
},
});
}
findPlannedTimesByJobId(jobId: string): Promise<PlannedTimeSource[]> {
return this.prisma.plannedTimeSource.findMany({
where: {
jobId,
},
});
}
findWorkTimesByJobId(jobId: string): Promise<WorkTimeSource[]> {
return this.prisma.workTimeSource.findMany({
where: {
jobId,
},
});
}
findAttendanceRegistrationsByJobId(
jobId: string,
): Promise<AttendanceRegistrationSource[]> {
return this.prisma.attendanceRegistrationSource.findMany({
where: {
jobId,
},
});
}
findEmployeesDyflexisByJobId(
jobId: string,
): Promise<EmployeeDyflexisSource[]> {
return this.prisma.employeeDyflexisSource.findMany({
where: {
jobId,
},
});
}
findEmployeesDispoLiveByJobId(
jobId: string,
): Promise<EmployeeDispoLiveSource[]> {
return this.prisma.employeeDispoLiveSource.findMany({
where: {
jobId,
},
});
}
}
service/data-hub/data-connector/src/modules/feat-importers/events/event-types.enum.ts
+5
View File
@@ -0,0 +1,5 @@
export enum EventTypes {
ExtractionDone = 'etl.extraction',
TransformationDone = 'etl.transformation',
LoadingDone = 'etl.loading',
}
service/data-hub/data-connector/src/modules/feat-importers/events/extraction-done.event.ts
+3
View File
@@ -0,0 +1,3 @@
export class ExtractionDoneEvent {
constructor(public readonly jobId: string) {}
}
service/data-hub/data-connector/src/modules/feat-importers/events/loading-done.event.ts
+3
View File
@@ -0,0 +1,3 @@
export class LoadingDoneEvent {
constructor(public readonly jobId: string) {}
}
service/data-hub/data-connector/src/modules/feat-importers/events/transformation-done.event.ts
+3
View File
@@ -0,0 +1,3 @@
export class TransformationDoneEvent {
constructor(public readonly jobId: string) {}
}
service/data-hub/data-connector/src/modules/feat-importers/feature-importer.module.ts
+17
View File
@@ -0,0 +1,17 @@
import { Module } from '@nestjs/common';
import { DataAccessExtractorsModule } from '../feat-extractors/data-access-extractors.module';
import { DataAccessLoadersModule } from '../feat-loaders/data-access-loaders.module';
import { DataAccessTransformersModule } from '../feat-transformers/data-access-transformers.module';
import { ImporterService } from './importer.service';
@Module({
imports: [
DataAccessExtractorsModule,
DataAccessTransformersModule,
DataAccessLoadersModule,
],
providers: [ImporterService],
exports: [ImporterService],
})
export class FeatureImporterModule {}
service/data-hub/data-connector/src/modules/feat-importers/importer.service.ts
+113
View File
@@ -0,0 +1,113 @@
import { Injectable, Logger } from '@nestjs/common';
import { EventEmitter2 } from '@nestjs/event-emitter';
import { randomUUID } from 'crypto';
import { Interval } from 'date-fns';
import { Observable, combineLatest, from, map, switchMap, tap } from 'rxjs';
import { ExtractorsService } from '../feat-extractors/extractors.service';
import { LoadersService } from '../feat-loaders/loaders.service';
import { TransformersService } from '../feat-transformers/transformers.service';
import { EventTypes } from './events/event-types.enum';
import { ExtractionDoneEvent } from './events/extraction-done.event';
import { LoadingDoneEvent } from './events/loading-done.event';
import { TransformationDoneEvent } from './events/transformation-done.event';
@Injectable()
export class ImporterService {
private logger = new Logger(ImporterService.name);
constructor(
private readonly extractorsService: ExtractorsService,
private readonly transformersService: TransformersService,
private readonly loadersService: LoadersService,
private readonly eventEmitter: EventEmitter2,
) {}
runImport(interval: Interval) {
const startTS = new Date().getTime();
const jobId = randomUUID();
return from(this.extractorsService.clearSource()).pipe(
switchMap(() => this.runExtractors(interval, jobId)),
tap(() =>
this.logger.log(
`Extraction of ${jobId} took ${new Date().getTime() - startTS}ms`,
),
),
switchMap(() =>
from(this.transformersService.clearStaging()).pipe(
switchMap(() => this.runTransformers(jobId)),
),
),
tap(() =>
this.logger.log(
`Transformation of ${jobId} took ${new Date().getTime() - startTS}ms`,
),
),
switchMap(() => this.runLoaders(jobId)),
tap(() =>
this.logger.log(
`Loading of ${jobId} took ${new Date().getTime() - startTS}ms`,
),
),
);
}
private runExtractors(interval: Interval, jobId: string): Observable<void> {
return combineLatest([
this.extractorsService.extractAttendanceRegistrations(interval, jobId),
this.extractorsService.extractEmployees(jobId),
this.extractorsService.extractPatients(jobId),
this.extractorsService.extractPlannedTimes(interval, jobId),
this.extractorsService.extractTours(interval, jobId),
this.extractorsService.extractWorkTimes(interval, jobId),
]).pipe(
tap(() =>
this.eventEmitter.emit(
EventTypes.ExtractionDone,
new ExtractionDoneEvent(jobId),
),
),
map(() => {
return;
}),
);
}
private runTransformers(jobId: string): Observable<void> {
return combineLatest([
this.transformersService.transformAttendanceRegistrations(jobId),
this.transformersService.transformEmployees(jobId),
this.transformersService.transformPatients(jobId),
this.transformersService.transformPlannedTimes(jobId),
this.transformersService.transformWorkTimes(jobId),
this.transformersService.transformTours(jobId),
]).pipe(
tap(() =>
this.eventEmitter.emit(
EventTypes.TransformationDone,
new TransformationDoneEvent(jobId),
),
),
map(() => {
return;
}),
);
}
private runLoaders(jobId: string): Observable<void> {
return this.loadersService.loadTours().pipe(
switchMap(() => this.loadersService.loadPatients()),
switchMap(() => this.loadersService.loadAttendanceRegistrations()),
switchMap(() => this.loadersService.loadPlannedTimes()),
switchMap(() => this.loadersService.loadWorkTimes()),
tap(() =>
this.eventEmitter.emit(
EventTypes.LoadingDone,
new LoadingDoneEvent(jobId),
),
),
map(() => {
return;
}),
);
}
}
service/data-hub/data-connector/src/modules/feat-loaders/data-access-loaders.module.ts
+13
View File
@@ -0,0 +1,13 @@
import { Module } from '@nestjs/common';
import { PrismaModule } from 'src/core/database/prisma.module';
import { DataAccessTransformersModule } from '../feat-transformers/data-access-transformers.module';
import { FinalRepository } from './final.repository';
import { LoadersService } from './loaders.service';
@Module({
imports: [PrismaModule, DataAccessTransformersModule],
providers: [LoadersService, FinalRepository],
exports: [LoadersService],
})
export class DataAccessLoadersModule {}
service/data-hub/data-connector/src/modules/feat-loaders/final.repository.ts
+278
View File
@@ -0,0 +1,278 @@
import { Injectable } from '@nestjs/common';
import {
AttendanceRegistration,
Patient,
PlannedTime,
Prisma,
Tour,
WorkTime,
} from '@prisma/client';
import { Interval } from 'date-fns';
import { PrismaService } from 'src/core/database/prisma.service';
@Injectable()
export class FinalRepository {
constructor(private readonly prisma: PrismaService) {}
findToursInInterval(
interval: Interval,
operationIds: string[],
): Promise<Tour[]> {
return this.prisma.tour.findMany({
where: {
OR: [
{
AND: [
{
startDate: {
gte: new Date(interval.start),
},
},
{
startDate: {
lte: new Date(interval.end),
},
},
],
},
{
operationId: {
in: operationIds,
},
},
],
},
});
}
createTours(
tourCreateInput: Prisma.TourCreateInput[],
): Promise<Prisma.BatchPayload> {
return this.prisma.tour.createMany({
data: tourCreateInput,
skipDuplicates: true,
});
}
async updateTours(
tourCreateInput: Prisma.TourCreateInput[],
): Promise<Prisma.BatchPayload> {
await this.prisma.tour.deleteMany({
where: {
operationId: {
in: tourCreateInput.map(({ operationId }) => operationId),
},
},
});
return this.createTours(tourCreateInput);
}
deleteTours(operationIds: string[]): Promise<Prisma.BatchPayload> {
return this.prisma.tour.updateMany({
where: {
operationId: {
in: operationIds,
},
},
data: {
deletedAt: new Date(),
},
});
}
findPatients(): Promise<Patient[]> {
return this.prisma.patient.findMany({
orderBy: {
surname: 'asc',
},
});
}
createPatients(
patientCreateInput: Prisma.PatientCreateInput[],
): Promise<Prisma.BatchPayload> {
return this.prisma.patient.createMany({
data: patientCreateInput,
skipDuplicates: true,
});
}
async updatePatients(
patientCreateInput: Prisma.PatientCreateInput[],
): Promise<Prisma.BatchPayload> {
await this.deletePatients(patientCreateInput.map(({ id }) => id));
return this.createPatients(patientCreateInput);
}
deletePatients(ids: string[]): Promise<Prisma.BatchPayload> {
return this.prisma.patient.deleteMany({
where: {
id: {
in: ids,
},
},
});
}
findAttendanceRegistrationsInInterval(
interval: Interval,
): Promise<AttendanceRegistration[]> {
return this.prisma.attendanceRegistration.findMany({
where: {
AND: [
{
dateTime: {
gte: new Date(interval.start),
},
},
{
dateTime: {
lte: new Date(interval.end),
},
},
],
},
});
}
createAttendanceRegistrations(
attendanceRegistrationCreateInput: Prisma.AttendanceRegistrationCreateInput[],
): Promise<Prisma.BatchPayload> {
return this.prisma.attendanceRegistration.createMany({
data: attendanceRegistrationCreateInput,
skipDuplicates: true,
});
}
async updateAttendanceRegistrations(
attendanceRegistrationCreateInput: Prisma.AttendanceRegistrationCreateInput[],
): Promise<Prisma.BatchPayload> {
await this.prisma.attendanceRegistration.deleteMany({
where: {
attendanceRegistrationId: {
in: attendanceRegistrationCreateInput.map(
({ attendanceRegistrationId }) => attendanceRegistrationId,
),
},
},
});
return this.createAttendanceRegistrations(
attendanceRegistrationCreateInput,
);
}
deleteAttendanceRegistrations(
attendanceRegistrationIds: number[],
): Promise<Prisma.BatchPayload> {
return this.prisma.attendanceRegistration.deleteMany({
where: {
attendanceRegistrationId: {
in: attendanceRegistrationIds,
},
},
});
}
findPlannedTimesInInterval(interval: Interval): Promise<PlannedTime[]> {
return this.prisma.plannedTime.findMany({
where: {
AND: [
{
startDate: {
gte: new Date(interval.start),
},
},
{
startDate: {
lte: new Date(interval.end),
},
},
],
},
});
}
createPlannedTimes(
plannedTimeCreateInput: Prisma.PlannedTimeCreateInput[],
): Promise<Prisma.BatchPayload> {
return this.prisma.plannedTime.createMany({
data: plannedTimeCreateInput,
skipDuplicates: true,
});
}
async updatePlannedTimes(
plannedTimeCreateInput: Prisma.PlannedTimeCreateInput[],
): Promise<Prisma.BatchPayload> {
await this.prisma.plannedTime.deleteMany({
where: {
id: {
in: plannedTimeCreateInput.map(({ id }) => id),
},
},
});
return this.createPlannedTimes(plannedTimeCreateInput);
}
deletePlannedTimes(plannedTimeIds: string[]): Promise<Prisma.BatchPayload> {
return this.prisma.plannedTime.deleteMany({
where: {
id: {
in: plannedTimeIds,
},
},
});
}
findWorkTimesInInterval(interval: Interval): Promise<WorkTime[]> {
return this.prisma.workTime.findMany({
where: {
AND: [
{
startDate: {
gte: new Date(interval.start),
},
},
{
startDate: {
lte: new Date(interval.end),
},
},
],
},
});
}
createWorkTimes(
workTimeCreateInput: Prisma.WorkTimeCreateInput[],
): Promise<Prisma.BatchPayload> {
return this.prisma.workTime.createMany({
data: workTimeCreateInput,
skipDuplicates: true,
});
}
async updateWorkTimes(
workTimeCreateInput: Prisma.WorkTimeCreateInput[],
): Promise<Prisma.BatchPayload> {
await this.prisma.workTime.deleteMany({
where: {
id: {
in: workTimeCreateInput.map(({ id }) => id),
},
},
});
return this.createWorkTimes(workTimeCreateInput);
}
deleteWorkTimes(workTimeIds: string[]): Promise<Prisma.BatchPayload> {
return this.prisma.workTime.deleteMany({
where: {
id: {
in: workTimeIds,
},
},
});
}
}
service/data-hub/data-connector/src/modules/feat-loaders/loaders.service.ts
+204
View File
@@ -0,0 +1,204 @@
import { Injectable } from '@nestjs/common';
import { from, map, of, switchMap } from 'rxjs';
import { TransformersService } from '../feat-transformers/transformers.service';
import { FinalRepository } from './final.repository';
import { AttendanceRegistrationLoader } from './loaders/attendance-registration.loader';
import { PatientLoader } from './loaders/patient.loader';
import { PlannedTimeLoader } from './loaders/planned-time.loader';
import { TourLoader } from './loaders/tour.loader';
import { WorkTimeLoader } from './loaders/work-time.loader';
@Injectable()
export class LoadersService {
constructor(
private readonly repository: FinalRepository,
private readonly transformersService: TransformersService,
) {}
loadTours() {
return from(this.transformersService.findTours()).pipe(
switchMap((stagingTours) => {
if (stagingTours.length === 0) {
return of({
staging: [],
final: [],
});
}
const { startDate: firstTourDate } = stagingTours.at(0);
const { startDate: lastTourDate } = stagingTours.at(-1);
return from(
this.repository.findToursInInterval(
{
start: firstTourDate,
end: lastTourDate,
},
stagingTours.map(({ operationId }) => operationId),
),
).pipe(
map((finalTours) => ({
final: finalTours,
staging: stagingTours,
})),
);
}),
map(({ staging, final }) =>
TourLoader.determineLoadOperations(staging, final),
),
switchMap(({ deleted, updated, created }) => {
return from(
Promise.all([
this.repository.deleteTours(deleted),
this.repository.updateTours(updated),
this.repository.createTours(created),
]),
);
}),
);
}
loadPatients() {
return from(this.transformersService.findPatients()).pipe(
switchMap((stagingPatients) =>
from(this.repository.findPatients()).pipe(
map((finalPatients) => ({
final: finalPatients,
staging: stagingPatients,
})),
),
),
map(({ staging, final }) =>
PatientLoader.determineLoadOperations(staging, final),
),
switchMap(({ deleted, updated, created }) =>
from(
Promise.all([
this.repository.deletePatients(deleted),
this.repository.updatePatients(updated),
this.repository.createPatients(created),
]),
),
),
);
}
loadAttendanceRegistrations() {
return from(this.transformersService.findAttendanceRegistrations()).pipe(
switchMap((stagingAttendanceRegistrations) => {
if (stagingAttendanceRegistrations.length === 0) {
return of({
staging: [],
final: [],
});
}
const { dateTime: firstAttendanceRegistration } = [
...stagingAttendanceRegistrations,
].shift();
const { dateTime: lastAttendanceRegistration } = [
...stagingAttendanceRegistrations,
].pop();
return from(
this.repository.findAttendanceRegistrationsInInterval({
start: firstAttendanceRegistration,
end: lastAttendanceRegistration,
}),
).pipe(
map((finalAttendanceRegistrations) => ({
final: finalAttendanceRegistrations,
staging: stagingAttendanceRegistrations,
})),
);
}),
map(({ staging, final }) =>
AttendanceRegistrationLoader.determineLoadOperations(staging, final),
),
switchMap(({ deleted, updated, created }) => {
return from(
Promise.all([
this.repository.deleteAttendanceRegistrations(deleted),
this.repository.updateAttendanceRegistrations(updated),
this.repository.createAttendanceRegistrations(created),
]),
);
}),
);
}
loadPlannedTimes() {
return from(this.transformersService.findPlannedTimes()).pipe(
switchMap((stagingPlannedTimes) => {
if (stagingPlannedTimes.length === 0) {
return of({
staging: [],
final: [],
});
}
const { startDate: firstPlannedTime } = [
...stagingPlannedTimes,
].shift();
const { startDate: lastPlannedTime } = [...stagingPlannedTimes].pop();
return from(
this.repository.findPlannedTimesInInterval({
start: firstPlannedTime,
end: lastPlannedTime,
}),
).pipe(
map((finalPlannedTimes) => ({
final: finalPlannedTimes,
staging: stagingPlannedTimes,
})),
);
}),
map(({ staging, final }) =>
PlannedTimeLoader.determineLoadOperations(staging, final),
),
switchMap(({ deleted, updated, created }) => {
return from(
Promise.all([
this.repository.deletePlannedTimes(deleted),
this.repository.updatePlannedTimes(updated),
this.repository.createPlannedTimes(created),
]),
);
}),
);
}
loadWorkTimes() {
return from(this.transformersService.findWorkTimes()).pipe(
switchMap((stagingWorkTimes) => {
if (stagingWorkTimes.length === 0) {
return of({
staging: [],
final: [],
});
}
const { startDate: firstWorkTime } = [...stagingWorkTimes].shift();
const { startDate: lastWorkTime } = [...stagingWorkTimes].pop();
return from(
this.repository.findWorkTimesInInterval({
start: firstWorkTime,
end: lastWorkTime,
}),
).pipe(
map((finalTours) => ({
final: finalTours,
staging: stagingWorkTimes,
})),
);
}),
map(({ staging, final }) =>
WorkTimeLoader.determineLoadOperations(staging, final),
),
switchMap(({ deleted, updated, created }) => {
return from(
Promise.all([
this.repository.deleteWorkTimes(deleted),
this.repository.updateWorkTimes(updated),
this.repository.createWorkTimes(created),
]),
);
}),
);
}
}
service/data-hub/data-connector/src/modules/feat-loaders/loaders/attendance-registration.loader.ts
+82
View File
@@ -0,0 +1,82 @@
import {
Prisma,
AttendanceRegistration,
AttendanceRegistrationStaging,
} from '@prisma/client';
export class AttendanceRegistrationLoader {
public static determineLoadOperations(
stagingAttendanceRegistrations: AttendanceRegistrationStaging[],
finalAttendanceRegistrations: AttendanceRegistration[],
): {
deleted: number[];
updated: Prisma.AttendanceRegistrationCreateInput[];
created: Prisma.AttendanceRegistrationCreateInput[];
} {
const deletedAttendanceRegistrations = finalAttendanceRegistrations.filter(
(finalAttendanceRegistration) =>
!stagingAttendanceRegistrations.find(
(stagingAttendanceRegistration) =>
stagingAttendanceRegistration.attendanceRegistrationId ===
finalAttendanceRegistration.attendanceRegistrationId,
),
);
const attendanceRegistrationsToInsert =
stagingAttendanceRegistrations.filter(
({ attendanceRegistrationId }) =>
!!attendanceRegistrationId &&
!deletedAttendanceRegistrations.find(
(deletedAttendanceRegistration) =>
deletedAttendanceRegistration.attendanceRegistrationId ===
attendanceRegistrationId,
),
);
const newAttendanceRegistrations: Prisma.AttendanceRegistrationCreateInput[] =
[];
const updatedAttendanceRegistrations: Prisma.AttendanceRegistrationCreateInput[] =
[];
for (const stagingAttendanceRegistration of attendanceRegistrationsToInsert) {
const finalAttendanceRegistration = finalAttendanceRegistrations.find(
(finalAttendanceRegistration) =>
finalAttendanceRegistration.attendanceRegistrationId ===
stagingAttendanceRegistration.attendanceRegistrationId,
);
if (finalAttendanceRegistration) {
updatedAttendanceRegistrations.push(
this.attendanceRegistrationStagingToFinal(
stagingAttendanceRegistration,
),
);
} else {
newAttendanceRegistrations.push(
this.attendanceRegistrationStagingToFinal(
stagingAttendanceRegistration,
),
);
}
}
return {
deleted: deletedAttendanceRegistrations.map(
({ attendanceRegistrationId }) => attendanceRegistrationId,
),
updated: updatedAttendanceRegistrations,
created: newAttendanceRegistrations,
};
}
private static attendanceRegistrationStagingToFinal(
attendanceregistration: AttendanceRegistrationStaging,
): Prisma.AttendanceRegistrationCreateInput {
const { attendanceRegistrationId, employeeId, dateTime, event } =
attendanceregistration;
return {
attendanceRegistrationId,
employeeId,
dateTime,
event,
personnelNumber: '',
};
}
}
service/data-hub/data-connector/src/modules/feat-loaders/loaders/patient.loader.ts
+77
View File
@@ -0,0 +1,77 @@
import { Prisma, Patient, PatientStaging } from '@prisma/client';
export class PatientLoader {
public static determineLoadOperations(
stagingPatients: PatientStaging[],
finalPatients: Patient[],
): {
deleted: string[];
updated: Prisma.PatientCreateInput[];
created: Prisma.PatientCreateInput[];
} {
const deletedPatients = finalPatients.filter(
(finalPatient) =>
!stagingPatients.find(
(stagingPatient) => stagingPatient.id === finalPatient.id,
),
);
const patientsToInsert = stagingPatients.filter(
({ id }) =>
!!id &&
!deletedPatients.find((deletedPatient) => deletedPatient.id === id),
);
const newPatients: Prisma.PatientCreateInput[] = [];
const updatedPatients: Prisma.PatientCreateInput[] = [];
for (const stagingPatient of patientsToInsert) {
const finalPatient = finalPatients.find(
(finalPatient) => finalPatient.id === stagingPatient.id,
);
if (finalPatient) {
updatedPatients.push(this.patientStagingToFinal(stagingPatient));
} else {
newPatients.push(this.patientStagingToFinal(stagingPatient));
}
}
return {
deleted: deletedPatients.map(({ id }) => id),
updated: updatedPatients,
created: newPatients,
};
}
private static patientStagingToFinal(
patient: PatientStaging,
): Prisma.PatientCreateInput {
const {
id,
surname,
name,
street,
zip,
city,
birthday,
healthinsurance,
healthinsuranceId,
careDegree,
disabilityMark1,
disabilityMark2,
} = patient;
return {
id,
surname,
name,
street,
zip,
city,
birthday,
healthinsurance,
healthinsuranceId,
careDegree,
disabilityMark1,
disabilityMark2,
};
}
}
service/data-hub/data-connector/src/modules/feat-loaders/loaders/planned-time.loader.ts
+75
View File
@@ -0,0 +1,75 @@
import { Prisma, PlannedTime, PlannedTimeStaging } from '@prisma/client';
export class PlannedTimeLoader {
public static determineLoadOperations(
stagingPlannedTimes: PlannedTimeStaging[],
finalPlannedTimes: PlannedTime[],
): {
deleted: string[];
updated: Prisma.PlannedTimeCreateInput[];
created: Prisma.PlannedTimeCreateInput[];
} {
const deletedPlannedTimes = stagingPlannedTimes.filter(
({ deleted }) => deleted,
);
console.log(deletedPlannedTimes.length);
const plannedTimesToInsert = stagingPlannedTimes.filter(
({ id }) =>
!!id &&
!deletedPlannedTimes.find(
(deletedPlannedTime) => deletedPlannedTime.id === id,
),
);
const newPlannedTimes: Prisma.PlannedTimeCreateInput[] = [];
const updatedPlannedTimes: Prisma.PlannedTimeCreateInput[] = [];
for (const stagingPlannedTime of plannedTimesToInsert) {
const finalPlannedTime = finalPlannedTimes.find(
(finalPlannedTime) =>
finalPlannedTime.id === `${stagingPlannedTime.id}`,
);
if (finalPlannedTime) {
updatedPlannedTimes.push(
this.plannedTimeStagingToFinal(stagingPlannedTime),
);
} else {
newPlannedTimes.push(
this.plannedTimeStagingToFinal(stagingPlannedTime),
);
}
}
return {
deleted: deletedPlannedTimes.map(({ id }) => `${id}`),
updated: updatedPlannedTimes,
created: newPlannedTimes,
};
}
private static plannedTimeStagingToFinal(
plannedtime: PlannedTimeStaging,
): Prisma.PlannedTimeCreateInput {
const {
id,
employeeId,
firstname,
surname,
departmentName,
startDate,
endDate,
mark,
} = plannedtime;
return {
firstname,
surname,
startDate,
endDate,
id: `${id}`,
userId: `${employeeId}`,
department: departmentName,
pause: 0,
duration: 0,
note: mark,
};
}
}
service/data-hub/data-connector/src/modules/feat-loaders/loaders/tour.loader.ts
+149
View File
@@ -0,0 +1,149 @@
import { Prisma, Tour, TourStaging } from '@prisma/client';
export class TourLoader {
public static determineLoadOperations(
stagingTours: TourStaging[],
finalTours: Tour[],
): {
deleted: string[];
updated: Prisma.TourCreateInput[];
created: Prisma.TourCreateInput[];
} {
const deletedTours = finalTours.filter(
(finalTour) =>
!stagingTours.find(
(stagingTour) => stagingTour.operationId === finalTour.operationId,
),
);
const toursToInsert = stagingTours.filter(
({ id, operationId }) =>
!!id &&
!deletedTours.find(
(deletedTour) => deletedTour.operationId === operationId,
),
);
const newTours: Prisma.TourCreateInput[] = [];
const updatedTours: Prisma.TourCreateInput[] = [];
for (const stagingTour of toursToInsert) {
const finalTour = finalTours.find(
(finalTour) => finalTour.operationId === stagingTour.operationId,
);
if (finalTour) {
updatedTours.push(
this.tourStagingToFinal(stagingTour, finalTour.ticketId),
);
} else {
newTours.push(this.tourStagingToFinal(stagingTour));
}
}
return {
deleted: deletedTours.map(({ operationId }) => operationId),
updated: updatedTours,
created: newTours,
};
}
private static tourStagingToFinal(tour: TourStaging, ticketId?: string) {
const {
done,
direction,
carName,
operationId,
driverId,
driverName,
codriverId,
codriverName,
patientId,
patientName,
patientSurname,
patientStreet,
patientZip,
patientCity,
healthInsurance,
healthInsuranceNumber,
type,
category,
transportType,
ordinanceType,
rangeEndDate,
startInstitution,
startStreet,
startZip,
startCity,
targetInstitution,
targetStreet,
targetZip,
targetCity,
startDate,
check,
occupiedKm,
totalKm,
startBegin,
startEnd,
target,
targetBegin,
empty,
hasInfection,
infectionName,
revenueDispoLive,
revenue,
revenueDeviation,
consumptionCosts,
billDate,
billNumber,
createdAt,
} = tour;
return {
done,
direction,
carName,
operationId,
driverId,
driverName,
codriverId,
codriverName,
patientId,
patientName,
patientSurname,
patientStreet,
patientZip,
patientCity,
healthInsurance,
healthInsuranceNumber,
type,
category,
transportType,
ordinanceType,
rangeEndDate,
startInstitution,
startStreet,
startZip,
startCity,
targetInstitution,
targetStreet,
targetZip,
targetCity,
startDate,
check,
occupiedKm,
totalKm,
startBegin,
startEnd,
target,
targetBegin,
empty,
hasInfection,
infectionName,
revenueDispoLive,
revenue,
revenueDeviation,
consumptionCosts,
billDate,
billNumber,
createdAt,
...(ticketId ? { ticketId } : {}),
};
}
}
service/data-hub/data-connector/src/modules/feat-loaders/loaders/work-time.loader.ts
+70
View File
@@ -0,0 +1,70 @@
import { Prisma, WorkTime, WorkTimeStaging } from '@prisma/client';
export class WorkTimeLoader {
public static determineLoadOperations(
stagingWorkTimes: WorkTimeStaging[],
finalWorkTimes: WorkTime[],
): {
deleted: string[];
updated: Prisma.WorkTimeCreateInput[];
created: Prisma.WorkTimeCreateInput[];
} {
const deletedWorkTimes = finalWorkTimes.filter(
(finalWorkTime) =>
!stagingWorkTimes.find(
(stagingWorkTime) => `${stagingWorkTime.id}` === finalWorkTime.id,
),
);
const workTimesToInsert = stagingWorkTimes.filter(
({ id }) =>
!!id &&
!deletedWorkTimes.find(
(deletedWorkTime) => deletedWorkTime.id === `${id}`,
),
);
const newWorkTimes: Prisma.WorkTimeCreateInput[] = [];
const updatedWorkTimes: Prisma.WorkTimeCreateInput[] = [];
for (const stagingWorkTime of workTimesToInsert) {
const finalWorkTime = finalWorkTimes.find(
(finalWorkTime) => finalWorkTime.id === `${stagingWorkTime.id}`,
);
if (finalWorkTime) {
updatedWorkTimes.push(this.workTimeStagingToFinal(stagingWorkTime));
} else {
newWorkTimes.push(this.workTimeStagingToFinal(stagingWorkTime));
}
}
return {
deleted: deletedWorkTimes.map(({ id }) => id),
updated: updatedWorkTimes,
created: newWorkTimes,
};
}
private static workTimeStagingToFinal(
worktime: WorkTimeStaging,
): Prisma.WorkTimeCreateInput {
const {
id,
employeeId,
firstname,
surname,
departmentName,
startDate,
endDate,
} = worktime;
return {
firstname,
surname,
startDate,
endDate,
id: `${id}`,
userId: `${employeeId}`,
department: departmentName,
pause: 0,
duration: 0,
};
}
}
service/data-hub/data-connector/src/modules/feat-transformers/adapter/tours-calculation.adapter.ts
+54
View File
@@ -0,0 +1,54 @@
// import { CalculationService } from '@dks/avicenna/api-util-calculation';
// import { Injectable } from '@nestjs/common';
// import { Tour } from '@prisma/client';
// import { chunk } from 'src/utils';
// import { TourPreStaging } from '../staging.interface';
// @Injectable()
// export class ToursCalculationAdapater {
// constructor(private readonly calculationService: CalculationService) {}
// public async addRevenueAndConsumptionCostsToTours(
// tours: TourPreStaging[],
// ): Promise<TourPreStaging[]> {
// const resultTours = [];
// for (const tourChunk of chunk(tours, 5)) {
// resultTours.push(
// await Promise.all(
// tourChunk.map((tour) =>
// this.addRevenueAndConsumptionCostsToTour(tour),
// ),
// ),
// );
// }
// return resultTours.flat();
// }
// public async addRevenueAndConsumptionCostsToTour(
// tour: TourPreStaging,
// ): Promise<TourPreStaging> {
// if (!tour?.ordinanceType) {
// return tour;
// }
// const revenue = await this.calculationService.calculateTourRevenue(
// tour as unknown as Tour,
// );
// const revenueDeviation =
// tour.revenueDispoLive > 0
// ? Math.floor((revenue - tour.revenueDispoLive) * 100) / 100
// : 0;
// const { consumptionCostsPerTour } =
// this.calculationService.getBillingInformationByOrdinanceType(
// tour.ordinanceType.match(/([A-Z])\w+/).at(0),
// tour.startDate,
// );
// return {
// ...tour,
// revenue,
// revenueDeviation,
// consumptionCosts: consumptionCostsPerTour,
// };
// }
// }
service/data-hub/data-connector/src/modules/feat-transformers/data-access-transformers.module.ts
+13
View File
@@ -0,0 +1,13 @@
import { Module } from '@nestjs/common';
// import { ToursCalculationAdapater } from './adapter/tours-calculation.adapter';
import { StagingRepository } from './staging.repository';
import { TransformersService } from './transformers.service';
import { PrismaModule } from 'src/core/database/prisma.module';
import { DataAccessExtractorsModule } from '../feat-extractors/data-access-extractors.module';
@Module({
imports: [PrismaModule, DataAccessExtractorsModule],
providers: [TransformersService, StagingRepository],
exports: [TransformersService],
})
export class DataAccessTransformersModule {}
service/data-hub/data-connector/src/modules/feat-transformers/staging.interface.ts
+117
View File
@@ -0,0 +1,117 @@
export interface TourPreStaging {
id?: string;
done: boolean;
direction: string;
carName?: string;
operationId: string;
driverId?: string;
driverName?: string;
codriverId?: string;
codriverName?: string;
patientId?: string;
patientName?: string;
patientSurname?: string;
patientStreet?: string;
patientZip?: string;
patientCity?: string;
healthInsurance?: string;
healthInsuranceNumber?: string;
type?: string;
category?: string;
transportType?: string;
ordinanceType?: string;
rangeEndDate?: string;
startInstitution?: string;
startStreet: string;
startZip: string;
startCity: string;
targetInstitution?: string;
targetStreet: string;
targetZip: string;
targetCity: string;
startDate: Date;
check: number;
occupiedKm?: number;
totalKm?: number;
startBegin?: string;
startEnd?: string;
target?: string;
targetBegin?: string;
empty?: string;
hasInfection: boolean;
infectionName?: string;
revenueDispoLive?: number;
billDate?: Date;
billNumber?: string;
revenue?: number;
revenueDeviation?: number;
consumptionCosts?: number;
createdAt?: Date;
}
export interface PatientPreStaging {
id?: string;
surname?: string;
name?: string;
street?: string;
zip?: string;
city?: string;
birthday?: Date;
healthinsuranceId?: string;
healthinsurance?: string;
careDegree?: string;
disabilityMark1?: string;
disabilityMark2?: string;
}
export interface AttendanceRegistrationPreStaging {
attendanceRegistrationId: number;
employeeId: number;
dateTime: Date;
event: string;
}
export interface PlannedTimePreStaging {
id: number;
employeeId: number;
firstname?: string;
surname?: string;
departmentName: string;
startDate: Date;
endDate: Date;
deleted: boolean;
mark?: string;
}
export interface WorkTimePreStaging {
id: number;
employeeId: number;
firstname: string;
surname: string;
departmentName: string;
startDate: Date;
endDate: Date;
}
export interface EmployeePreStaging {
id: number;
firstname: string;
surname: string;
contractStart: Date;
contractEnd: Date;
contractHoursWeek: number;
contractSalaryHour: number;
personTransportCertificate?: Date;
qualification?: string;
}
service/data-hub/data-connector/src/modules/feat-transformers/staging.repository.ts
+118
View File
@@ -0,0 +1,118 @@
import { Injectable } from '@nestjs/common';
import {
AttendanceRegistrationStaging,
PatientStaging,
PlannedTimeStaging,
Prisma,
TourStaging,
WorkTimeStaging,
} from '@prisma/client';
import { PrismaService } from 'src/core/database/prisma.service';
import {
AttendanceRegistrationPreStaging,
EmployeePreStaging,
PatientPreStaging,
PlannedTimePreStaging,
TourPreStaging,
WorkTimePreStaging,
} from './staging.interface';
@Injectable()
export class StagingRepository {
constructor(private readonly prisma: PrismaService) {}
clearStaging(): Promise<Prisma.BatchPayload[]> {
return Promise.all([
this.prisma.tourStaging.deleteMany(),
this.prisma.employeeStaging.deleteMany(),
this.prisma.workTimeStaging.deleteMany(),
this.prisma.plannedTimeStaging.deleteMany(),
this.prisma.attendanceRegistrationStaging.deleteMany(),
this.prisma.patientStaging.deleteMany(),
]);
}
createPatients(patients: PatientPreStaging[]): Promise<Prisma.BatchPayload> {
return this.prisma.patientStaging.createMany({
data: patients,
});
}
createPlannedTimes(
plannedTimes: PlannedTimePreStaging[],
): Promise<Prisma.BatchPayload> {
return this.prisma.plannedTimeStaging.createMany({
data: plannedTimes,
});
}
createWorkTimes(
workTimes: WorkTimePreStaging[],
): Promise<Prisma.BatchPayload> {
return this.prisma.workTimeStaging.createMany({
data: workTimes,
});
}
createAttendanceRegistrations(
attendanceRegistrations: AttendanceRegistrationPreStaging[],
): Promise<Prisma.BatchPayload> {
return this.prisma.attendanceRegistrationStaging.createMany({
data: attendanceRegistrations,
});
}
createTours(tours: TourPreStaging[]): Promise<Prisma.BatchPayload> {
return this.prisma.tourStaging.createMany({
data: tours,
});
}
createEmployees(
employees: EmployeePreStaging[],
): Promise<Prisma.BatchPayload> {
return this.prisma.employeeStaging.createMany({
data: employees,
});
}
findPatients(): Promise<PatientStaging[]> {
return this.prisma.patientStaging.findMany({
orderBy: {
surname: 'asc',
},
});
}
findTours(): Promise<TourStaging[]> {
return this.prisma.tourStaging.findMany({
orderBy: {
startDate: 'asc',
},
});
}
findAttendanceRegistrations(): Promise<AttendanceRegistrationStaging[]> {
return this.prisma.attendanceRegistrationStaging.findMany({
orderBy: {
dateTime: 'asc',
},
});
}
findPlannedTimes(): Promise<PlannedTimeStaging[]> {
return this.prisma.plannedTimeStaging.findMany({
orderBy: {
startDate: 'asc',
},
});
}
findWorkTimes(): Promise<WorkTimeStaging[]> {
return this.prisma.workTimeStaging.findMany({
orderBy: {
startDate: 'asc',
},
});
}
}
service/data-hub/data-connector/src/modules/feat-transformers/transformer.functions.ts
+54
View File
@@ -0,0 +1,54 @@
import { parse, parseISO } from 'date-fns';
import { fromZonedTime } from 'date-fns-tz';
export const parseDateTime = (format: string) => (value: string) =>
parse(value, format, new Date());
export const parseIsoDateTime = (value: string) => parseISO(value);
export const parseZonedDateTime =
(tz: string) => (format: string) => (value: string) =>
fromZonedTime(parse(value, format, new Date()), tz);
export const parseCETDateTime = parseZonedDateTime('Europe/Berlin');
export const parseInteger = (value = '') => parseInt(value);
export const parseFloat = (value = '') =>
Number((value ?? '').replace(',', '.'));
export const parseBoolean = (value = '') => value === '1' || value === 'true';
export const pick =
<T extends Record<string, T[keyof T]>>(keys: (keyof T)[]) =>
(value: Record<keyof T, T[keyof T]>) =>
keys.reduce(
(obj, key) => {
if (value && Object.prototype.hasOwnProperty.call(value, key)) {
obj[key] = value[key];
}
return obj;
},
{} as Record<keyof T, T[keyof T]>,
);
export const pick2 = <T extends Record<string, unknown>>(
keys: (keyof T)[],
value: T,
) =>
keys.reduce((obj, key) => {
if (value && Object.prototype.hasOwnProperty.call(value, key)) {
obj[key] = value[key];
}
return obj;
}, {} as T);
export const transformProperties =
<T extends string, U>(transformer: (value: T) => U) =>
(value: Record<string, T>) =>
Object.entries(value).reduce(
(obj, [key, value]) => ({
...obj,
[key]: transformer(value),
}),
{} as Record<string, U>,
);
service/data-hub/data-connector/src/modules/feat-transformers/transformers.service.ts
+149
View File
@@ -0,0 +1,149 @@
import { Injectable } from '@nestjs/common';
import {
AttendanceRegistrationStaging,
EmployeeDispoLiveSource,
EmployeeDyflexisSource,
PatientStaging,
PlannedTimeStaging,
Prisma,
TourStaging,
WorkTimeStaging,
} from '@prisma/client';
import { combineLatest, from, map, switchMap } from 'rxjs';
import { ExtractorsService } from '../feat-extractors/extractors.service';
// import { ToursCalculationAdapater } from './adapter/tours-calculation.adapter';
import { StagingRepository } from './staging.repository';
import { AttendanceRegistrationTransformer } from './transformers/attendance-registration.transformer';
import { EmployeeTransformer } from './transformers/employee.transformer';
import { PatientTransformer } from './transformers/patient.transformer';
import { PlannedTimeTransformer } from './transformers/planned-time.transformer';
import { TourTransformer } from './transformers/tour.transformer';
import { WorkTimeTransformer } from './transformers/work-time.transformer';
@Injectable()
export class TransformersService {
constructor(
private readonly extractorsService: ExtractorsService,
private readonly repository: StagingRepository,
// private readonly tourCalculationAdapter: ToursCalculationAdapater,
) {}
clearStaging(): Promise<Prisma.BatchPayload[]> {
return this.repository.clearStaging();
}
transformPlannedTimes(jobId: string) {
return from(this.extractorsService.findPlannedTimesByJobId(jobId)).pipe(
map((plannedTimes) =>
plannedTimes.map((plannedTime) =>
PlannedTimeTransformer.transform(plannedTime),
),
),
switchMap((plannedTimes) =>
this.repository.createPlannedTimes(plannedTimes),
),
);
}
transformWorkTimes(jobId: string) {
return from(this.extractorsService.findWorkTimesByJobId(jobId)).pipe(
map((workTimes) =>
workTimes.map((workTime) => WorkTimeTransformer.transform(workTime)),
),
switchMap((workTimes) => this.repository.createWorkTimes(workTimes)),
);
}
transformAttendanceRegistrations(jobId: string) {
return from(
this.extractorsService.findAttendanceRegistrationsByJobId(jobId),
).pipe(
map((attendanceRegistrations) =>
attendanceRegistrations.map((attendanceRegistration) =>
AttendanceRegistrationTransformer.transform(attendanceRegistration),
),
),
switchMap((attendanceRegistrations) =>
this.repository.createAttendanceRegistrations(attendanceRegistrations),
),
);
}
transformTours(jobId: string) {
return from(this.extractorsService.findToursByJobId(jobId)).pipe(
// TODO: Add validation filter
map((tours) =>
tours
.filter(({ einsatzNummer }) => !!einsatzNummer)
.map((tour) => TourTransformer.transform(tour)),
),
// switchMap((tours) =>
// from(
// this.tourCalculationAdapter.addRevenueAndConsumptionCostsToTours(
// tours,
// ),
// ),
// ),
switchMap((tours) => this.repository.createTours(tours)),
);
}
transformEmployees(jobId: string) {
return combineLatest([
from(this.extractorsService.findEmployeesDispoLiveByJobId(jobId)),
from(this.extractorsService.findEmployeesDyflexisByJobId(jobId)),
]).pipe(
map(([dispoLiveEmployees, dyflexisEmployees]) =>
dyflexisEmployees
.map((dyflexisEmployee) => {
const dispoLiveEmployee = dispoLiveEmployees.find(
(e) => e.importPersId === dyflexisEmployee.id,
);
if (!dispoLiveEmployee) return undefined;
return [dyflexisEmployee, dispoLiveEmployee];
})
.filter(Boolean),
),
map((employees) =>
employees.map(([dyflexisEmployee, dispoLiveEmployee]) =>
EmployeeTransformer.transform(
dyflexisEmployee as EmployeeDyflexisSource,
dispoLiveEmployee as EmployeeDispoLiveSource,
),
),
),
switchMap((employees) => this.repository.createEmployees(employees)),
);
}
transformPatients(jobId: string) {
return from(this.extractorsService.findPatientsByJobId(jobId)).pipe(
map((patients) =>
patients.map((patient) => PatientTransformer.transform(patient)),
),
switchMap((patients) => this.repository.createPatients(patients)),
);
}
findPatients(): Promise<PatientStaging[]> {
return this.repository.findPatients();
}
findTours(): Promise<TourStaging[]> {
return this.repository.findTours();
}
findAttendanceRegistrations(): Promise<AttendanceRegistrationStaging[]> {
return this.repository.findAttendanceRegistrations();
}
findPlannedTimes(): Promise<PlannedTimeStaging[]> {
return this.repository.findPlannedTimes();
}
findWorkTimes(): Promise<WorkTimeStaging[]> {
return this.repository.findWorkTimes();
}
}
service/data-hub/data-connector/src/modules/feat-transformers/transformers/attendance-registration.transformer.ts
+36
View File
@@ -0,0 +1,36 @@
import { AttendanceRegistrationSource } from '@prisma/client';
import { AttendanceRegistrationPreStaging } from '../staging.interface';
import {
parseInteger,
parseIsoDateTime,
pick,
transformProperties,
} from '../transformer.functions';
export class AttendanceRegistrationTransformer {
public static transform(
attendanceRegistrationSource: AttendanceRegistrationSource,
): AttendanceRegistrationPreStaging {
const attendanceRegistrationParsed = {
...attendanceRegistrationSource,
...transformProperties(parseInteger)(
pick(['attendanceRegistrationId', 'employeeId'])(
attendanceRegistrationSource,
),
),
...transformProperties(parseIsoDateTime)(
pick(['dateTime'])(attendanceRegistrationSource),
),
};
const { attendanceRegistrationId, employeeId, dateTime, event } =
attendanceRegistrationParsed;
return {
attendanceRegistrationId,
employeeId,
dateTime,
event,
} as unknown as AttendanceRegistrationPreStaging;
}
}
service/data-hub/data-connector/src/modules/feat-transformers/transformers/employee.transformer.ts
+67
View File
@@ -0,0 +1,67 @@
import {
EmployeeDispoLiveSource,
EmployeeDyflexisSource,
} from '@prisma/client';
import { isValidDate } from 'src/utils';
import { EmployeePreStaging } from '../staging.interface';
import {
parseDateTime,
parseFloat,
parseInteger,
pick,
transformProperties,
} from '../transformer.functions';
export class EmployeeTransformer {
public static transform(
dyflexisEmployeeSource: EmployeeDyflexisSource,
dispoLiveEmployeeSource: EmployeeDispoLiveSource,
): EmployeePreStaging {
const dyflexisEmployeeParsed = {
...dyflexisEmployeeSource,
...transformProperties(parseFloat)(
pick(['contract_hours_week', 'contract_salary_hour'])(
dyflexisEmployeeSource,
),
),
...transformProperties(parseInteger)(
pick(['id'])(dyflexisEmployeeSource),
),
...transformProperties(parseDateTime('yyyy-MM-dd'))(
pick(['contract_start', 'contract_end'])(dyflexisEmployeeSource),
),
};
const dispoLiveEmployeeParsed = {
...dispoLiveEmployeeSource,
...transformProperties(parseDateTime('yyyy-MM-dd'))(
pick(['pScheinAblaufDatum'])(dispoLiveEmployeeSource),
),
};
const {
id,
firstname,
surname,
contract_start,
contract_end,
contract_hours_week,
contract_salary_hour,
} = dyflexisEmployeeParsed;
const { pScheinAblaufDatum, qualifikation } = dispoLiveEmployeeParsed;
return {
id,
firstname,
surname,
contractStart: contract_start,
contractEnd: contract_end,
contractHoursWeek: contract_hours_week,
contractSalaryHour: contract_salary_hour,
personTransportCertificate: isValidDate(new Date(pScheinAblaufDatum))
? pScheinAblaufDatum
: undefined,
qualification: qualifikation,
} as unknown as EmployeePreStaging;
}
}
service/data-hub/data-connector/src/modules/feat-transformers/transformers/patient.transformer.ts
+50
View File
@@ -0,0 +1,50 @@
import { PatientSource } from '@prisma/client';
import { isValidDate } from 'src/utils';
import { PatientPreStaging } from '../staging.interface';
import {
parseDateTime,
pick,
transformProperties,
} from '../transformer.functions';
export class PatientTransformer {
public static transform(patientSource: PatientSource): PatientPreStaging {
const patientParsed = {
...patientSource,
...transformProperties(parseDateTime('yyyy-MM-dd'))(
pick(['birthday'])(patientSource),
),
};
const {
id,
surname,
name,
street,
zip,
city,
birthday,
kkId,
krankenkasse,
pflegestufe,
behinderung1,
behinderung2,
} = patientParsed;
return {
id,
surname,
name,
street,
zip,
city,
birthday: isValidDate(new Date(birthday)) ? birthday : undefined,
healthinsurance: krankenkasse,
healthinsuranceId: kkId,
careDegree: pflegestufe,
disabilityMark1: behinderung1,
disabilityMark2: behinderung2,
} as unknown as PatientPreStaging;
}
}
service/data-hub/data-connector/src/modules/feat-transformers/transformers/planned-time.transformer.ts
+52
View File
@@ -0,0 +1,52 @@
import { PlannedTimeSource } from '@prisma/client';
import { PlannedTimePreStaging } from '../staging.interface';
import {
parseBoolean,
parseCETDateTime,
parseInteger,
pick,
transformProperties,
} from '../transformer.functions';
export class PlannedTimeTransformer {
public static transform(
plannedTimeSource: PlannedTimeSource,
): PlannedTimePreStaging {
const plannedTimeParsed = {
...plannedTimeSource,
...transformProperties(parseInteger)(
pick(['id', 'user_id'])(plannedTimeSource),
),
...transformProperties(parseCETDateTime('yyyy-MM-dd HH:mm:ss'))(
pick(['start_date', 'end_date'])(plannedTimeSource),
),
...transformProperties(parseBoolean)(
pick(['deleted'])(plannedTimeSource),
),
};
const {
id,
user_id,
firstname,
surname,
department_name,
start_date,
end_date,
deleted,
mark,
} = plannedTimeParsed;
return {
id,
firstname,
surname,
deleted,
mark,
employeeId: user_id,
departmentName: department_name,
startDate: start_date,
endDate: end_date,
} as unknown as PlannedTimePreStaging;
}
}
service/data-hub/data-connector/src/modules/feat-transformers/transformers/tour.transformer.ts
+142
View File
@@ -0,0 +1,142 @@
import { TourSource } from '@prisma/client';
import { TourPreStaging } from '../staging.interface';
import {
parseBoolean,
parseCETDateTime,
parseDateTime,
parseFloat,
parseInteger,
parseIsoDateTime,
pick,
transformProperties,
} from '../transformer.functions';
export class TourTransformer {
public static transform(tourSource: TourSource): TourPreStaging {
const tourParsed = {
...tourSource,
...transformProperties(parseInteger)(pick(['check'])(tourSource)),
...transformProperties(parseFloat)(
pick(['besetztKm', 'gesamtPreis'])(tourSource),
),
...transformProperties(parseIsoDateTime)(pick(['date_iso'])(tourSource)),
...(tourSource.abrDate
? transformProperties(parseDateTime('yyyy-MM-dd'))(
pick(['abrDate'])(tourSource),
)
: undefined),
...(tourSource.endgen
? transformProperties(parseDateTime('yyyy-MM-dd'))(
pick(['endgen'])(tourSource),
)
: undefined),
...(tourSource.createTime.length > 10
? transformProperties(parseCETDateTime('yyyy-MM-dd HH:mm:ss'))(
pick(['createTime'])(tourSource),
)
: transformProperties(parseCETDateTime('yyyy-MM-dd'))(
pick(['createTime'])(tourSource),
)),
...transformProperties(parseBoolean)(
pick(['infektion', 'gefahren'])(tourSource),
),
};
const {
id,
gefahren,
direction,
carName,
einsatzNummer,
fahrer,
fahrerName,
beifahrer,
beifahrerName,
patId,
patName,
patSurname,
patStreet,
patZip,
patCity,
krankenkasse,
kkId,
type,
serienKategorName,
transportKrz,
verordnungsName,
endgen,
startInstitution,
startStreet,
startZip,
startCity,
targetInstitution,
targetStreet,
targetZip,
targetCity,
date_iso,
check,
besetztKm,
startBegin,
startEnd,
ziel,
zielBegin,
frei,
infektion,
itKrz,
gesamtPreis,
abrDate,
abrNum,
createTime,
} = tourParsed;
return {
id,
direction,
carName,
startInstitution,
startStreet,
startZip,
startCity,
targetInstitution,
targetStreet,
targetZip,
targetCity,
check,
type,
done: gefahren,
operationId: einsatzNummer,
driverId: fahrer,
driverName: fahrerName,
codriverId: beifahrer,
codriverName: beifahrerName,
patientId: patId,
patientName: patName,
patientSurname: patSurname,
patientStreet: patStreet,
patientZip: patZip,
patientCity: patCity,
healthInsurance: krankenkasse,
healthInsuranceNumber: kkId,
category: serienKategorName,
transportType: transportKrz,
ordinanceType: verordnungsName,
rangeEndDate: endgen,
startDate: date_iso,
occupiedKm: besetztKm,
totalKm: besetztKm,
startBegin,
startEnd,
target: ziel,
targetBegin: zielBegin,
empty: frei,
hasInfection: infektion,
infectionName: itKrz,
revenueDispoLive: gesamtPreis,
billDate: abrDate,
billNumber: abrNum,
createdAt: createTime,
revenue: 0,
consumptionCosts: 0,
} as unknown as TourPreStaging;
}
}
service/data-hub/data-connector/src/modules/feat-transformers/transformers/work-time.transformer.ts
+42
View File
@@ -0,0 +1,42 @@
import { WorkTimeSource } from '@prisma/client';
import { WorkTimePreStaging } from '../staging.interface';
import {
parseCETDateTime,
parseInteger,
pick,
transformProperties,
} from '../transformer.functions';
export class WorkTimeTransformer {
public static transform(workTimeSource: WorkTimeSource): WorkTimePreStaging {
const workTimeParsed = {
...workTimeSource,
...transformProperties(parseInteger)(
pick(['id', 'user_id'])(workTimeSource),
),
...transformProperties(parseCETDateTime('yyyy-MM-dd HH:mm:ss'))(
pick(['start_date', 'end_date'])(workTimeSource),
),
};
const {
id,
user_id,
firstname,
surname,
department_name,
start_date,
end_date,
} = workTimeParsed;
return {
id,
firstname,
surname,
employeeId: user_id,
departmentName: department_name,
startDate: start_date,
endDate: end_date,
} as unknown as WorkTimePreStaging;
}
}
service/data-hub/data-connector/src/utils/chunk.spec.ts
+19
View File
@@ -0,0 +1,19 @@
import { chunk } from './chunk';
import { range } from './range';
describe('chunk', () => {
it('should split an array with 10 elements into two chunks with 5 elements', () => {
const given = range(10);
const when = chunk(given, 5);
expect(when).toHaveLength(2);
when.map((sub) => expect(sub).toHaveLength(5));
});
it('should split an array with 8 elements into two chunks, first with 5 elements, second with 3', () => {
const given = range(8);
const when = chunk(given, 5);
expect(when).toHaveLength(2);
expect(when[0]).toHaveLength(5);
expect(when[1]).toHaveLength(3);
});
});
service/data-hub/data-connector/src/utils/chunk.ts
+9
View File
@@ -0,0 +1,9 @@
export function chunk<T>(input: T[], size: number): T[][] {
return input.reduce(
(array, item, index) =>
index % size === 0
? [...array, [item]]
: [...array.slice(0, -1), [...array.slice(-1)[0], item]],
[] as T[][]
);
}
service/data-hub/data-connector/src/utils/index.ts
+3
View File
@@ -0,0 +1,3 @@
export * from './chunk';
export * from './is-valid-date';
export * from './range';
service/data-hub/data-connector/src/utils/is-valid-date.spec.ts
+12
View File
@@ -0,0 +1,12 @@
import { isValidDate } from './is-valid-date';
describe('isValidDate', () => {
it('should return false if date is invalid', () => {
const given = new Date(NaN);
expect(isValidDate(given)).toBe(false);
});
it('should return true if date is valid', () => {
const given = new Date();
expect(isValidDate(given)).toBe(true);
});
});
service/data-hub/data-connector/src/utils/is-valid-date.ts
+3
View File
@@ -0,0 +1,3 @@
export function isValidDate(date: Date): boolean {
return date instanceof Date && !isNaN(date.getTime());
}
service/data-hub/data-connector/src/utils/range.spec.ts
+9
View File
@@ -0,0 +1,9 @@
import { range } from './range';
describe('range', () => {
it('should generate an array with the supplied length and numbers ascending from 0', () => {
const when = range(10);
expect(when).toHaveLength(10);
when.map((item, index) => expect(item).toBe(index));
});
});
service/data-hub/data-connector/src/utils/range.ts
+3
View File
@@ -0,0 +1,3 @@
export function range(length: number): number[] {
return Array.from({ length }, (_, i) => i);
}