diff --git a/src/main/java/ru/nbch/credit_tracker/entities/app/Hit.java b/src/main/java/ru/nbch/credit_tracker/entities/app/Hit.java index 9a7630b..47ecf04 100644 --- a/src/main/java/ru/nbch/credit_tracker/entities/app/Hit.java +++ b/src/main/java/ru/nbch/credit_tracker/entities/app/Hit.java @@ -1,12 +1,14 @@ package ru.nbch.credit_tracker.entities.app; -import java.time.LocalDateTime; +import lombok.Builder; import lombok.Getter; import lombok.Setter; +import java.time.LocalDateTime; + @Getter @Setter -//@Builder +@Builder public class Hit { private Long id; private Long phone; @@ -15,4 +17,13 @@ public class Hit { private Integer signalType; private LocalDateTime responseSentAt; private String reportFileName; + private Long inn; + private Long regNum; + private String busCategory; + private LocalDateTime inqDt; + private String inqP; + private Integer inqT; + private Long inqA; + private Integer segment; + private LocalDateTime recordedAt; } diff --git a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringService.java b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringService.java index 833c710..fe8fb4e 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringService.java @@ -19,6 +19,7 @@ import ru.nbch.credit_tracker.model.SignalKafkaEvent; import ru.nbch.credit_tracker.model.UserPackageLegalData; import ru.nbch.credit_tracker.model.UserPackagePhoneData; import ru.nbch.credit_tracker.service.nbch.SourceService; +import ru.nbch.credit_tracker.service.signal.HitsService; import ru.nbch.credit_tracker.service.signal.PackageRecordsService; import ru.nbch.credit_tracker.service.signal.SegmentMapService; import ru.nbch.credit_tracker.utils.CTUtil; @@ -45,17 +46,20 @@ public class KafkaMonitoringService { private final PackageRecordsService packageRecordsService; private final SourceService sourceService; private final SegmentMapService segmentMapService; + private final HitsService hitsService; public KafkaMonitoringService(@Qualifier("signalKafkaEventQueue") BlockingQueue signalEventsQueue, PackageRecordsService packageRecordsService, SourceService sourceService, SegmentMapService segmentMapService, - CreditTrackerSettings settings) { + CreditTrackerSettings settings, + HitsService hitsService) { this.signalEventsQueue = signalEventsQueue; this.packageRecordsService = packageRecordsService; this.sourceService = sourceService; this.segmentMapService = segmentMapService; this.permits = Integer.parseInt(settings.getKafkaConsumerSettings().concurency()); // количество равно concurency листенера + this.hitsService = hitsService; this.processingPermits = new Semaphore(permits); } @@ -106,7 +110,8 @@ public class KafkaMonitoringService { List lpHits = processLpMessages(lpMessages, segmentMap); hits.addAll(lpHits); } - // TODO insert hits + + hitsService.insertHits(hits); } private List processNpMessages(List messages, Map segmentMap) { @@ -186,6 +191,21 @@ public class KafkaMonitoringService { event.setSegment(segment != null ? segment : -1); putToQueue(event); + Hit hit = Hit.builder() + .packageId(userPackagePhoneData.getPackageId()) + .subjectId(subjectId) + .signalType(signalType) + .reportFileName("") // TODO + .inqT(message.getInquiryType()) + .busCategory(busCategory) + .inqDt(message.getInquiryDate()) + .inqP(message.getInquiryPurpose()) + .inqA(message.getInquiryAmount()) + .segment(segment != null ? segment : -1) + .recordedAt(LocalDateTime.now()) + .phone(userPackagePhoneData.getPhone()) + .build(); + hits.add(hit); } } } @@ -251,15 +271,15 @@ public class KafkaMonitoringService { continue; } - for (UserPackageLegalData userPackagePhoneData : userPackageLegalDataList) { - String subjectId = userPackagePhoneData.getSubjectId(); - String signalTypes = userPackagePhoneData.getSignalTypes(); + for (UserPackageLegalData userPackageLegalData : userPackageLegalDataList) { + String subjectId = userPackageLegalData.getSubjectId(); + String signalTypes = userPackageLegalData.getSignalTypes(); if (signalTypes.contains(Integer.toString(signalType))) { SignalKafkaEvent event = new SignalKafkaEvent(); event.setEventType(EventType.WRITE); event.setNatural(false); - event.setUserPackageLegalData(userPackagePhoneData); + event.setUserPackageLegalData(userPackageLegalData); event.setPackageMessage(message); event.setSubjectId(subjectId); event.setFlag90days(flag90days); @@ -271,6 +291,22 @@ public class KafkaMonitoringService { event.setSegment(segment != null ? segment : -1); putToQueue(event); + Hit hit = Hit.builder() + .packageId(userPackageLegalData.getPackageId()) + .subjectId(subjectId) + .signalType(signalType) + .reportFileName("") // TODO + .inqT(message.getInquiryType()) + .busCategory(busCategory) + .inqDt(message.getInquiryDate()) + .inqP(message.getInquiryPurpose()) + .inqA(message.getInquiryAmount()) + .segment(segment != null ? segment : -1) + .recordedAt(LocalDateTime.now()) + .inn(userPackageLegalData.getInn()) + .regNum(userPackageLegalData.getRegNum()) + .build(); + hits.add(hit); } } } diff --git a/src/main/resources/mybatis/mapper/signal/HitsMapper.xml b/src/main/resources/mybatis/mapper/signal/HitsMapper.xml index 479586f..237cc19 100644 --- a/src/main/resources/mybatis/mapper/signal/HitsMapper.xml +++ b/src/main/resources/mybatis/mapper/signal/HitsMapper.xml @@ -6,10 +6,42 @@ - INSERT INTO signals_hits (package_id, subject_id, phone, signal_type, response_sent_at, report_file_name) + INSERT INTO signals_hits ( + package_id, + subject_id, + phone, + signal_type, + response_sent_at, + report_file_name, + inn, + reg_num, + bus_category, + inq_dt, + inq_p, + inq_t, + inq_a, + segment, + recorded_at + ) values - (#{hit.packageId}, #{hit.subjectId}, #{hit.phone}, #{hit.signalType}, #{hit.responseSentAt}, #{hit.reportFileName}) + ( + #{hit.packageId}, + #{hit.subjectId}, + #{hit.phone}, + #{hit.signalType}, + #{hit.responseSentAt}, + #{hit.reportFileName}, + #{hit.inn}, + #{hit.regNum}, + #{hit.busCategory}, + #{hit.inqDt}, + #{hit.inqP}, + #{hit.inqT}, + #{hit.inqA}, + #{hit.segment}, + #{hit.recordedAt} + ) diff --git a/src/main/resources/sql/v2.22.ddl b/src/main/resources/sql/v2.22.ddl index 34344d5..26770ee 100644 --- a/src/main/resources/sql/v2.22.ddl +++ b/src/main/resources/sql/v2.22.ddl @@ -70,7 +70,7 @@ alter table signals_hits alter table signals_hits add inq_t smallint; alter table signals_hits - add inq_a float; + add inq_a bigint; alter table signals_hits add segment smallint; alter table signals_hits