From a7a59c0c1014763435a60d7faa11591f1d1841ec Mon Sep 17 00:00:00 2001 From: Mikhail Trofimov Date: Thu, 25 Jun 2026 14:48:12 +0300 Subject: [PATCH] Added processing for legal persons --- .../entities/app/LegalsMap.java | 22 ++++ .../mapper/signal/LegalsMapMapper.java | 14 ++ .../mapper/signal/PackagesRecordsMapper.java | 3 + .../model/SignalKafkaEvent.java | 3 + .../model/UserPackageLegalData.java | 18 +++ .../monitoring/KafkaMonitoringService.java | 123 +++++++++++++++++- .../service/signal/PackageRecordsService.java | 34 ++++- .../mybatis/mapper/signal/LegalsMapMapper.xml | 36 +++++ .../mapper/signal/PackageRecordsMapper.xml | 15 +++ 9 files changed, 263 insertions(+), 5 deletions(-) create mode 100644 src/main/java/ru/nbch/credit_tracker/entities/app/LegalsMap.java create mode 100644 src/main/java/ru/nbch/credit_tracker/mapper/signal/LegalsMapMapper.java create mode 100644 src/main/java/ru/nbch/credit_tracker/model/UserPackageLegalData.java create mode 100644 src/main/resources/mybatis/mapper/signal/LegalsMapMapper.xml diff --git a/src/main/java/ru/nbch/credit_tracker/entities/app/LegalsMap.java b/src/main/java/ru/nbch/credit_tracker/entities/app/LegalsMap.java new file mode 100644 index 0000000..9f14760 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/entities/app/LegalsMap.java @@ -0,0 +1,22 @@ +package ru.nbch.credit_tracker.entities.app; + +import lombok.*; + +import java.time.LocalDateTime; + +@Getter +@Setter +@AllArgsConstructor +@NoArgsConstructor +@Builder +public class LegalsMap { + private Long id; + private Long inn; + private Long regNUm; + private Integer fid; + private LocalDateTime searchedAt; + private Integer flag90days; + private LocalDateTime calculatedAt; + private LocalDateTime lastRejectedDate; +} + diff --git a/src/main/java/ru/nbch/credit_tracker/mapper/signal/LegalsMapMapper.java b/src/main/java/ru/nbch/credit_tracker/mapper/signal/LegalsMapMapper.java new file mode 100644 index 0000000..7c31c1b --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/mapper/signal/LegalsMapMapper.java @@ -0,0 +1,14 @@ +package ru.nbch.credit_tracker.mapper.signal; + +import org.apache.ibatis.annotations.Mapper; +import org.apache.ibatis.annotations.Param; +import ru.nbch.credit_tracker.entities.app.LegalsMap; + +import java.util.List; + +@Mapper +public interface LegalsMapMapper { + + List findByFids(@Param("fids") List fids); + +} diff --git a/src/main/java/ru/nbch/credit_tracker/mapper/signal/PackagesRecordsMapper.java b/src/main/java/ru/nbch/credit_tracker/mapper/signal/PackagesRecordsMapper.java index 8c52eed..d8cae7e 100644 --- a/src/main/java/ru/nbch/credit_tracker/mapper/signal/PackagesRecordsMapper.java +++ b/src/main/java/ru/nbch/credit_tracker/mapper/signal/PackagesRecordsMapper.java @@ -3,6 +3,7 @@ package ru.nbch.credit_tracker.mapper.signal; import org.apache.ibatis.annotations.Mapper; import org.apache.ibatis.annotations.Param; import ru.nbch.credit_tracker.entities.app.PackagesRecordsMap; +import ru.nbch.credit_tracker.model.UserPackageLegalData; import ru.nbch.credit_tracker.model.UserPackagePhoneData; import java.util.List; @@ -21,6 +22,8 @@ public interface PackagesRecordsMapper { List selectUserPackagePhoneData(@Param("json") String json); + List selectUserPackageLegalData(@Param("json") String json); + Long deleteByPackageId(@Param("packageId") Long id); PackagesRecordsMap findById(Long id); diff --git a/src/main/java/ru/nbch/credit_tracker/model/SignalKafkaEvent.java b/src/main/java/ru/nbch/credit_tracker/model/SignalKafkaEvent.java index b673693..06d5ef6 100644 --- a/src/main/java/ru/nbch/credit_tracker/model/SignalKafkaEvent.java +++ b/src/main/java/ru/nbch/credit_tracker/model/SignalKafkaEvent.java @@ -13,6 +13,7 @@ public class SignalKafkaEvent { private EventType eventType; private UserPackagePhoneData userPackagePhoneData; + private UserPackageLegalData userPackageLegalData; private PackageMessage packageMessage; private String subjectId; @@ -21,4 +22,6 @@ public class SignalKafkaEvent { private String signalType; private String busCategory; private Integer segment; + + private boolean natural; } diff --git a/src/main/java/ru/nbch/credit_tracker/model/UserPackageLegalData.java b/src/main/java/ru/nbch/credit_tracker/model/UserPackageLegalData.java new file mode 100644 index 0000000..8452b62 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/model/UserPackageLegalData.java @@ -0,0 +1,18 @@ +package ru.nbch.credit_tracker.model; + +import lombok.*; + +@Getter +@Setter +@Builder +@NoArgsConstructor +@AllArgsConstructor +public class UserPackageLegalData { + private Long packageId; + private String signalTypes; + private String subjectId; + private String packageName; + + private Long inn; + private Long regNum; +} 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 36ccbf6..df42d9b 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 @@ -3,20 +3,25 @@ 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.kafka.annotation.KafkaListener; import org.springframework.kafka.support.Acknowledgment; import org.springframework.stereotype.Service; +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.PackageRecordsService; import ru.nbch.credit_tracker.service.signal.SegmentMapService; import ru.nbch.credit_tracker.utils.CTUtil; +import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.Set; @@ -74,7 +79,32 @@ public class KafkaMonitoringService { } private void process(List messages) { + List hits = new ArrayList<>(); + + Map> partition = messages.stream() + .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(); + + if (npMessages != null && !npMessages.isEmpty()) { + List npHits = processNpMessages(npMessages, segmentMap); + hits.addAll(npHits); + } + + if (lpMessages != null && !lpMessages.isEmpty()) { + List lpHits = processLpMessages(lpMessages, segmentMap); + hits.addAll(lpHits); + } + // TODO insert hits + } + + 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)) @@ -136,6 +166,7 @@ public class KafkaMonitoringService { 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); @@ -153,7 +184,97 @@ public class KafkaMonitoringService { } } - // TODO insert hits + 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) { + // флаги просрочек + Integer flag90days = legalsMap.getFlag90days(); + + // в userPackageLegalDataMap только нужные inn, regnum, те что не вошли в userPackageLegalDataList игнорируются + List userPackageLegalDataList = userPackageLegalDataMap.get(Pair.of(legalsMap.getInn(), legalsMap.getRegNUm())); + + if (userPackageLegalDataList == null) { + continue; + } + + for (UserPackageLegalData userPackagePhoneData : userPackageLegalDataList) { + String subjectId = userPackagePhoneData.getSubjectId(); + String signalTypes = userPackagePhoneData.getSignalTypes(); + + if (signalTypes.contains(Integer.toString(signalType))) { + SignalKafkaEvent event = new SignalKafkaEvent(); + event.setEventType(EventType.WRITE); + event.setNatural(false); + event.setUserPackageLegalData(userPackagePhoneData); + event.setPackageMessage(message); + event.setSubjectId(subjectId); + event.setFlag90days(flag90days); + event.setSignalType(Integer.toString(signalType)); + event.setBusCategory(signalType == 3 ? busCategory : null); + + Integer segment = segmentMap.get(memberCode.substring(0, 6)); + event.setSegment(segment != null ? segment : -1); + + putToQueue(event); + } + } + } + } + + 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 void putToQueue(SignalKafkaEvent event) { diff --git a/src/main/java/ru/nbch/credit_tracker/service/signal/PackageRecordsService.java b/src/main/java/ru/nbch/credit_tracker/service/signal/PackageRecordsService.java index a480c2d..f2baefc 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/signal/PackageRecordsService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/signal/PackageRecordsService.java @@ -3,15 +3,15 @@ package ru.nbch.credit_tracker.service.signal; import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; import lombok.extern.slf4j.Slf4j; +import org.apache.commons.lang3.tuple.Pair; import org.springframework.stereotype.Service; +import ru.nbch.credit_tracker.entities.app.LegalsMap; import ru.nbch.credit_tracker.entities.app.PackagesRecordsMap; import ru.nbch.credit_tracker.entities.app.PhonesMap; +import ru.nbch.credit_tracker.mapper.signal.LegalsMapMapper; import ru.nbch.credit_tracker.mapper.signal.PackagesRecordsMapper; import ru.nbch.credit_tracker.mapper.signal.PhonesMapMapper; -import ru.nbch.credit_tracker.model.FlagData; -import ru.nbch.credit_tracker.model.IdData; -import ru.nbch.credit_tracker.model.SubjectProfile; -import ru.nbch.credit_tracker.model.UserPackagePhoneData; +import ru.nbch.credit_tracker.model.*; import java.util.*; import java.util.function.Function; @@ -28,13 +28,16 @@ import java.util.stream.Collectors; public class PackageRecordsService { private final PhonesMapMapper phonesMapMapper; + private final LegalsMapMapper legalsMapMapper; private final PackagesRecordsMapper packagesRecordsMapper; private final ObjectMapper entityToJsonMapper; public PackageRecordsService(PhonesMapMapper phonesMapMapper, + LegalsMapMapper legalsMapMapper, PackagesRecordsMapper packagesRecordsMapper, ObjectMapper entityToJsonMapper) { this.phonesMapMapper = phonesMapMapper; + this.legalsMapMapper = legalsMapMapper; this.packagesRecordsMapper = packagesRecordsMapper; this.entityToJsonMapper = entityToJsonMapper; } @@ -148,4 +151,27 @@ public class PackageRecordsService { public List findByPhones(List phones) { return phones.isEmpty() ? Collections.emptyList() : packagesRecordsMapper.findByPhones(phones); } + + public List findCachedLegalsMapByFids(List fids) { + return legalsMapMapper.findByFids(fids); + } + + public List selectUserPackageLegalData(Set> uniqueInnRegnumPairs) { + try { + String json = entityToJsonMapper.writeValueAsString(uniqueInnRegnumPairs.stream() + .filter(Objects::nonNull) + .map(pair -> new HashMap() {{ + put("inn", pair.getKey()); + put("regnum", pair.getValue()); + }}).toList()); + return packagesRecordsMapper.selectUserPackageLegalData(json); + } catch (JsonProcessingException e) { + throw new RuntimeException(e); + } + } + +// public List findByInnAndEgrn(Long phone) { +// return packagesRecordsMapper.findByPhone(phone); +// } + } diff --git a/src/main/resources/mybatis/mapper/signal/LegalsMapMapper.xml b/src/main/resources/mybatis/mapper/signal/LegalsMapMapper.xml new file mode 100644 index 0000000..fdcf6e5 --- /dev/null +++ b/src/main/resources/mybatis/mapper/signal/LegalsMapMapper.xml @@ -0,0 +1,36 @@ + + + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/src/main/resources/mybatis/mapper/signal/PackageRecordsMapper.xml b/src/main/resources/mybatis/mapper/signal/PackageRecordsMapper.xml index 68e135a..4b0d7a4 100644 --- a/src/main/resources/mybatis/mapper/signal/PackageRecordsMapper.xml +++ b/src/main/resources/mybatis/mapper/signal/PackageRecordsMapper.xml @@ -68,6 +68,21 @@ where spr.phone_l in (select * from src) and sup.status = 'ON_MONITORING' + +