From e3246bda6420937e429f7ac795f4d5d26ae011a2 Mon Sep 17 00:00:00 2001 From: Mikhail Trofimov Date: Thu, 18 Jun 2026 15:54:15 +0300 Subject: [PATCH] Setup logic for listener --- .../monitoring/KafkaMonitoringService.java | 83 ++++++++++++++++++- .../ru/nbch/credit_tracker/utils/CTUtil.java | 14 ++-- 2 files changed, 85 insertions(+), 12 deletions(-) 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 3473597..3284bb5 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 @@ -1,25 +1,37 @@ package ru.nbch.credit_tracker.service.mode.monitoring; import lombok.extern.slf4j.Slf4j; +import org.apache.commons.lang3.StringUtils; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.support.Acknowledgment; import org.springframework.kafka.support.SendResult; +import ru.nbch.credit_tracker.entities.app.PhonesMap; import ru.nbch.credit_tracker.entities.external_request.kafka.PackageMessage; +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.util.List; +import java.util.Map; +import java.util.Set; import java.util.concurrent.CompletableFuture; +import java.util.stream.Collectors; @Slf4j //@Service public class KafkaMonitoringService { private final KafkaTemplate kafkaTemplate; + private final PackageRecordsService packageRecordsService; + private final SourceService sourceService; -// @Value("${ct.tmp.dir}") - private String tmpDir; - - public KafkaMonitoringService(KafkaTemplate kafkaTemplate) { + public KafkaMonitoringService(KafkaTemplate kafkaTemplate, + PackageRecordsService packageRecordsService, + SourceService sourceService) { this.kafkaTemplate = kafkaTemplate; + this.packageRecordsService = packageRecordsService; + this.sourceService = sourceService; } // @KafkaListener( @@ -35,7 +47,70 @@ public class KafkaMonitoringService { } private void process(List messages) { + 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; + } + + Integer 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(signalType.toString())) { + // TODO put to queue for process + } + } + } + } } private void send(String id, String data) { diff --git a/src/main/java/ru/nbch/credit_tracker/utils/CTUtil.java b/src/main/java/ru/nbch/credit_tracker/utils/CTUtil.java index 74c871c..1f03fa6 100644 --- a/src/main/java/ru/nbch/credit_tracker/utils/CTUtil.java +++ b/src/main/java/ru/nbch/credit_tracker/utils/CTUtil.java @@ -1,5 +1,10 @@ package ru.nbch.credit_tracker.utils; +import lombok.extern.slf4j.Slf4j; +import org.apache.commons.lang3.StringUtils; +import ru.nbch.credit_tracker.entities.app.UserPackage; +import ru.nbch.credit_tracker.entities.indic.Id; + import java.io.File; import java.io.IOException; import java.nio.file.Files; @@ -8,10 +13,6 @@ import java.time.LocalDateTime; import java.time.format.DateTimeFormatter; import java.util.regex.Matcher; import java.util.regex.Pattern; -import lombok.extern.slf4j.Slf4j; -import org.apache.commons.lang3.StringUtils; -import ru.nbch.credit_tracker.entities.app.UserPackage; -import ru.nbch.credit_tracker.entities.indic.Id; @Slf4j public class CTUtil { @@ -41,9 +42,6 @@ public class CTUtil { } public static Integer checkBusCategory(String busCategory) { - if (StringUtils.isBlank(busCategory)) { - return -1; - } switch (busCategory) { case "MFO", "MKK", "MFK" -> { return 1; @@ -52,7 +50,7 @@ public class CTUtil { return 2; } default -> { - return -1; + return 3; } } }