KafkaSignalProcessor refactoring

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,119 +8,84 @@ 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 KafkaSignalProcessorSupport processorSupport;
-    private final SignalKafkaEventPublisher eventPublisher;
 
     public LegalPersonKafkaSignalProcessor(PackageRecordsService packageRecordsService,
-                                           SourceService sourceService,
+                                           KafkaSignalProcessorSupport processorSupport) {
-                                           SignalKafkaEventPublisher eventPublisher) {
         this.packageRecordsService = packageRecordsService;
-        this.sourceService = sourceService;
+        this.processorSupport = processorSupport;
-        this.eventPublisher = eventPublisher;
     }
 
     public List<Hit> process(List<PackageMessage> messages, Map<String, Integer> segmentMap) {
-        List<Hit> hits = new ArrayList<>();
+        return processorSupport.process(
+                messages,
+                segmentMap,
+                packageRecordsService::findCachedLegalsMapByFids,
+                LegalsMap::getFid,
+                this::mappingKey,
+                this::loadPackageData,
+                this::buildEvent,
+                this::buildHit
+        );
+    }
 
-        List<Integer> fids = messages.stream().map(PackageMessage::getFid).collect(Collectors.toList());
+    private Pair<Long, Long> mappingKey(LegalsMap legalsMap) {
-        List<String> memberCodes = messages.stream()
+        return Pair.of(legalsMap.getInn(), legalsMap.getRegNum());
-                .map(message -> message.getMemberCode().substring(0, 6))
+    }
-                .toList();
-
-        List<LegalsMap> legalsMaps = packageRecordsService.findCachedLegalsMapByFids(fids);
-
-        Map<Integer, List<LegalsMap>> legalsDataMap = legalsMaps.stream()
-                .collect(Collectors.groupingBy(LegalsMap::getFid));
 
+    private Map<Pair<Long, Long>, List<UserPackageLegalData>> loadPackageData(List<LegalsMap> legalsMaps) {
         Set<Pair<Long, Long>> uniqueInnRegnumPairs = legalsMaps.stream()
-                .map(l -> Pair.of(l.getInn(), l.getRegNum()))
+                .map(this::mappingKey)
                 .collect(Collectors.toSet());
 
-        // прелоад по memberCode категорий
-        Map<String, String> busCategories = sourceService.findBusCategories(memberCodes);
-
         // subjectId, packageId, signalTypes, inn, regNum только для статуса ON_MONITORING
-        List<UserPackageLegalData> packageLegalData = packageRecordsService
+        List<UserPackageLegalData> packageLegalData = packageRecordsService.selectUserPackageLegalData(uniqueInnRegnumPairs);
-                .selectUserPackageLegalData(uniqueInnRegnumPairs);
 
-        Map<Pair<Long, Long>, List<UserPackageLegalData>> 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);
-
-            String busCategory = busCategories.get(commonMemberCode);
-            if (StringUtils.isBlank(busCategory)) {
-                log.error("Для memberсode = {} отсутствует bus_category или не найден memberсode", commonMemberCode);
-                continue;
     }
 
-            int signalType = CTUtil.checkBusCategory(busCategory);
+    private SignalKafkaEvent buildEvent(PackageMessage message,
-
+                                        LegalsMap legalsMap,
-            List<LegalsMap> legalsMapList = legalsDataMap.get(fid);
+                                        UserPackageLegalData packageData,
-
+                                        int signalType,
-            if (legalsMapList == null) {
+                                        String busCategory,
-                continue;
+                                        Integer segment) {
-            }
-
-            for (LegalsMap legalsMap : legalsMapList) {
-                // флаги просрочек
-                Integer flag90days = legalsMap.getFlag90days();
-                LocalDateTime lastRejectedDate = legalsMap.getLastRejectedDate();
-
-                // в userPackageLegalDataMap только нужные inn, regnum, те что не вошли в userPackageLegalDataList игнорируются
-                List<UserPackageLegalData> 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.setUserPackageLegalData(packageData);
         event.setPackageMessage(message);
-                        event.setSubjectId(subjectId);
+        event.setSubjectId(packageData.getSubjectId());
-                        event.setFlag90days(flag90days);
+        event.setFlag90days(legalsMap.getFlag90days());
         event.setSignalType(Integer.toString(signalType));
         event.setBusCategory(signalType == 3 ? busCategory : null);
-                        event.setLastRejectedDate(lastRejectedDate);
+        event.setLastRejectedDate(legalsMap.getLastRejectedDate());
-
-                        Integer segment = segmentMap.get(commonMemberCode);
         event.setSegment(segment != null ? segment : -1);
-
+        return event;
-                        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())
+    private Hit buildHit(PackageMessage message,
-                                .subjectId(subjectId)
+                         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())
@@ -132,15 +95,8 @@ public class LegalPersonKafkaSignalProcessor {
                 .inqA(message.getInquiryAmount())
                 .segment(segment != null ? segment : -1)
                 .recordedAt(LocalDateTime.now())
-                                .inn(userPackageLegalData.getInn())
+                .inn(packageData.getInn())
-                                .regNum(userPackageLegalData.getRegNum())
+                .regNum(packageData.getRegNum())
                 .build();
-                        hits.add(hit);
-                    }
-                }
-            }
-        }
-
-        return hits;
     }
 }

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,119 +7,81 @@ 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 KafkaSignalProcessorSupport processorSupport;
-    private final SignalKafkaEventPublisher eventPublisher;
 
     public NaturalPersonKafkaSignalProcessor(PackageRecordsService packageRecordsService,
-                                             SourceService sourceService,
+                                             KafkaSignalProcessorSupport processorSupport) {
-                                             SignalKafkaEventPublisher eventPublisher) {
         this.packageRecordsService = packageRecordsService;
-        this.sourceService = sourceService;
+        this.processorSupport = processorSupport;
-        this.eventPublisher = eventPublisher;
     }
 
     public List<Hit> process(List<PackageMessage> messages, Map<String, Integer> segmentMap) {
-        List<Hit> hits = new ArrayList<>();
+        return processorSupport.process(
+                messages,
+                segmentMap,
+                packageRecordsService::findCachedByFids,
+                PhonesMap::getFid,
+                PhonesMap::getPhone,
+                this::loadPackageData,
+                this::buildEvent,
+                this::buildHit
+        );
+    }
 
-        List<Integer> fids = messages.stream().map(PackageMessage::getFid).collect(Collectors.toList());
+    private Map<Long, List<UserPackagePhoneData>> loadPackageData(List<PhonesMap> phonesMaps) {
-        List<String> memberCodes = messages.stream()
+        Set<Long> uniquePhones = phonesMaps.stream()
-                .map(message -> message.getMemberCode().substring(0, 6))
+                .map(PhonesMap::getPhone)
-                .toList();
+                .collect(Collectors.toSet());
-
-        List<PhonesMap> phonesMaps = packageRecordsService.findCachedByFids(fids);
-
-        // key: fid; value: список флагов просрочек (у одного фида может быть несколько номеров)
-        Map<Integer, List<PhonesMap>> phonesDataMap = phonesMaps.stream()
-                .collect(Collectors.groupingBy(PhonesMap::getFid));
-
-        Set<Long> uniquePhones = phonesMaps.stream().map(PhonesMap::getPhone).collect(Collectors.toSet());
-
-        // прелоад по memberCode категорий
-        Map<String, String> busCategories = sourceService.findBusCategories(memberCodes);
 
         // subjectId, packageId, signalTypes, phone только для статуса ON_MONITORING
-        List<UserPackagePhoneData> packagePhoneData = packageRecordsService
+        List<UserPackagePhoneData> packagePhoneData = packageRecordsService.selectUserPackagePhoneData(uniquePhones);
-                .selectUserPackagePhoneData(uniquePhones);
 
         // 1 номер может быть у нескольких пакетов
-        Map<Long, List<UserPackagePhoneData>> 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);
-
-            String busCategory = busCategories.get(commonMemberCode);
-            if (StringUtils.isBlank(busCategory)) {
-                log.error("Для memberсode = {} отсутствует bus_category или не найден memberсode", commonMemberCode);
-                continue;
     }
 
-            int signalType = CTUtil.checkBusCategory(busCategory);
+    private SignalKafkaEvent buildEvent(PackageMessage message,
-
+                                        PhonesMap phonesMap,
-            List<PhonesMap> phonesMapList = phonesDataMap.get(fid);
+                                        UserPackagePhoneData packageData,
-
+                                        int signalType,
-            if (phonesMapList == null) {
+                                        String busCategory,
-                continue;
+                                        Integer segment) {
-            }
-
-            for (PhonesMap phonesMap : phonesMapList) {
-                // флаги просрочек
-                Integer flagDebt5000 = phonesMap.getFlagDebt5000();
-                Integer flag90days = phonesMap.getFlag90days();
-
-                // в userPackagePhoneDataMap только нужные номера, те что не вошли в userPackagePhoneDataList игнорируются
-                List<UserPackagePhoneData> 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.setUserPackagePhoneData(packageData);
         event.setPackageMessage(message);
-                        event.setSubjectId(subjectId);
+        event.setSubjectId(packageData.getSubjectId());
-                        event.setFlagDebt5000(flagDebt5000);
+        event.setFlagDebt5000(phonesMap.getFlagDebt5000());
-                        event.setFlag90days(flag90days);
+        event.setFlag90days(phonesMap.getFlag90days());
         event.setSignalType(Integer.toString(signalType));
         event.setBusCategory(signalType == 3 ? busCategory : null);
-
-                        Integer segment = segmentMap.get(commonMemberCode);
         event.setSegment(segment != null ? segment : -1);
-
+        return event;
-                        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())
+    private Hit buildHit(PackageMessage message,
-                                .subjectId(subjectId)
+                         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())
@@ -131,14 +91,7 @@ public class NaturalPersonKafkaSignalProcessor {
                 .inqA(message.getInquiryAmount())
                 .segment(segment != null ? segment : -1)
                 .recordedAt(LocalDateTime.now())
-                                .phone(userPackagePhoneData.getPhone())
+                .phone(packageData.getPhone())
                 .build();
-                        hits.add(hit);
-                    }
-                }
-            }
-        }
-
-        return hits;
     }
 }