Added signals_segment_map table; complete data preparing for natural persons

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;
+}

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<SegmentMap> findAllSegmentMaps();
+
+}

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<SignalKafkaEvent> 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<PackageMessage> messages) {
+        Map<String, Integer> segmentMap = segmentMapService.loadSegmentMap();
         List<Integer> fids = messages.stream().map(PackageMessage::getFid).collect(Collectors.toList());
         List<String> 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<PhonesMap> 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);
                     }
                 }

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<String, Integer> loadSegmentMap() {
+        // TODO guava cache
+        List<SegmentMap> 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
+                ));
+    }
+}

src/main/resources/mybatis/mapper/signal/SegmentMapMapper.xml

@@ -0,0 +1,16 @@
+<?xml version="1.0" encoding="UTF-8" ?>
+<!DOCTYPE mapper
+        PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
+        "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
+
+<mapper namespace="ru.nbch.credit_tracker.mapper.signal.SegmentMapMapper">
+
+    <resultMap id="appVersionResultMap" type="ru.nbch.credit_tracker.entities.app.AppVersion">
+        <result property="dbVersion" column="db_version"/>
+    </resultMap>
+
+    <select id="findAllSegmentMaps" resultType="ru.nbch.credit_tracker.entities.app.SegmentMap">
+        SELECT memberCode, segment
+        FROM signals_segment_map
+    </select>
+</mapper>

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';