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 index cefd8a6..3985995 100644 --- 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 @@ -1,7 +1,5 @@ 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; @@ -10,137 +8,95 @@ 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; + private final KafkaSignalProcessorSupport processorSupport; public LegalPersonKafkaSignalProcessor(PackageRecordsService packageRecordsService, - SourceService sourceService, - SignalKafkaEventPublisher eventPublisher) { + KafkaSignalProcessorSupport processorSupport) { this.packageRecordsService = packageRecordsService; - this.sourceService = sourceService; - this.eventPublisher = eventPublisher; + this.processorSupport = processorSupport; } public List process(List messages, Map segmentMap) { - List hits = new ArrayList<>(); + return processorSupport.process( + messages, + segmentMap, + packageRecordsService::findCachedLegalsMapByFids, + LegalsMap::getFid, + this::mappingKey, + this::loadPackageData, + this::buildEvent, + this::buildHit + ); + } - 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)); + private Pair mappingKey(LegalsMap legalsMap) { + return Pair.of(legalsMap.getInn(), legalsMap.getRegNum()); + } + private Map, List> loadPackageData(List legalsMaps) { Set> uniqueInnRegnumPairs = legalsMaps.stream() - .map(l -> Pair.of(l.getInn(), l.getRegNum())) + .map(this::mappingKey) .collect(Collectors.toSet()); - // прелоад по memberCode категорий - Map busCategories = sourceService.findBusCategories(memberCodes); - // subjectId, packageId, signalTypes, inn, regNum только для статуса ON_MONITORING - List packageLegalData = packageRecordsService - .selectUserPackageLegalData(uniqueInnRegnumPairs); + List packageLegalData = packageRecordsService.selectUserPackageLegalData(uniqueInnRegnumPairs); - Map, List> userPackageLegalDataMap = packageLegalData.stream() + return 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); + private SignalKafkaEvent buildEvent(PackageMessage message, + LegalsMap legalsMap, + UserPackageLegalData packageData, + int signalType, + String busCategory, + Integer segment) { + SignalKafkaEvent event = new SignalKafkaEvent(); + event.setEventType(EventType.WRITE); + event.setNatural(false); + event.setUserPackageLegalData(packageData); + event.setPackageMessage(message); + event.setSubjectId(packageData.getSubjectId()); + event.setFlag90days(legalsMap.getFlag90days()); + event.setSignalType(Integer.toString(signalType)); + event.setBusCategory(signalType == 3 ? busCategory : null); + event.setLastRejectedDate(legalsMap.getLastRejectedDate()); + event.setSegment(segment != null ? segment : -1); + return event; + } - 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; + private Hit buildHit(PackageMessage message, + UserPackageLegalData packageData, + int signalType, + String busCategory, + Integer segment, + String reportFileName) { + return Hit.builder() + .packageId(packageData.getPackageId()) + .subjectId(packageData.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(packageData.getInn()) + .regNum(packageData.getRegNum()) + .build(); } } 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 index 4285e64..b3cb9e6 100644 --- 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 @@ -1,7 +1,5 @@ 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; @@ -9,136 +7,91 @@ 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; + private final KafkaSignalProcessorSupport processorSupport; public NaturalPersonKafkaSignalProcessor(PackageRecordsService packageRecordsService, - SourceService sourceService, - SignalKafkaEventPublisher eventPublisher) { + KafkaSignalProcessorSupport processorSupport) { this.packageRecordsService = packageRecordsService; - this.sourceService = sourceService; - this.eventPublisher = eventPublisher; + this.processorSupport = processorSupport; } public List process(List messages, Map segmentMap) { - List hits = new ArrayList<>(); + return processorSupport.process( + messages, + segmentMap, + packageRecordsService::findCachedByFids, + PhonesMap::getFid, + PhonesMap::getPhone, + this::loadPackageData, + this::buildEvent, + this::buildHit + ); + } - 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); + private Map> loadPackageData(List phonesMaps) { + Set uniquePhones = phonesMaps.stream() + .map(PhonesMap::getPhone) + .collect(Collectors.toSet()); // subjectId, packageId, signalTypes, phone только для статуса ON_MONITORING - List packagePhoneData = packageRecordsService - .selectUserPackagePhoneData(uniquePhones); + List packagePhoneData = packageRecordsService.selectUserPackagePhoneData(uniquePhones); // 1 номер может быть у нескольких пакетов - Map> userPackagePhoneDataMap = packagePhoneData.stream() + return 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); + private SignalKafkaEvent buildEvent(PackageMessage message, + PhonesMap phonesMap, + UserPackagePhoneData packageData, + int signalType, + String busCategory, + Integer segment) { + SignalKafkaEvent event = new SignalKafkaEvent(); + event.setEventType(EventType.WRITE); + event.setNatural(true); + event.setUserPackagePhoneData(packageData); + event.setPackageMessage(message); + event.setSubjectId(packageData.getSubjectId()); + event.setFlagDebt5000(phonesMap.getFlagDebt5000()); + event.setFlag90days(phonesMap.getFlag90days()); + event.setSignalType(Integer.toString(signalType)); + event.setBusCategory(signalType == 3 ? busCategory : null); + event.setSegment(segment != null ? segment : -1); + return event; + } - 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; + private Hit buildHit(PackageMessage message, + UserPackagePhoneData packageData, + int signalType, + String busCategory, + Integer segment, + String reportFileName) { + return Hit.builder() + .packageId(packageData.getPackageId()) + .subjectId(packageData.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(packageData.getPhone()) + .build(); } }