From b9973f8cdd55fb5aea803c7bd22c93a730468071 Mon Sep 17 00:00:00 2001 From: Mikhail Trofimov Date: Wed, 24 Jun 2026 11:04:04 +0300 Subject: [PATCH] Added signals_segment_map table; complete data preparing for natural persons --- .../entities/app/SegmentMap.java | 15 +++++++ .../mapper/signal/SegmentMapMapper.java | 13 ++++++ .../monitoring/KafkaMonitoringService.java | 16 +++++++- .../service/signal/SegmentMapService.java | 41 +++++++++++++++++++ .../mapper/signal/SegmentMapMapper.xml | 16 ++++++++ src/main/resources/sql/v2.22.ddl | 20 +++++++++ 6 files changed, 119 insertions(+), 2 deletions(-) create mode 100644 src/main/java/ru/nbch/credit_tracker/entities/app/SegmentMap.java create mode 100644 src/main/java/ru/nbch/credit_tracker/mapper/signal/SegmentMapMapper.java create mode 100644 src/main/java/ru/nbch/credit_tracker/service/signal/SegmentMapService.java create mode 100644 src/main/resources/mybatis/mapper/signal/SegmentMapMapper.xml diff --git a/src/main/java/ru/nbch/credit_tracker/entities/app/SegmentMap.java b/src/main/java/ru/nbch/credit_tracker/entities/app/SegmentMap.java new file mode 100644 index 0000000..aba6b28 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/entities/app/SegmentMap.java @@ -0,0 +1,15 @@ +package ru.nbch.credit_tracker.entities.app; + +import lombok.AllArgsConstructor; +import lombok.Getter; +import lombok.NoArgsConstructor; +import lombok.Setter; + +@Getter +@Setter +@AllArgsConstructor +@NoArgsConstructor +public class SegmentMap { + private String memberCode; + private Integer segment; +} diff --git a/src/main/java/ru/nbch/credit_tracker/mapper/signal/SegmentMapMapper.java b/src/main/java/ru/nbch/credit_tracker/mapper/signal/SegmentMapMapper.java new file mode 100644 index 0000000..5542ec9 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/mapper/signal/SegmentMapMapper.java @@ -0,0 +1,13 @@ +package ru.nbch.credit_tracker.mapper.signal; + +import org.apache.ibatis.annotations.Mapper; +import ru.nbch.credit_tracker.entities.app.SegmentMap; + +import java.util.List; + +@Mapper +public interface SegmentMapMapper { + + List findAllSegmentMaps(); + +} 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 8bb19e6..6a32645 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 @@ -15,6 +15,7 @@ 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.service.signal.SegmentMapService; import ru.nbch.credit_tracker.utils.CTUtil; import java.util.List; @@ -34,16 +35,19 @@ public class KafkaMonitoringService { private final PackageRecordsService packageRecordsService; private final SourceService sourceService; + private final SegmentMapService segmentMapService; private final CreditTrackerSettings config; public KafkaMonitoringService(@Qualifier("signalKafkaEventQueue") BlockingQueue signalEventsQueue, PackageRecordsService packageRecordsService, SourceService sourceService, + SegmentMapService segmentMapService, CreditTrackerSettings config) { this.signalEventsQueue = signalEventsQueue; this.packageRecordsService = packageRecordsService; this.sourceService = sourceService; + this.segmentMapService = segmentMapService; this.config = config; } @@ -73,6 +77,7 @@ public class KafkaMonitoringService { } private void process(List messages) { + Map segmentMap = segmentMapService.loadSegmentMap(); List fids = messages.stream().map(PackageMessage::getFid).collect(Collectors.toList()); List memberCodes = messages.stream() .map(message -> message.getMemberCode().substring(0, 6)) @@ -107,7 +112,7 @@ public class KafkaMonitoringService { continue; } - Integer signalType = CTUtil.checkBusCategory(busCategory); + int signalType = CTUtil.checkBusCategory(busCategory); List phonesMapList = phonesDataMap.get(fid); @@ -131,10 +136,17 @@ public class KafkaMonitoringService { String subjectId = userPackagePhoneData.getSubjectId(); String signalTypes = userPackagePhoneData.getSignalTypes(); - if (signalTypes.contains(signalType.toString())) { + if (signalTypes.contains(Integer.toString(signalType))) { SignalKafkaEvent event = new SignalKafkaEvent(); event.setEventType(EventType.WRITE); event.setUserPackagePhoneData(userPackagePhoneData); + event.setSubjectId(subjectId); + event.setFlagDebt5000(flagDebt5000); + event.setFlag90days(flag90days); + event.setSignalType(Integer.toString(signalType)); + event.setBusCategory(signalType == 3 ? busCategory : null); + event.setSegment(segmentMap.get(memberCode.substring(0, 6))); + putToQueue(event); } } diff --git a/src/main/java/ru/nbch/credit_tracker/service/signal/SegmentMapService.java b/src/main/java/ru/nbch/credit_tracker/service/signal/SegmentMapService.java new file mode 100644 index 0000000..e2f7181 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/signal/SegmentMapService.java @@ -0,0 +1,41 @@ +package ru.nbch.credit_tracker.service.signal; + +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Service; +import ru.nbch.credit_tracker.entities.app.SegmentMap; +import ru.nbch.credit_tracker.mapper.signal.SegmentMapMapper; + +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; + +@Slf4j +@Service +public class SegmentMapService { + + private final SegmentMapMapper mapper; + + public SegmentMapService(SegmentMapMapper mapper) { + this.mapper = mapper; + } + + public Map loadSegmentMap() { + // TODO guava cache + List segmentMaps = mapper.findAllSegmentMaps(); + if (segmentMaps.isEmpty()) { + log.warn("No segmentMaps found"); + return Collections.emptyMap(); + } + return segmentMaps.stream() + .flatMap(item -> Arrays.stream(item.getMemberCode().split(",\\s*")) + .map(code -> Map.entry(code.trim(), item.getSegment())) + ) + .collect(Collectors.toMap( + Map.Entry::getKey, + Map.Entry::getValue, + (existing, replacement) -> existing + )); + } +} diff --git a/src/main/resources/mybatis/mapper/signal/SegmentMapMapper.xml b/src/main/resources/mybatis/mapper/signal/SegmentMapMapper.xml new file mode 100644 index 0000000..e30fb95 --- /dev/null +++ b/src/main/resources/mybatis/mapper/signal/SegmentMapMapper.xml @@ -0,0 +1,16 @@ + + + + + + + + + + + \ No newline at end of file diff --git a/src/main/resources/sql/v2.22.ddl b/src/main/resources/sql/v2.22.ddl index 33ed8b7..a0c5b33 100644 --- a/src/main/resources/sql/v2.22.ddl +++ b/src/main/resources/sql/v2.22.ddl @@ -1,3 +1,6 @@ +alter table signals_user_packages + add subject_type smallint; + create index signals_phone_fid_map_phone_l_index on signals_phone_fid_map (phone_l); @@ -16,6 +19,9 @@ CREATE TABLE signals_legal_fid_map reg_num bigint ); +alter table signals_legal_fid_map + owner to credit_tracker_user; + -- будем искать данные по связке inn + reg_num create index signals_legal_fid_map_inn_reg_num_index on signals_legal_fid_map (inn, reg_num); @@ -32,12 +38,26 @@ CREATE TABLE signals_legals_map last_rejected_date timestamp ); +alter table signals_legals_map + owner to credit_tracker_user; + create index signals_legals_map_fid_index on signals_legals_map (fid); ALTER TABLE signals_legals_map ADD CONSTRAINT signals_legals_map_inn_reg_num_key UNIQUE (inn, reg_num); +CREATE TABLE signals_segment_map +( + memberСode text, + segment smallint +); + +alter table signals_segment_map + owner to credit_tracker_user; + + +-- TODO new fields for hits UPDATE app_version SET db_version = '2.22' WHERE db_version = '2.21';