Added processing for legal persons

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

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<LegalsMap> findByFids(@Param("fids") List<Integer> fids);
+
+}

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<UserPackagePhoneData> selectUserPackagePhoneData(@Param("json") String json);
 
+    List<UserPackageLegalData> selectUserPackageLegalData(@Param("json") String json);
+
     Long deleteByPackageId(@Param("packageId") Long id);
 
     PackagesRecordsMap findById(Long id);

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

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

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<PackageMessage> messages) {
+        List<Hit> hits = new ArrayList<>();
+
+        Map<Boolean, List<PackageMessage>> partition = messages.stream()
+                .filter(this::filterMessages)
+                .collect(Collectors.partitioningBy(m -> m.getPersons() != null && !m.getPersons().isEmpty()));
+
+        List<PackageMessage> npMessages = partition.get(true);
+        List<PackageMessage> lpMessages = partition.get(false);
+
         Map<String, Integer> segmentMap = segmentMapService.loadSegmentMap();
+
+        if (npMessages != null && !npMessages.isEmpty()) {
+            List<Hit> npHits = processNpMessages(npMessages, segmentMap);
+            hits.addAll(npHits);
+        }
+
+        if (lpMessages != null && !lpMessages.isEmpty()) {
+            List<Hit> lpHits = processLpMessages(lpMessages, segmentMap);
+            hits.addAll(lpHits);
+        }
+        // TODO insert hits
+    }
+
+    private List<Hit> processNpMessages(List<PackageMessage> messages, Map<String, Integer> segmentMap) {
+        List<Hit> hits = new ArrayList<>();
+
         List<Integer> fids = messages.stream().map(PackageMessage::getFid).collect(Collectors.toList());
         List<String> 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<Hit> processLpMessages(List<PackageMessage> messages, Map<String, Integer> segmentMap) {
+        List<Hit> hits = new ArrayList<>();
+
+        List<Integer> fids = messages.stream().map(PackageMessage::getFid).collect(Collectors.toList());
+        List<String> memberCodes = messages.stream()
+                .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));
+
+        Set<Pair<Long, Long>> uniqueInnRegnumPairs = legalsMaps.stream()
+                .map(l -> Pair.of(l.getInn(), l.getRegNUm()))
+                .collect(Collectors.toSet());
+
+        // прелоад по memberCode категорий
+        Map<String, String> busCategories = sourceService.findBusCategories(memberCodes);
+
+        // subjectId, packageId, signalTypes, inn, regNum только для статуса ON_MONITORING
+        List<UserPackageLegalData> packageLegalData = packageRecordsService
+                .selectUserPackageLegalData(uniqueInnRegnumPairs);
+
+        Map<Pair<Long, Long>, List<UserPackageLegalData>> 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<LegalsMap> legalsMapList = legalsDataMap.get(fid);
+
+            if (legalsMapList == null) {
+                continue;
+            }
+
+            for (LegalsMap legalsMap : legalsMapList) {
+                // флаги просрочек
+                Integer flag90days = legalsMap.getFlag90days();
+
+                // в userPackageLegalDataMap только нужные inn, regnum, те что не вошли в userPackageLegalDataList игнорируются
+                List<UserPackageLegalData> 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) {

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.*;
-import ru.nbch.credit_tracker.model.IdData;
-import ru.nbch.credit_tracker.model.SubjectProfile;
-import ru.nbch.credit_tracker.model.UserPackagePhoneData;
 
 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<PackagesRecordsMap> findByPhones(List<Long> phones) {
         return phones.isEmpty() ? Collections.emptyList() : packagesRecordsMapper.findByPhones(phones);
     }
+
+    public List<LegalsMap> findCachedLegalsMapByFids(List<Integer> fids) {
+        return legalsMapMapper.findByFids(fids);
+    }
+
+    public List<UserPackageLegalData> selectUserPackageLegalData(Set<Pair<Long, Long>> uniqueInnRegnumPairs) {
+        try {
+            String json = entityToJsonMapper.writeValueAsString(uniqueInnRegnumPairs.stream()
+                    .filter(Objects::nonNull)
+                    .map(pair -> new HashMap<String, Object>() {{
+                        put("inn", pair.getKey());
+                        put("regnum", pair.getValue());
+                    }}).toList());
+            return packagesRecordsMapper.selectUserPackageLegalData(json);
+        } catch (JsonProcessingException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+//    public List<PackagesRecordsMap> findByInnAndEgrn(Long phone) {
+//        return packagesRecordsMapper.findByPhone(phone);
+//    }
+
 }

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

@@ -0,0 +1,36 @@
+<?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.LegalsMapMapper">
+
+    <resultMap id="LegalsMap" type="ru.nbch.credit_tracker.entities.app.LegalsMap">
+        <result property="id" column="id"/>
+        <result property="inn" column="inn"/>
+        <result property="regNUm" column="reg_num"/>
+        <result property="fid" column="fid"/>
+        <result property="searchedAt" column="searched_at"/>
+        <result property="flag90days" column="flag_90_12"/>
+        <result property="calculatedAt" column="calculated_at"/>
+        <result property="lastRejectedDate" column="last_rejected_date"/>
+    </resultMap>
+
+    <select id="findByFids" resultMap="LegalsMap">
+        select id,
+               inn,
+               reg_num,
+               fid,
+               searched_at,
+               flag_90_12,
+               calculated_at,
+               last_rejected_date
+        FROM signals_legals_map
+        WHERE fid in
+        <foreach item='fid' collection='fids' open='(' separator=',' close=')'>
+            #{fid}
+        </foreach>
+    </select>
+
+
+</mapper>

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'
     </select>
 
+    <select id="selectUserPackageLegalData" resultType="ru.nbch.credit_tracker.model.UserPackageLegalData">
+        WITH src AS (SELECT inn, regNum
+                     FROM json_to_recordset(#{json}::json) AS t(
+                                                                inn bigint,
+                                                                regNum bigint
+                         ))
+        SELECT sup.id AS packageId, sup.signal_types AS signalTypes, sup.membercode AS userId,
+               sup.package_name AS packageName, slfm.inn AS inn, slfm.reg_num AS regNum,
+               slfm.subject_id AS subjectId
+        FROM signals_legal_fid_map slfm
+                 JOIN signals_user_packages sup ON sup.id = slfm.package_id
+        WHERE (slfm.inn, slfm.reg_num) IN (SELECT inn, regNum FROM src)
+          AND sup.status = 'ON_MONITORING'
+    </select>
+
     <select id="registerJson" parameterType="string" resultType="long">
         WITH src AS (SELECT row_number() OVER () AS rn,
                             subject_id,