diff --git a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringBatchProcessor.java b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringBatchProcessor.java new file mode 100644 index 0000000..74c38f0 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringBatchProcessor.java @@ -0,0 +1,94 @@ +package ru.nbch.credit_tracker.service.mode.monitoring; + +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Component; +import ru.nbch.credit_tracker.entities.app.Hit; +import ru.nbch.credit_tracker.entities.external_request.kafka.PackageMessage; +import ru.nbch.credit_tracker.service.signal.HitsService; +import ru.nbch.credit_tracker.service.signal.SegmentMapService; +import ru.nbch.credit_tracker.utils.error.ExceptionUtils; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; + +@Slf4j +@Component +public class KafkaMonitoringBatchProcessor { + + private final SegmentMapService segmentMapService; + private final HitsService hitsService; + private final KafkaMonitoringMessageFilter messageFilter; + private final NaturalPersonKafkaSignalProcessor naturalPersonSignalProcessor; + private final LegalPersonKafkaSignalProcessor legalPersonSignalProcessor; + + public KafkaMonitoringBatchProcessor(SegmentMapService segmentMapService, + HitsService hitsService, + KafkaMonitoringMessageFilter messageFilter, + NaturalPersonKafkaSignalProcessor naturalPersonSignalProcessor, + LegalPersonKafkaSignalProcessor legalPersonSignalProcessor) { + this.segmentMapService = segmentMapService; + this.hitsService = hitsService; + this.messageFilter = messageFilter; + this.naturalPersonSignalProcessor = naturalPersonSignalProcessor; + this.legalPersonSignalProcessor = legalPersonSignalProcessor; + } + + public void process(List messages, + ExecutorService processingExecutor, + ExecutorService dbExecutor) { + Map> partition = messages.stream() + .filter(Objects::nonNull) + .filter(messageFilter::isSupported) + .collect(Collectors.partitioningBy(messageFilter::isNaturalPerson)); + + List npMessages = partition.get(true); + List lpMessages = partition.get(false); + + Map segmentMap = segmentMapService.loadSegmentMap(); + + CompletableFuture> npFuture = CompletableFuture.supplyAsync(() -> { + if (npMessages != null && !npMessages.isEmpty()) { + return naturalPersonSignalProcessor.process(npMessages, segmentMap); + } + return Collections.emptyList(); + }, processingExecutor); + + CompletableFuture> lpFuture = CompletableFuture.supplyAsync(() -> { + if (lpMessages != null && !lpMessages.isEmpty()) { + return legalPersonSignalProcessor.process(lpMessages, segmentMap); + } + return Collections.emptyList(); + }, processingExecutor); + + CompletableFuture> combinedFuture = npFuture + .thenCombine(lpFuture, (npHits, lpHits) -> { + List allHits = new ArrayList<>(); + if (npHits != null) allHits.addAll(npHits); + if (lpHits != null) allHits.addAll(lpHits); + return allHits; + }); + + combinedFuture.thenAcceptAsync(hits -> { + if (!hits.isEmpty()) { + try { + hitsService.insertHits(hits); + } catch (Exception e) { + log.error("Failed to save hits. {}", ExceptionUtils.getStackTrace(e)); + } + } + }, dbExecutor); + + try { + combinedFuture.get(1, TimeUnit.MINUTES); + } catch (Exception e) { + log.error("Error processing messages", e); + } + } +} diff --git a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringMessageFilter.java b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringMessageFilter.java new file mode 100644 index 0000000..ca48768 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringMessageFilter.java @@ -0,0 +1,19 @@ +package ru.nbch.credit_tracker.service.mode.monitoring; + +import org.springframework.stereotype.Component; +import ru.nbch.credit_tracker.entities.external_request.kafka.PackageMessage; + +@Component +public class KafkaMonitoringMessageFilter { + + public boolean isSupported(PackageMessage message) { + Integer inquiryType = message.getInquiryType(); + String billingCode = message.getBillingCode(); + return inquiryType == 1 || inquiryType == 2 || + (inquiryType == 3 && billingCode.equalsIgnoreCase("CHSP")); + } + + public boolean isNaturalPerson(PackageMessage message) { + return message.getPersons() != null && !message.getPersons().isEmpty(); + } +} 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 9a4a1a7..add25d9 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 @@ -2,8 +2,6 @@ package ru.nbch.credit_tracker.service.mode.monitoring; import jakarta.annotation.PreDestroy; import lombok.extern.slf4j.Slf4j; -import org.apache.commons.lang3.StringUtils; -import org.apache.commons.lang3.tuple.Pair; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.annotation.DependsOn; import org.springframework.kafka.annotation.KafkaListener; @@ -11,25 +9,12 @@ import org.springframework.kafka.support.Acknowledgment; import org.springframework.scheduling.concurrent.CustomizableThreadFactory; import org.springframework.stereotype.Service; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; -import ru.nbch.credit_tracker.entities.app.Hit; -import ru.nbch.credit_tracker.entities.app.LegalsMap; -import ru.nbch.credit_tracker.entities.app.PhonesMap; import ru.nbch.credit_tracker.entities.external_request.kafka.PackageMessage; import ru.nbch.credit_tracker.enums.EventType; 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; -import ru.nbch.credit_tracker.utils.error.ExceptionUtils; -import java.time.LocalDateTime; -import java.util.*; import java.util.concurrent.*; -import java.util.stream.Collectors; +import java.util.List; @Slf4j @Service @@ -45,26 +30,17 @@ public class KafkaMonitoringService { private final BlockingQueue signalEventsQueue; - private final PackageRecordsService packageRecordsService; - private final SourceService sourceService; - private final SegmentMapService segmentMapService; - private final HitsService hitsService; + private final KafkaMonitoringBatchProcessor batchProcessor; private final ReportNameRegistry reportNameRegistry; public KafkaMonitoringService(@Qualifier("signalKafkaEventQueue") BlockingQueue signalEventsQueue, - PackageRecordsService packageRecordsService, - SourceService sourceService, - SegmentMapService segmentMapService, CreditTrackerSettings settings, - HitsService hitsService, + KafkaMonitoringBatchProcessor batchProcessor, ReportNameRegistry reportNameRegistry) { this.signalEventsQueue = signalEventsQueue; - this.packageRecordsService = packageRecordsService; - this.sourceService = sourceService; - this.segmentMapService = segmentMapService; + this.batchProcessor = batchProcessor; this.reportNameRegistry = reportNameRegistry; this.permits = Integer.parseInt(settings.getKafkaConsumerSettings().concurency()); // количество равно concurency листенера - this.hitsService = hitsService; this.processingPermits = new Semaphore(permits); this.processingExecutor = Executors.newFixedThreadPool( @@ -96,329 +72,13 @@ public class KafkaMonitoringService { try { acknowledgment.acknowledge(); long startTime = System.currentTimeMillis(); - process(records); + batchProcessor.process(records, processingExecutor, dbExecutor); log.info("Processed {} messages from Kafka. Time {} ms", records.size(), System.currentTimeMillis() - startTime); } finally { processingPermits.release(); } } - private void process(List messages) { - Map> partition = messages.stream() - .filter(Objects::nonNull) - .filter(this::filterMessages) - .collect(Collectors.partitioningBy(m -> m.getPersons() != null && !m.getPersons().isEmpty())); - - List npMessages = partition.get(true); - List lpMessages = partition.get(false); - - Map segmentMap = segmentMapService.loadSegmentMap(); - - CompletableFuture> npFuture = CompletableFuture.supplyAsync(() -> { - if (npMessages != null && !npMessages.isEmpty()) { - return processNpMessages(npMessages, segmentMap); - } - return Collections.emptyList(); - }, processingExecutor); - - CompletableFuture> lpFuture = CompletableFuture.supplyAsync(() -> { - if (lpMessages != null && !lpMessages.isEmpty()) { - return processLpMessages(lpMessages, segmentMap); - } - return Collections.emptyList(); - }, processingExecutor); - - CompletableFuture> combinedFuture = npFuture - .thenCombine(lpFuture, (npHits, lpHits) -> { - List allHits = new ArrayList<>(); - if (npHits != null) allHits.addAll(npHits); - if (lpHits != null) allHits.addAll(lpHits); - return allHits; - }); - - combinedFuture.thenAcceptAsync(hits -> { - if (!hits.isEmpty()) { - try { - hitsService.insertHits(hits); - } catch (Exception e) { - log.error("Failed to save hits. {}", ExceptionUtils.getStackTrace(e)); - } - } - }, dbExecutor); - - try { - combinedFuture.get(1, TimeUnit.MINUTES); - } catch (Exception e) { - log.error("Error processing messages", e); - } - } - - private List processNpMessages(List messages, Map segmentMap) { - List hits = new ArrayList<>(); - - List fids = messages.stream().map(PackageMessage::getFid).collect(Collectors.toList()); - List memberCodes = messages.stream() - .map(message -> message.getMemberCode().substring(0, 6)) - .toList(); - - List phonesMaps = packageRecordsService.findCachedByFids(fids); - - // key: fid; value: список флагов просрочек (у одного фида может быть несколько номеров) - Map> phonesDataMap = phonesMaps.stream() - .collect(Collectors.groupingBy(PhonesMap::getFid)); - - Set uniquePhones = phonesMaps.stream().map(PhonesMap::getPhone).collect(Collectors.toSet()); - - // прелоад по memberCode категорий - Map busCategories = sourceService.findBusCategories(memberCodes); - - // subjectId, packageId, signalTypes, phone только для статуса ON_MONITORING - List packagePhoneData = packageRecordsService - .selectUserPackagePhoneData(uniquePhones); - - // 1 номер может быть у нескольких пакетов - Map> userPackagePhoneDataMap = packagePhoneData.stream() - .collect(Collectors.groupingBy(UserPackagePhoneData::getPhone)); - - for (PackageMessage message : messages) { - Integer fid = message.getFid(); - String memberCode = message.getMemberCode(); - - String busCategory = busCategories.get(memberCode.substring(0, 6)); - if (StringUtils.isBlank(busCategory)) { - log.error("Для memberсode = {} отсутствует bus_category или не найден memberсode", memberCode.substring(0, 6)); - continue; - } - - int signalType = CTUtil.checkBusCategory(busCategory); - - List phonesMapList = phonesDataMap.get(fid); - - if (phonesMapList == null) { - continue; - } - - for (PhonesMap phonesMap : phonesMapList) { - // в userPackagePhoneDataMap только нужные номера, те что не вошли в userPackagePhoneDataList игнорируются - List userPackagePhoneDataList = userPackagePhoneDataMap.get(phonesMap.getPhone()); - - if (userPackagePhoneDataList == null) { - continue; - } - - for (UserPackagePhoneData userPackagePhoneData : userPackagePhoneDataList) { - String signalTypes = userPackagePhoneData.getSignalTypes(); - - if (signalTypes.contains(Integer.toString(signalType))) { - Integer segment = segmentMap.get(memberCode.substring(0, 6)); - SignalKafkaEvent event = buildSignalKafkaEvent( - true, message, - phonesMap, userPackagePhoneData, - null, null, // legal person fields - signalType, busCategory, segment - ); - - String reportKey = userPackagePhoneData.getUserId().substring(0, 6) - + "_" + userPackagePhoneData.getPackageName(); - String reportFileName = reportNameRegistry.resolveNameAndEnqueue(reportKey, event); - if (reportFileName == null) { - continue; // событие не встало в очередь — hit не пишем - } - Hit hit = buildHit( - true, message, - userPackagePhoneData, - null, - signalType, busCategory, - segment, reportFileName - ); - hits.add(hit); - } - } - } - } - - return hits; - } - - private List processLpMessages(List messages, Map segmentMap) { - List hits = new ArrayList<>(); - - List fids = messages.stream().map(PackageMessage::getFid).collect(Collectors.toList()); - List memberCodes = messages.stream() - .map(message -> message.getMemberCode().substring(0, 6)) - .toList(); - - List legalsMaps = packageRecordsService.findCachedLegalsMapByFids(fids); - - Map> legalsDataMap = legalsMaps.stream() - .collect(Collectors.groupingBy(LegalsMap::getFid)); - - Set> uniqueInnRegnumPairs = legalsMaps.stream() - .map(l -> Pair.of(l.getInn(), l.getRegNum())) - .collect(Collectors.toSet()); - - // прелоад по memberCode категорий - Map busCategories = sourceService.findBusCategories(memberCodes); - - // subjectId, packageId, signalTypes, inn, regNum только для статуса ON_MONITORING - List packageLegalData = packageRecordsService - .selectUserPackageLegalData(uniqueInnRegnumPairs); - - Map, List> userPackageLegalDataMap = packageLegalData.stream() - .collect(Collectors.groupingBy(p -> Pair.of(p.getInn(), p.getRegNum()))); - - for (PackageMessage message : messages) { - Integer fid = message.getFid(); - String memberCode = message.getMemberCode(); - - String busCategory = busCategories.get(memberCode.substring(0, 6)); - if (StringUtils.isBlank(busCategory)) { - log.error("Для memberсode = {} отсутствует bus_category или не найден memberсode", memberCode.substring(0, 6)); - continue; - } - - int signalType = CTUtil.checkBusCategory(busCategory); - - List legalsMapList = legalsDataMap.get(fid); - - if (legalsMapList == null) { - continue; - } - - for (LegalsMap legalsMap : legalsMapList) { - // в userPackageLegalDataMap только нужные inn, regnum, те что не вошли в userPackageLegalDataList игнорируются - List userPackageLegalDataList = userPackageLegalDataMap.get(Pair.of(legalsMap.getInn(), legalsMap.getRegNum())); - - if (userPackageLegalDataList == null) { - continue; - } - - for (UserPackageLegalData userPackageLegalData : userPackageLegalDataList) { - String signalTypes = userPackageLegalData.getSignalTypes(); - - if (signalTypes.contains(Integer.toString(signalType))) { - Integer segment = segmentMap.get(memberCode.substring(0, 6)); - - SignalKafkaEvent event = buildSignalKafkaEvent( - false, message, - null, null, // np fields - legalsMap, - userPackageLegalData, - signalType, busCategory, segment - ); - - String reportKey = userPackageLegalData.getUserId().substring(0, 6) - + "_" + userPackageLegalData.getPackageName(); - String reportFileName = reportNameRegistry.resolveNameAndEnqueue(reportKey, event); - if (reportFileName == null) { - continue; // событие не встало в очередь — hit не пишем - } - Hit hit = buildHit( - false, message, - null, - userPackageLegalData, - signalType, busCategory, - segment, reportFileName - ); - hits.add(hit); - } - } - } - } - - return hits; - } - - private boolean filterMessages(PackageMessage message) { - Integer inquiryType = message.getInquiryType(); - String billingCode = message.getBillingCode(); - return inquiryType == 1 || inquiryType == 2 || - (inquiryType == 3 && billingCode.equalsIgnoreCase("CHSP")); - } - - private SignalKafkaEvent buildSignalKafkaEvent(boolean isNatural, PackageMessage message, - PhonesMap phonesMap, UserPackagePhoneData userPackagePhoneData, - LegalsMap legalsMap, UserPackageLegalData userPackageLegalData, - int signalType, String busCategory, Integer segment) { - if (isNatural) { - // флаги просрочек - Integer flagDebt5000 = phonesMap.getFlagDebt5000(); - Integer flag90days = phonesMap.getFlag90days(); - - String subjectId = userPackagePhoneData.getSubjectId(); - - return SignalKafkaEvent.builder() - .eventType(EventType.WRITE) - .natural(true) - .userPackagePhoneData(userPackagePhoneData) - .packageMessage(message) - .subjectId(subjectId) - .flagDebt5000(flagDebt5000) - .flag90days(flag90days) - .signalType(Integer.toString(signalType)) - .busCategory(signalType == 3 ? busCategory : null) - .segment(segment != null ? segment : -1) - .build(); - } else { - // флаги просрочек - Integer flag90days = legalsMap.getFlag90days(); - LocalDateTime lastRejectedDate = legalsMap.getLastRejectedDate(); - - String subjectId = userPackageLegalData.getSubjectId(); - return SignalKafkaEvent.builder() - .eventType(EventType.WRITE) - .natural(false) - .userPackageLegalData(userPackageLegalData) - .packageMessage(message) - .subjectId(subjectId) - .flag90days(flag90days) - .signalType(Integer.toString(signalType)) - .busCategory(signalType == 3 ? busCategory : null) - .lastRejectedDate(lastRejectedDate) - .segment(segment != null ? segment : -1) - .build(); - } - } - - private Hit buildHit(boolean isNatural, PackageMessage message, - UserPackagePhoneData userPackagePhoneData, - UserPackageLegalData userPackageLegalData, - int signalType, String busCategory, - Integer segment, String reportFileName) { - if (isNatural) { - return Hit.builder() - .packageId(userPackagePhoneData.getPackageId()) - .subjectId(userPackagePhoneData.getSubjectId()) - .signalType(signalType) - .reportFileName(reportFileName.replace(".xml", ".zip")) - .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(); - } else { - return Hit.builder() - .packageId(userPackageLegalData.getPackageId()) - .subjectId(userPackageLegalData.getSubjectId()) - .signalType(signalType) - .reportFileName(reportFileName.replace(".xml", ".zip")) - .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(); - } - } - @PreDestroy public void stopKafkaMonitoringService() throws InterruptedException { log.info("Stopping KafkaMonitoringService..."); diff --git a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/LegalPersonKafkaSignalProcessor.java b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/LegalPersonKafkaSignalProcessor.java new file mode 100644 index 0000000..cefd8a6 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/LegalPersonKafkaSignalProcessor.java @@ -0,0 +1,146 @@ +package ru.nbch.credit_tracker.service.mode.monitoring; + +import lombok.extern.slf4j.Slf4j; +import org.apache.commons.lang3.StringUtils; +import org.apache.commons.lang3.tuple.Pair; +import org.springframework.stereotype.Component; +import ru.nbch.credit_tracker.entities.app.Hit; +import ru.nbch.credit_tracker.entities.app.LegalsMap; +import ru.nbch.credit_tracker.entities.external_request.kafka.PackageMessage; +import ru.nbch.credit_tracker.enums.EventType; +import ru.nbch.credit_tracker.model.SignalKafkaEvent; +import ru.nbch.credit_tracker.model.UserPackageLegalData; +import ru.nbch.credit_tracker.service.nbch.SourceService; +import ru.nbch.credit_tracker.service.signal.PackageRecordsService; +import ru.nbch.credit_tracker.utils.CTUtil; + +import java.time.LocalDateTime; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.stream.Collectors; + +@Slf4j +@Component +public class LegalPersonKafkaSignalProcessor { + + private final PackageRecordsService packageRecordsService; + private final SourceService sourceService; + private final SignalKafkaEventPublisher eventPublisher; + + public LegalPersonKafkaSignalProcessor(PackageRecordsService packageRecordsService, + SourceService sourceService, + SignalKafkaEventPublisher eventPublisher) { + this.packageRecordsService = packageRecordsService; + this.sourceService = sourceService; + this.eventPublisher = eventPublisher; + } + + public List process(List messages, Map segmentMap) { + List hits = new ArrayList<>(); + + List fids = messages.stream().map(PackageMessage::getFid).collect(Collectors.toList()); + List memberCodes = messages.stream() + .map(message -> message.getMemberCode().substring(0, 6)) + .toList(); + + List legalsMaps = packageRecordsService.findCachedLegalsMapByFids(fids); + + Map> legalsDataMap = legalsMaps.stream() + .collect(Collectors.groupingBy(LegalsMap::getFid)); + + Set> uniqueInnRegnumPairs = legalsMaps.stream() + .map(l -> Pair.of(l.getInn(), l.getRegNum())) + .collect(Collectors.toSet()); + + // прелоад по memberCode категорий + Map busCategories = sourceService.findBusCategories(memberCodes); + + // subjectId, packageId, signalTypes, inn, regNum только для статуса ON_MONITORING + List packageLegalData = packageRecordsService + .selectUserPackageLegalData(uniqueInnRegnumPairs); + + Map, List> userPackageLegalDataMap = packageLegalData.stream() + .collect(Collectors.groupingBy(p -> Pair.of(p.getInn(), p.getRegNum()))); + + for (PackageMessage message : messages) { + Integer fid = message.getFid(); + String memberCode = message.getMemberCode(); + String commonMemberCode = memberCode.substring(0, 6); + + String busCategory = busCategories.get(commonMemberCode); + if (StringUtils.isBlank(busCategory)) { + log.error("Для memberсode = {} отсутствует bus_category или не найден memberсode", commonMemberCode); + continue; + } + + int signalType = CTUtil.checkBusCategory(busCategory); + + List legalsMapList = legalsDataMap.get(fid); + + if (legalsMapList == null) { + continue; + } + + for (LegalsMap legalsMap : legalsMapList) { + // флаги просрочек + Integer flag90days = legalsMap.getFlag90days(); + LocalDateTime lastRejectedDate = legalsMap.getLastRejectedDate(); + + // в userPackageLegalDataMap только нужные inn, regnum, те что не вошли в userPackageLegalDataList игнорируются + List userPackageLegalDataList = userPackageLegalDataMap.get(Pair.of(legalsMap.getInn(), legalsMap.getRegNum())); + + if (userPackageLegalDataList == null) { + continue; + } + + 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(userPackageLegalData); + event.setPackageMessage(message); + event.setSubjectId(subjectId); + event.setFlag90days(flag90days); + event.setSignalType(Integer.toString(signalType)); + event.setBusCategory(signalType == 3 ? busCategory : null); + event.setLastRejectedDate(lastRejectedDate); + + Integer segment = segmentMap.get(commonMemberCode); + event.setSegment(segment != null ? segment : -1); + + String reportKey = userPackageLegalData.getUserId().substring(0, 6) + + "_" + userPackageLegalData.getPackageName(); + String reportFileName = eventPublisher.resolveNameAndEnqueue(reportKey, event); + if (reportFileName == null) { + continue; // событие не встало в очередь — hit не пишем + } + Hit hit = Hit.builder() + .packageId(userPackageLegalData.getPackageId()) + .subjectId(subjectId) + .signalType(signalType) + .reportFileName(reportFileName.replace(".xml", ".zip")) + .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); + } + } + } + } + + return hits; + } +} diff --git a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/NaturalPersonKafkaSignalProcessor.java b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/NaturalPersonKafkaSignalProcessor.java new file mode 100644 index 0000000..4285e64 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/NaturalPersonKafkaSignalProcessor.java @@ -0,0 +1,144 @@ +package ru.nbch.credit_tracker.service.mode.monitoring; + +import lombok.extern.slf4j.Slf4j; +import org.apache.commons.lang3.StringUtils; +import org.springframework.stereotype.Component; +import ru.nbch.credit_tracker.entities.app.Hit; +import ru.nbch.credit_tracker.entities.app.PhonesMap; +import ru.nbch.credit_tracker.entities.external_request.kafka.PackageMessage; +import ru.nbch.credit_tracker.enums.EventType; +import ru.nbch.credit_tracker.model.SignalKafkaEvent; +import ru.nbch.credit_tracker.model.UserPackagePhoneData; +import ru.nbch.credit_tracker.service.nbch.SourceService; +import ru.nbch.credit_tracker.service.signal.PackageRecordsService; +import ru.nbch.credit_tracker.utils.CTUtil; + +import java.time.LocalDateTime; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.stream.Collectors; + +@Slf4j +@Component +public class NaturalPersonKafkaSignalProcessor { + + private final PackageRecordsService packageRecordsService; + private final SourceService sourceService; + private final SignalKafkaEventPublisher eventPublisher; + + public NaturalPersonKafkaSignalProcessor(PackageRecordsService packageRecordsService, + SourceService sourceService, + SignalKafkaEventPublisher eventPublisher) { + this.packageRecordsService = packageRecordsService; + this.sourceService = sourceService; + this.eventPublisher = eventPublisher; + } + + public List process(List messages, Map segmentMap) { + List hits = new ArrayList<>(); + + List fids = messages.stream().map(PackageMessage::getFid).collect(Collectors.toList()); + List memberCodes = messages.stream() + .map(message -> message.getMemberCode().substring(0, 6)) + .toList(); + + List phonesMaps = packageRecordsService.findCachedByFids(fids); + + // key: fid; value: список флагов просрочек (у одного фида может быть несколько номеров) + Map> phonesDataMap = phonesMaps.stream() + .collect(Collectors.groupingBy(PhonesMap::getFid)); + + Set uniquePhones = phonesMaps.stream().map(PhonesMap::getPhone).collect(Collectors.toSet()); + + // прелоад по memberCode категорий + Map busCategories = sourceService.findBusCategories(memberCodes); + + // subjectId, packageId, signalTypes, phone только для статуса ON_MONITORING + List packagePhoneData = packageRecordsService + .selectUserPackagePhoneData(uniquePhones); + + // 1 номер может быть у нескольких пакетов + Map> userPackagePhoneDataMap = packagePhoneData.stream() + .collect(Collectors.groupingBy(UserPackagePhoneData::getPhone)); + + for (PackageMessage message : messages) { + Integer fid = message.getFid(); + String memberCode = message.getMemberCode(); + String commonMemberCode = memberCode.substring(0, 6); + + String busCategory = busCategories.get(commonMemberCode); + if (StringUtils.isBlank(busCategory)) { + log.error("Для memberсode = {} отсутствует bus_category или не найден memberсode", commonMemberCode); + continue; + } + + int signalType = CTUtil.checkBusCategory(busCategory); + + List phonesMapList = phonesDataMap.get(fid); + + if (phonesMapList == null) { + continue; + } + + for (PhonesMap phonesMap : phonesMapList) { + // флаги просрочек + Integer flagDebt5000 = phonesMap.getFlagDebt5000(); + Integer flag90days = phonesMap.getFlag90days(); + + // в userPackagePhoneDataMap только нужные номера, те что не вошли в userPackagePhoneDataList игнорируются + List userPackagePhoneDataList = userPackagePhoneDataMap.get(phonesMap.getPhone()); + + if (userPackagePhoneDataList == null) { + continue; + } + + for (UserPackagePhoneData userPackagePhoneData : userPackagePhoneDataList) { + String subjectId = userPackagePhoneData.getSubjectId(); + String signalTypes = userPackagePhoneData.getSignalTypes(); + + if (signalTypes.contains(Integer.toString(signalType))) { + SignalKafkaEvent event = new SignalKafkaEvent(); + event.setEventType(EventType.WRITE); + event.setNatural(true); + event.setUserPackagePhoneData(userPackagePhoneData); + event.setPackageMessage(message); + event.setSubjectId(subjectId); + event.setFlagDebt5000(flagDebt5000); + event.setFlag90days(flag90days); + event.setSignalType(Integer.toString(signalType)); + event.setBusCategory(signalType == 3 ? busCategory : null); + + Integer segment = segmentMap.get(commonMemberCode); + event.setSegment(segment != null ? segment : -1); + + String reportKey = userPackagePhoneData.getUserId().substring(0, 6) + + "_" + userPackagePhoneData.getPackageName(); + String reportFileName = eventPublisher.resolveNameAndEnqueue(reportKey, event); + if (reportFileName == null) { + continue; // событие не встало в очередь — hit не пишем + } + Hit hit = Hit.builder() + .packageId(userPackagePhoneData.getPackageId()) + .subjectId(subjectId) + .signalType(signalType) + .reportFileName(reportFileName.replace(".xml", ".zip")) + .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); + } + } + } + } + + return hits; + } +} diff --git a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/SignalKafkaEventPublisher.java b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/SignalKafkaEventPublisher.java new file mode 100644 index 0000000..0138d6d --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/SignalKafkaEventPublisher.java @@ -0,0 +1,23 @@ +package ru.nbch.credit_tracker.service.mode.monitoring; + +import org.springframework.stereotype.Component; +import ru.nbch.credit_tracker.model.SignalKafkaEvent; + +@Component +public class SignalKafkaEventPublisher { + + private final ReportNameRegistry reportNameRegistry; + + public SignalKafkaEventPublisher(ReportNameRegistry reportNameRegistry) { + this.reportNameRegistry = reportNameRegistry; + } + + /** + * Резолвит имя результирующего файла и ставит событие в очередь. + * + * @return имя xml-файла или null, если поток был прерван и событие не встало в очередь + */ + public String resolveNameAndEnqueue(String reportKey, SignalKafkaEvent event) { + return reportNameRegistry.resolveNameAndEnqueue(reportKey, event); + } +}